[Hadoop] Hadoop Streaming / mapperとreducerにPythonを使ってみる。

HadoopはJavaで作られている。
だからHadoopに何か操作をさせたい場合には、通常、Javaで記述する必要がある。
しかしHadoopにはHadoop Streamingという仕組みがあり、早い話UNIXのStandard Stream要するに標準入出力を扱うことができる。

すなわち、UNIXの標準入出力の流儀に則ってさえいれば、お好きな言語で操作ができる。
Javaがまったく合わない私としては、Hadoop Streamはとてもありがたい。
これがなければHadoopに手を付ける気にはならなかった。

Hadoop Streamに必要なもの。

mapperとreducerを、好きな言語で書くだけ。
べつにmapperだけでもよいけど。
私はPython。

Hadoop streamにおけるmapperとreducerの概要。

mapperは何らかの入力を得て、キーと値（key, value）を出力する。
reducerはmapperからのkey, valueを受けて、keyごとにvalueを処理する。

なお、mapperの出力がreducerに渡されるとき、Hadoopがkeyごとにソートしてくれる。
この点はreducerの処理を簡単にする。詳細は後述。

試しにやってみること。

LAN向けのApacheのアクセス状況をカウントしてみる。
IPアドレスごとのアクセス回数だ。
ログは以下のようなもの。
アクセス元はすべてIPアドレスで記録されている。

192.168.100.106 - - [24/Jul/2013:22:35:26 +0900] "GET /MT/js/common/Editor/Iframe.js?v=5.14-ja HTTP/1.1" 304 - "http://192.168.100.5/MTcgi/mt.cgi?__mode=view&_type=entry&blog_id=2&id=330" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:22.0) Gecko/20100101 Firefox/22.0"
192.168.100.106 - - [24/Jul/2013:22:35:26 +0900] "GET /MT/js/common/Editor/Textarea.js?v=5.14-ja HTTP/1.1" 304 - "http://192.168.100.5/MTcgi/mt.cgi?__mode=view&_type=entry&blog_id=2&id=330" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:22.0) Gecko/20100101 Firefox/22.0"
192.168.100.106 - - [24/Jul/2013:22:35:26 +0900] "GET /MT/mt.js?v=5.14-ja HTTP/1.1" 304 - "http://192.168.100.5/MTcgi/mt.cgi?__mode=view&_type=entry&blog_id=2&id=330" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:22.0) Gecko/20100101 Firefox/22.0"

処理の流れ

1.mapperは、アクセスログからIPアドレスを抜き出す。
そして「＜IPアドレス＞＜タブ＞1」を出力する。
これは、たとえば「192.168.1.1」から「1」回アクセスがあったよ、という意味。

2.hadoopがIPアドレスをキーにソート。

3.reducerは、IPアドレスごとに回数をカウントし、
「＜IPアドレス＞＜タブ＞＜集計回数＞」を出力する。

mapper.py

IPアドレスは、Apacheログにおいて、スペースを区切りにした第一フィールドに記載される。
だから一行ずつログを読んで、行頭のIPアドレスを抜き出し、その都度「＜IPアドレス＞＜タブ＞1」を出力する。
「if “newsyslog” not in line:」は、システムメッセージ行を読み飛ばすため。

#!/usr/bin/env python

import sys

for line in sys.stdin:
        if "newsyslog" not in line:
                fields = line.strip().split()
                print '%s\t%s' % (fields[0],1)

実行権限も忘れずにつける。

$ chmod a+x ./mapper.py

実験。意図したとおり動いていますね。

$ cat ../input/httpd-access.log |./mapper.py 

192.168.100.107 1
192.168.100.106 1
192.168.100.110 1
192.168.100.107 1
192.168.100.107 1
192.168.100.107 1

reducer.py

mapper.pyからの出力は、ソートされてreducer.pyに入力される。
上記のmapper.py出力例は、以下のようにソートされる。

192.168.100.106 1
192.168.100.107 1
192.168.100.107 1
192.168.100.107 1
192.168.100.107 1
192.168.100.110 1

だからreducerとしては、第一フィールドを上から読んでいって、keyが変化したら、そこまでのカウント数を出力する。
そしてそのkeyの事は、さっぱり忘れて次のkeyのカウントに移ることができる。
もしソートがなされていないならば、入力の終わりまですべてのkeyを保持しなければならない。Hadoopに感謝である。

#!/usr/bin/env python

import sys

(last_key, count) = (None, 0)

for line in sys.stdin:
        (key, val) = line.strip().split("\t")

        if last_key and last_key != key:
                print "%s\t%s" % (last_key, count)
                (last_key, count) = (key,int(val))
        else:
                last_key = key
                count += int(val)

if last_key:
        print "%s\t%s" % (last_key, count)

実行権限を付ける。

$ chmod a+x reducer.py

実験。間にsortを入れること。
問題なし。

$ cat ../input/httpd-access.log |./mapper.py |sort|./reducer.py 

127.0.0.1       567
192.168.100.106 327
192.168.100.107 671
192.168.100.109 2
192.168.100.150 193

Hadoopで動かしてみよう。

まずカウント対象となるログをHDFSにコピーする。

[hadoop@isis ~]$ hadoop dfs -put apachelog apachelog
[hadoop@isis ~]$
[hadoop@isis ~]$ hadoop dfs -ls
Found 1 items
drwxr-xr-x   - hadoop supergroup          0 2013-07-28 12:57 /user/hadoop/apachelog
[hadoop@isis ~]$
[hadoop@isis ~]$ hadoop dfs -ls apachelog
Found 1 items
-rw-r--r--   1 hadoop supergroup     999290 2013-07-28 12:57 /user/hadoop/apachelog/httpd-access.log
[hadoop@isis ~]$

実行。

hadoopにhadoop-streaming-1.0.0.jarを与え、input、outputのほかに、mapperとreducerも指定する。
-mapper -reducerとしてローカルファイルシステムでのパスを与える。
同時に、-fileでそれぞれのスクリプトを指定すると、スクリプトファイルをリモートのノードへ送ってくれる。

コマンドはすごく長くなる。
エスケープシーケンスを使って適宜改行し、見やすくしてタイプミスを防ぐ。

hadoop  jar \
/usr/local/share/hadoop/contrib/streaming/hadoop-streaming-1.0.0.jar \
    -input apachelog \
    -output apachelog.out \
    -mapper /home/hadoop/sandbox/mapper/mapper.py \
    -reducer /home/hadoop/sandbox/mapper/reducer.py \
    -file /home/hadoop/sandbox/mapper/mapper.py \
    -file /home/hadoop/sandbox/mapper/reducer.py

実際のログ

[hadoop@isis ~]$ hadoop  jar \
>    -output apachelog.out \
>    -mapper /home/hadoop/sandbox/mapper/mapper.py \
> /usr/local/share/hadoop/contrib/streaming/hadoop-streaming-1.0.0.jar \
>    -file /home/hadoop/sandbox/mapper/mapper.py \
>    -file /home/hadoop/sandbox/mapper/reducer.py \
>     -input apachelog \
>     -output apachelog.out \
>     -mapper /home/hadoop/sandbox/mapper/mapper.py \
>     -reducer /home/hadoop/sandbox/mapper/reducer.py \
>     -file /home/hadoop/sandbox/mapper/mapper.py \
>     -file /home/hadoop/sandbox/mapper/reducer.py

packageJobJar: [/home/hadoop/sandbox/mapper/mapper.py, /home/hadoop/sandbox/mapper/reducer.py, /tmp/hadoop-hadoop/hadoop-unjar1190814954490199586/] [] /tmp/streamjob8960127756596123730.jar tmpDir=null
13/07/28 13:30:15 INFO mapred.FileInputFormat: Total input paths to process : 1
13/07/28 13:30:43 INFO streaming.StreamJob: getLocalDirs(): [/tmp/hadoop-hadoop/mapred/local]
13/07/28 13:30:43 INFO streaming.StreamJob: Running job: job_201307281252_0001
13/07/28 13:30:43 INFO streaming.StreamJob: To kill this job, run:
13/07/28 13:30:43 INFO streaming.StreamJob: /usr/local/share/hadoop/bin/../bin/hadoop job  -Dmapred.job.tracker=localhost:8021 -kill job_201307281252_0001
13/07/28 13:30:43 INFO streaming.StreamJob: Tracking URL: http://localhost:50030/jobdetails.jsp?jobid=job_201307281252_0001
13/07/28 13:30:46 INFO streaming.StreamJob:  map 0%  reduce 0%
13/07/28 13:39:14 INFO streaming.StreamJob:  map 100%  reduce 0%
13/07/28 13:40:27 INFO streaming.StreamJob:  map 100%  reduce 33%
13/07/28 13:40:33 INFO streaming.StreamJob:  map 100%  reduce 67%
13/07/28 13:40:49 INFO streaming.StreamJob:  map 100%  reduce 100%
13/07/28 13:42:39 INFO streaming.StreamJob: Job complete: job_201307281252_0001
13/07/28 13:42:41 INFO streaming.StreamJob: Output: apachelog.out

結果の確認

hadoop dfs -catなどで。

[hadoop@isis ~]$ hadoop dfs -ls
Found 2 items
drwxr-xr-x   - hadoop supergroup          0 2013-07-28 12:57 /user/hadoop/apachelog
drwxr-xr-x   - hadoop supergroup          0 2013-07-28 13:42 /user/hadoop/apachelog.out
[hadoop@isis ~]$
[hadoop@isis ~]$ hadoop dfs -ls apachelog.out

Found 3 items
-rw-r--r--   1 hadoop supergroup          0 2013-07-28 13:42 /user/hadoop/apachelog.out/_SUCCESS
drwxr-xr-x   - hadoop supergroup          0 2013-07-28 13:30 /user/hadoop/apachelog.out/_logs
-rw-r--r--   1 hadoop supergroup       1438 2013-07-28 13:40 /user/hadoop/apachelog.out/part-00000
[hadoop@isis ~]$
[hadoop@isis ~]$ hadoop dfs -cat apachelog.out/part-00000

127.0.0.1       567
192.168.100.106 327
192.168.100.107 671
192.168.100.109 2
192.168.100.150 193

[hadoop@isis ~]$

以上

Hadoop Streamに必要なもの。#

Hadoop streamにおけるmapperとreducerの概要。#

試しにやってみること。#

処理の流れ#

mapper.py#

reducer.py#

Hadoopで動かしてみよう。#

結果の確認#