MacにHadoop入れてPHPでMapReduce書いてみる


Hadoop！Hadoop!と皆さん大騒ぎしますが、そんな大げさなものではありません。
Hadoopは本当にシンプル（かつ強力なソリューション）です。

試すだけなら、

環境構築 5分。
MapReduce理解（実装含む）30分

で試せます。

なので、ここではMac上にHadoop環境を作る方法をメモします。

##やりたいこと

* Hadoopの説明用にローカルMacにHadoop入れる
* 簡単なMapReduce処理を書いてみる
* Javaは私には高度すぎるのでPHPで書いてみる

##環境

* MacはEl Capitan(10.11.1)
* HomebrewでHadoopインストールします(2.7.1が入る）
* PHPはMacに最初から入ってるやつ(5.5.29)
* Hadoopはとりあえずスタンドアロンモードで試してみる

##インストール

###前提

インストールにはhomebrewを利用するので、未インストールならインストールして下さい。あと、JDKも。

私の手元のjavaは、

```bash
java -version

java version "1.8.0_05"
Java(TM) SE Runtime Environment (build 1.8.0_05-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.5-b02, mixed mode)
```
という状態。1.6.x以上ならいいという噂をよく目にしますね。

###インストール

brewを利用すればものすごく簡単です。

```bash
brew install hadoop
```

以上。2015年11月12日現在、2.7.1がインストールされるみたいです。
インストール場所は/usr/local/Cellar/hadoop/以下。

>hadoopの各種ファイルの場所はバージョン名に依存するため、確認しておきましょう。

##動作確認

###バージョン

とりあえずスタンドアロンモードでちゃんと動いているか確認してみます。

```bash
hadoop version

Hadoop 2.7.1
Subversion https://git-wip-us.apache.org/repos/asf/hadoop.git -r 15ecc87ccf4a0228f35af08fc56de536e6ce657a
Compiled by jenkins on 2015-06-29T06:04Z
Compiled with protoc 2.5.0
From source with checksum fc0a1a23fc1868e4d5ee7fa2b28a58a
This command was run using /usr/local/Cellar/hadoop/2.7.1/libexec/share/hadoop/common/hadoop-common-2.7.1.jar
```

バージョンが表示されればOKです。

###WordCount

次にHadoop界のHelloWorld。WordCountを実行してみます。ファイル中に含まれる文字をカウントするというもの。

サンプルを実行するために、といあえず、入力用のファイルを作る。

```bash
mkdir input
echo "a a a b b c" > input/test1.txt
echo "a b c" > input/test2.txt
```

inputとディレクトリの中に、せっかくなのでファイルを２つ作成。
a=4,b=3,c=2という感じになることを期待。
スペースで区切った文字をいくつか書いておけばOK。

では、実行。

```bash
hadoop jar /usr/local/opt/hadoop/libexec/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.1.jar wordcount input output
```
outputというフォルダがでて、その中にpart-r-00000というファイルができるので中身を見てみます。

```bash
cat part-r-00000
a	4
b	3
c	2
```
期待通りの値が得られています。

後は目的に応じたMapReduce処理を書いて、AWSでもAzureでもクラスタに投入するだけ。

##MapReduce(Streaming)

後はMapReduce処理を書くだけ！と言ってはみたものの、処理をJavaで書くのはしんどいので、ここではHadoop StreamingというJava以外でMapReduceを書ける仕組みを利用してPHPで処理を描いてみることにします。

処理はサンプルと同じくWordCountとします。

###HadoopStreaming

HadoopStreamingは、下記のような書式で、

```bash
$ hadoop jar /usr/local/Cellar/hadoop/2.7.1/libexec/share/hadoop/tools/lib/hadoop-streaming-2.7.1.jar
 -input input
 -output output
 -mapper map.php
 -reducer reduce.php
```

>実際は1行です（改行部スペース）

IN（入力ファイルのあるディレクトリ）,OUT（出力ディレクトリ）とMap処理、Reduce処理に利用するスクリプトを指定することで、MapReduce処理を実現するシンプルかつ強力な仕組みのことです。

>分散環境では、さらに-files optionを指定し、各マシンに配布するスクリプファイルを指定します。

Mapへの入力、Reducerへの入力処理はSTDINを利用して行われるので、目的に応じてINPUTに対する処理をそれぞれ記述します。

STDIN？？となるかもしれませんが、

```bash
hoge.php < test.txt
```

的な処理を行った時と同じ形式ということです。

###処理の流れ

実際に何を実装するか、整理しておきましょう。

####WordCount処理とは？

サンプルでも見ましたが、"a b c a"という内容のファイルを読み込み、それぞれの単語が何文字含まれるか？をカウントする処理です。ここで期待する結果は、

a 2
b 1
c 1

という感じです。これくらいの処理ならHadoopを利用する必要はありませんが、対象のファイルが1個 100GBあって、それが20ファイルあるみたいな状況で威力を発揮するわけです。

####どう書きます？（実装します？）

いくつかのアルゴリズムが想定されますが、Hadoopで早くなる書き方をしなければなりません。そのアルゴリズムがMapReduceです。ファイルが分散処理向きでかつ、分散処理でも矛盾が怒らない処理でなければなりません。

結論から言えば、以下のような手法を取ります。

まず、ファイル中の単語を全て分解し、それに1という値を持たせます（Map処理）。
例えば、

a - 1

というような感じ。いわゆるkey-valueですね。

これは、ファイルを分割して個別のサーバに振り分けても矛盾は発生しません（ページをまたがる文字はどうするかとかはさておき）。

で、単語毎にデータを収集し、持っている値（１）を足します（Reduce処理）。
例えば、

a - 1
a - 1

を

a - 2

とするような処理です。実際はもうちょっと複雑ですが、端折るとこんなもんです。

では、実装してみましょう。

###Map

では、Map処理から。ファイルの内容を分解し、key-value形式にします。

コマンドが実行されるとhadoop（クライアント）が、inputに指定されたフォルダ、ファイルから内容を読取り、標準入力（STDIN）で流し込んできますので、それを処理します。

例えばこんな感じ。
ここでは、"単語 - 1"という形式にします（-部分は実際にはタブ。とうか-でもいいけど）。

```php
#!/usr/bin/php

<?php
	
	//標準入力が終わり（EOF）に達するまでループ
	while(!feof(STDIN))
	{
		//標準入力から1行読み込み、かつ、前後のゴミ文字（スペースとか）を除去。
		$line = trim(fgets(STDIN));

		//スペースで行を分解
		$words = preg_split('/\s+/',$line);

		//ループしてReducerへ渡す形式を作る
		foreach($words as $word)
		{
			if(!empty($word))
			{
				//タブ区切りでkey-valueを出力（タブ区切りじゃなくてもOK）
				printf("%s\t1\n",$word);
			}
		}
	}
```

なお、これの動作を確認するには、例えば、

```bash
echo "hoge foo bar hoge" > map.txt
```
という感じで、ファイルを作成し、

```bash
chmod 755 map.php
./map.php < map.txt
```
として、作成したスクリプトに食わせて（入力して）見ましょう。

```bash
hoge	1
foo	1
hoge	1
bar	1
```

とう感じで出力されます。
ちゃんと、[単語][タブ][1]という形式になっています。

###Reduce

Mapでの処理結果は、同じく標準入力（STDIN）でReducerへ送られてきますので、受け取って処理します。こんな感じ。

```php
#!/usr/bin/php

<?php

	//結果を格納する配列の初期化
	$count = array();

	//標準入力が終わり（EOF）に達するまでループ
	while(!feof(STDIN))
	{
		//1行ずつ読み込み
		$line = trim(fgets(STDIN));

		//空行じゃなければ
		if(!empty($line))
		{	
			//タブで分割し、$key,$valueにセット
			list($key,$value) = preg_split('/\t/',$line);

			//$count["hoge"],$count["foo"]等の連想配列を作成し、
			//同じKeyのものをカウントアップしていく。
			$count[$key] += $value;
		}
	}

	//表示
	foreach($count as $key => $value)
	{
		// echo $key." ".$value."\n";
		printf("%s\t%s\n",$key,$value);
	}
```

これをテストするには、Mapの出力結果と同じ形式のファイルをreduce.txtなどとして用意し、

```txt
hoge	1
foo	1
hoge	1
bar	1
```

Mapのときと同じく標準入力でreduce.phpに食わせます。

```bash
chmod 755 reduce.php
reduce.php < reduce.txt
```

すると、

```bash
hoge	2
foo	1
bar	1
```
ちゃんと出力されています。

ここまでで、Map処理用、Reduce処理用のスクリプトができました。簡単です。

###MapReduce再び

では、2つのスクリプトを利用してHadoop Streaming処理してみたいと思います。
下記コマンドを実行してみます。

```bash
hadoop jar /usr/local/Cellar/hadoop/2.7.1/libexec/share/hadoop/tools/lib/hadoop-streaming-2.7.1.jar
 -input input
 -output output
 -mapper ./htdocs/hadoop/map.php
 -reducer ./htdocs/hadoop/reduce.php
```
>実際は1行です（改行部スペース）

すると、outputディレクトリが作成され、part-0000ファイルが出力されます。
その中身を見てみると、

```bash
cat part-00000

a	4
b	3
c	2
```

となり、サンプルと同じ結果が得られています！。

>ところでpart-x-00000という形式はもっと簡単に変更できないものか？まあ、outputフォルダの名前を変えればいいのですが。

こんな感じで、得意な言語で簡単にHadoopの恩恵に預かれるのは本当に素晴らしいことです。

ひとまずここまで。

##擬似分散モードにする

ここまではstandalone modeで利用してきたので、HDFSを利用するために擬似分散モードにしてみます。

###準備

####ssh

擬似分散モードでは、擬似サーバ間をパスフレーズ無しのSSH ログインできないといけないようなので、その設定をします。

>私は常にパスフレーズ有りの鍵を使っているので、.ssh内にoldフォルダを作成し、既存のファイルをそこに退避させました。

```bash
ssh-keygen -t rsa
```

パスフレーズを聞いてくるので、空のままEnterです。

```bash
cp id_rsa.pub authorized_keys
chmod 600 authorized_keys
```

公開鍵の方をauthorized_keysという名前に変更し、パーミッションも600にしておきます。

>なお、sshログインはデフォルトではブロックされているので、許可します。
許可は、「システム環境設定」->「共有」->左ペインの「リモートログイン」にチェックを入れます。

```bash
ssh localhost
```

で、ログインできることを確認しておきます。

>初回は、何か出るかもしれません。

###設定ファイルの編集
以下のファイルを編集します。

* core-site.xml（Hadoopの設定）
* hdfs-site.xml（HDFSの設定）

実際には、それぞれ、

* /usr/local/Cellar/hadoop/2.7.1/libexec/etc/hadoop/core-site.xml
* /usr/local/Cellar/hadoop/2.7.1/libexec/etc/hadoop/hdfs-site.xml

にあります。

core-site.xml

```xml
<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value>
    </property>
</configuration>
```

hdfs-site.xml

```xml
<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>
```

###HDFSのフォーマット

```bash
hadoop namenode -format
```

こうすることで、/tmp/hadoop-username/dfs/name/に管理ファイルが作成される。

###起動

起動、終了ファイルは/usr/local/Cellar/hadoop/2.7.1/libexec/sbin/以下にあるようです。

```bash
cd /usr/local/Cellar/hadoop/2.7.1/libexec/sbin
./start-all.sh
```
>擬似分散ではstart-dfs.shでも。
>途中でsshのログインを聞いてきたらyesとします。

起動確認は、

```bash
jps

6417 NameNode
6884 Jps
6743 ResourceManager
6840 NodeManager
6507 DataNode
6622 SecondaryNameNode
```

###動作確認

Standaloneで試したWordCountを擬似分散モードでも試してみます。

```bash
#ローカルにファイルを作成
echo "a b c a b" > test.txt

#HDFSにディレクトリ作成
hadoop fs -mkdir /input

#ローカルからHDFSにコピー
hadoop fs -put test.txt /input

#コピーできたか（再帰的に）確認
hadoop fs -ls -R /

#WordCount実行
hadoop jar /usr/local/opt/hadoop/libexec/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.1.jar wordcount  /input /output

#内容確認
hadoop fs -cat /output/part-r-00000

a	2
b	2
c	1
```



##参考

先人に感謝です。

http://qiita.com/ysk_1031/items/26752b5da1629c9db8f7
http://www.simplicity.be/article/big-data-php
http://d.hatena.ne.jp/stellaqua/20090305/1236222223