Hadoop MapReduceプログラムを解剖する

サンプルコードを使って、MapReduceの構造を理解する

4．環境設定

　本記事はソースコードの解釈を目的としているため、環境設定については細かく言及しませんが、Hadoop MapReduceをコンパイルして実行するための環境を整備するには、EclipseとEclipseプラグインが便利です。Eclipseにコードを入れ、コンパイルし、Driverを実行するなら、Hadoopノードとは別のノード（Windowsなど）からでもジョブをサブミットできます。

1. Linux上にインストールしたHadoopのフォルダをごっそりローカルディスクにも置きます（必要ないファイルもたくさんありますが）
2. /hadoop-0.20.2/contrib/eclipse-plugin/hadoop-0.20.2-eclipse-plugin.jarを、Eclipseのpluginsフォルダにコピーし、再起動します（各ビルドにも入っています）
3. Eclipseのワークスペース上で、MapReduceプロジェクトを作ります。このとき、Hadoopのロケーションを聞かれるので、先ほどのフォルダを指定します

　これだけで、Hadoop MapReduceに必要なJARにクラスパスが通ったJavaプロジェクトとなります。ネームノードなどの指定はコード中で行うので、プロジェクトの設定には必要ありません。

　このMapReduceプロジェクトは、プラグインが正常に動作する環境（Hadoopのネームノード上など）であれば右クリックしてサブミットできるのですが、Hadoopノードとは別のWindows環境などではその機能が働きませんので、MapReduceジョブをJARにエクスポートして、別のプロジェクトからサブミットする、という方法がスマートです。HDFSへのアクセス機能は正しく設定すれば働くので、利用すると便利です（クラスター外のWindowsで働かない理由は、シェルコマンドを使うからと推測されます）。

　筆者は、MapReduceジョブを入れたプロジェクトと、Driverを入れたプロジェクトを2つセットで作ります。MapReduceを作ったプロジェクトをJARとしてDriverプロジェクトに作ったlibフォルダにエクスポートしてクラスパスへ追加し、そのJARをMapReduceジョブとしてサブミットするのです（MapperやReducerのコードを修正したら、エクスポートしなおすのを忘れないようにします）。

図3．MapReduceジョブのJARとDriverの関係

5．MapperとReducer

　リスト1、リスト2に、それぞれ、MapperとReducerのサンプルを示しています。キーとバリューの関係を、赤→緑→青で表現し、先ほどの図と一致させることで比較しやすいようにしてみました。

リスト1．Mapper

package com.yone.mapreduce;

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
   private final static Text word = new Text();
  private final static IntWritable one = new IntWritable(1);

   public void map(LongWritable key, Text value, Context context)
     throws IOException, InterruptedException {

     // すべての単語に「1」という数値を付けて、出力する
     String line = value.toString();
     StringTokenizer tokenizer = new StringTokenizer(line);
      while (tokenizer.hasMoreTokens()) {
        word.set(tokenizer.nextToken());
        context.write(word, one);
     }

   }
}

リスト2．Reducer

package com.yone.mapreduce;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MyReducer extends
   Reducer<Text, IntWritable, Text, IntWritable> {
   private IntWritable result = new IntWritable();

   public void reduce(Text key, Iterable<IntWritable> values, Context context)
      throws IOException, InterruptedException {

     // 同じキーに対して、すべてのバリュー（数値が大量に渡ってくる）を足し算して合計する
      int sum = 0;
      for (IntWritable val : values) {
         sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
   }
}

　この2つのファイルをMepReduceプロジェクト（クラスパスの通ったJavaプロジェクト）に作り、JARにエクスポートします。