Hadoop StreamingでMapReduceをシェルで実装する


## はじめに

Hadoop StreamingでMapReduceをシェルで実装する方法を記述します。

## 環境

* CentOS 6.5
* CDH5

## 構成

|ホスト名      |IPアドレス     |ResourceManager|Namenode|NodeManager|Datanode|JobHistoryServer|
|:-------------|:--------------|:-------------:|:------:|:---------:|:------:|:--------------:|
|hadoop-master |192.168.122.101|○             |○      |-          |-       |○              |
|hadoop-master2|192.168.122.102|○             |○      |-          |-       |-               |
|hadoop-slave  |192.168.122.111|-              |-       |○         |○      |-               |
|hadoop-slave2 |192.168.122.112|-              |-       |○         |○      |-               |
|hadoop-slave3 |192.168.122.113|-              |-       |○         |○      |-               |
|hadoop-client |192.168.122.201|-              |-       |-          |-       |-               |

クラスタの構築方法は、[CDH5でhadoopのクラスタを構築する](http://qiita.com/toshiro3/items/6e010b837246598a92ee)をご参照ください。

## MapReduceの実装

今回は、ファイルに含まれる単語の数をカウントするプログラムを作成します。

* mapperの実装

```bash:sample_mapper.sh
#!/bin/bash

while read -a words
do
  for word in ${words[*]}
  do
    printf "%s\t1\n" $word
  done
done
```

* reducerの実装

```bash:sample_reducer.sh
#!/bin/bash

current_key=""
total=0

while read data
do
  key=$(echo $data | awk '{print $1}')
  value=$(echo $data | awk '{print $2}')
  if [ "$current_key" != "" ] && [ "$current_key" != "$key" ] ; then
    printf '%s\t%d\n' "$current_key" "$total"
    current_key=$key
    total=$value
  else
    current_key=$key
    total=$((total+value))
  fi
done
printf '%s\t%d\n' "$current_key" "$total"
```

## 動作確認

クライアントで動作確認をします。

* 動作確認用に解析対象のファイルをHDFSへputします。

```txt:sample.txt
Java Ruby Python Java Ruby Python Java
```

```shell
$ sudo -u hdfs hadoop fs -put sample.txt input/sample.txt
$ sudo -u hdfs hadoop fs -ls input/
Found 1 items
-rw-r--r--   3 hdfs hadoop         39 2014-09-20 06:59 input/sample.txt
```
* Hadoop Streamingを使用してMapReduceを実行します。

```shell
$ sudo -u hdfs hadoop                                   \
  jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar    \
  -files   /tmp/sample_mapper.sh,/tmp/sample_reducer.sh \
  -input   input/sample.txt                             \
  -output  output/sample                                \
  -mapper  sample_mapper.sh                             \
  -reducer sample_reducer.sh
```

Hadoop Streamingを使用する場合は、jarに __hadoop-streaming.jar__ を指定します。

その他のオプションは以下の通りとなります。

|オプション|概要                                                |
|:---------|:---------------------------------------------------|
|files     |Map/Reduceのクラスタへコピーするファイルを指定します|
|input     |解析の対象となるファイルを指定します                |
|output    |解析結果を保存するHDFS上のパスを指定します          |
|mapper    |map処理を実装したファイル名を指定します             |
|reducer   |reduce処理を実装したファイル名を指定します          |

filesオプションで指定したファイルは、各ノードへコピーされ、そのコピーへのシンボリックリングが各タスクの作業用ディレクトリに作成されます。
上記の場合は、filesオプションで、 __/tmp/sample_mapper.sh__ __/tmp/sample_reducer.sh__を指定しているため、２つのファイルが各ノードへコピーされます。
また、各タスクの作業用ディレクトリに __sample_mapper.sh__ __sample_reducer.sh__という名前のシンボリックリンクが作成されます。
そのため、mapperオプションへ、 __sample_mapper.sh__ reducerオプションへ、 __sample_reducer.sh__を指定すればよいことになります。

* 解析結果を確認します。

```shell
$ sudo -u hdfs hadoop fs -cat 'output/sample/part-*'
Java    3
Python  2
Ruby    2
```

## 参考

* [Hadoop Streaming](http://hadoop.apache.org/docs/r2.5.1/hadoop-mapreduce-client/hadoop-mapreduce-client-core/HadoopStreaming.html)
* [Tips and Guidelines](https://www.cloudera.com/content/cloudera-content/cloudera-docs/CDH5/5.0/CDH5-Installation-Guide/cdh5ig_tips_guidelines.html)