HiveでWordCount

HiveでWordCountを作ってみる。

単語数を数える処理

Reduce処理に当たる、単語数を数える部分を先に考える。
というのは、Hiveではcount関数が使えるので、単語数を数えるのは簡単だから。

（Windows上のHiveで試しているので、ファイルやディレクトリーはWindowsのパス表記）
（実行にはCygwinを使用）

テキストファイルに単語データ（Map処理で分割されたものを想定）を用意し、テーブルに読み込む。

C:/cygwin/tmp/words.txt（入力データ）：

Hello
World
Hello
Hive

hiveのコマンド：

--テーブル作成
create table words(
  word string
);

--データをロード
load data local inpath'C:/cygwin/tmp/words.txt'
overwrite into table words;

hive> select * from words;
OK
Hello
World
Hello
Hive

hive> select word, count(*) from words
    > group by word;
〜
OK
Hello   2
Hive    1
World   1

スクリプトを使った単語分割処理

文字列を単語に分割して複数行のデータを作るには、まずはスクリプトを使う方法が思い浮かぶ。
特にawkは空白文字をデリミター（区切り文字）とした単語分割を自動的に行ってくれるので、これを使うと簡単。

C:/cygwin/tmp/hello.txt（入力データ）：

Hello World
Hello Hive

C:/cygwin/tmp/split.awk：

{ for(i = 1; i <= NF; i++) print $i }

Cygwinから試しに実行。

$ cd /tmp
$ awk -f split.awk < hello.txt
Hello
World
Hello
Hive

OK。

次にテーブルを作成してデータを読み込んでおく。

--テーブル作成
create table hello(
  text string
);

--データをロード
load data local inpath'C:/cygwin/tmp/hello.txt'
overwrite into table hello;

スクリプトを実行するには、transform〜using文を使う。

select transform(入力項目名) using'スクリプト' as 出力項目名 from 入力テーブル;
select transform(入力項目名) using'スクリプト' as (出力項目名 型) from 入力テーブル;

• 入力テーブルの指定された項目をスクリプトの標準入力へ渡し、その標準出力への出力データを出力項目名で受け取る。
• asの後ろの項目名を括弧でくくって型（stringとか）を指定することも出来る。

さっそくやってみよう。

hive> select transform(text) using'awk -f /tmp/split.awk' as word from hello;
Total MapReduce jobs = 1
Launching Job 1 out of 1
Number of reduce tasks is set to 0 since there's no reduce operator
Execution log at: /tmp/hishidama/hishidama_20110809213636_86c15a1f-22df-46ee-9821-24dbf94e3c5f.log
Job running in-process (local Hadoop)
2011-08-09 21:36:44,296 null map = 0%, reduce = 0%
Ended Job = job_local_0001 with errors
FAILED: Execution Error, return code 2 from org.apache.hadoop.hive.ql.exec.MapRedTask

ははは、エラーが出たよ(苦笑)
実行ログはここで表示されている通り、「C:\tmp/ユーザー/〜」に出ているので、それを見てみる。

java.lang.RuntimeException: org.apache.hadoop.hive.ql.metadata.HiveException: Hive Runtime Error while processing row {"text":"Hello World"}
	at org.apache.hadoop.hive.ql.exec.ExecMapper.map(ExecMapper.java:161)
〜
Caused by: java.io.IOException: CreateProcess error=5, ?A?N?Z?X
	at java.lang.ProcessImpl.create(Native Method)
	at java.lang.ProcessImpl.<init>(ProcessImpl.java:81)
	at java.lang.ProcessImpl.start(ProcessImpl.java:30)
	at java.lang.ProcessBuilder.start(ProcessBuilder.java:453)

よくあるProcessBuilderでのエラー。ということは、awkが実行できない可能性大。
awkコマンドは実際どうなっているかと言うと、

$ which awk
/usr/bin/awk

$ ls -l /usr/bin/awk
lrwxrwxrwx 1 hishidama Administrators 8 Aug  8 00:21 /usr/bin/awk -> gawk.exe

gawk.exeへのリンクだった。
exeファイルならProcessBuilderで実行できるので、gawkに変えてみる。

hive> select transform(text) using'gawk -f /tmp/split.awk' as word from hello;
〜
OK
Hello
World
Hello
Hive

出来た！
（なお、gawk.exeはCygwin用に作られた実行ファイルなので、引数のパスがUNIX形式（/tmp/split.awk）でも正しく解釈してくれる）

ちなみにawkの場合、命令をファイルにしなくても、引数で書くことが出来る。
（UNIXのシェルが$を環境変数の開始として解釈するのを防ぐ為に、シングルクォーテーションでくくる）

$ awk '{ for(i = 1; i <= NF; i++) print $i }' < /tmp/hello.txt
Hello
World
Hello
Hive

これをHiveから実行してやると…またエラーになった。

試行錯誤の末、上手くいったのは以下の様な書き方だった。

hive> select transform(text) using'gawk "{ for(i = 1\; i <= NF\; i++) print $i }"' as word from hello;
〜
OK
Hello
World
Hello
Hive

セミコロン「;」の前に「\」を付けてエスケープしている。
つまり、HiveQLはセミコロンが来たら文の終わりと認識しているのだろう。

最終的に、countによる単語数カウントと組み合わせて以下の様になる。

select word, count(*) from (
  select transform(text) using'gawk "{ for(i = 1\; i <= NF\; i++) print $i }"' as word from hello
) a
group by word;

※Hiveでは、副問合せに必ず別名（上記のa）を付ける必要がある。（その別名を使わなくても）

Hiveの関数を使った単語分割処理

awkを使って単語分割をすることは出来たが、やはりHiveの機能だけを使って何とかしたい。

Hiveの組込関数を見ていたら、splitというものを見つけた。

hive> desc function extended split;
OK
split(str, regex) - Splits str around occurances that match regex
Example:
  > SELECT split('oneAtwoBthreeC', '[ABC]') FROM src LIMIT 1;
  ["one", "two", "three"]

正規表現で指定された文字によって分割するものらしい。これを使ってみると…

hive> select split(text, '[ \t]+') from hello;
〜
OK
["Hello","World"]
["Hello","Hive"]

なんか各括弧で囲まれているが、分割されているようだ。
これは一体何なのか？ を調べてみる。（ダミーのテーブルを作って、属性を見る（属性を確認するならexplainの方が高速[2011-08-12]））

hive> create table dummy1 as select split(text, '[ \t]+') from hello;
〜
OK

hive> desc dummy1;
OK
_c0 array<string>

なんか、Stringの配列らしい。
そして、配列は、explodeという関数によって行データに変換することが出来るらしい。→LanguageManual LateralView
（explodeを使うときは、後ろに項目の別名を指定しないといけない模様）

hive> select explode(split(text, '[ \t]+')) word from hello;
〜
OK
Hello
World
Hello
Hive

まさに望んだ通りの動作！

最終的に単語数を求める方法は以下のようなものが考えられる。

create table words(word string);

insert overwrite table words
select explode(split(text, '[ \t]+')) word from hello;

select word, count(*) from words
group by word;

create table words as
select explode(split(text, '[ \t]+')) word from hello;

select word, count(*) from words
group by word;

select s.word, count(*) from (
 select explode(split(text, '[ \t]+')) word from hello
) s
group by s.word;

hive> select word, count(*) from (
    >  select explode(split(text, '[ \t]+')) word from hello
    > ) s
    > group by word;
Total MapReduce jobs = 1
Launching Job 1 out of 1
Number of reduce tasks not specified. Estimated from input data size: 1
In order to change the average load for a reducer (in bytes):
  set hive.exec.reducers.bytes.per.reducer=<number>
In order to limit the maximum number of reducers:
  set hive.exec.reducers.max=<number>
In order to set a constant number of reducers:
  set mapred.reduce.tasks=<number>
Execution log at: /tmp/hishidama/hishidama_20110809231212_ca30a369-9f5f-4473-8574-beac806a4577.log
Job running in-process (local Hadoop)
2011-08-09 23:13:50,015 null map = 0%, reduce = 0%
2011-08-09 23:13:52,015 null map = 100%, reduce = 100%
Ended Job = job_local_0001
OK
Hello   2
Hive    1
World   1