6．Driver
Driverは、ジョブをキックするプログラムです。Configuredを継承し、Toolを実装します。単体で動かす場合はmain関数からToolRunnerを使って起動し、スレッドが起動するので、そのスレッドのrun関数でジョブの構成情報を登録して、ジョブが終了するのを待ちます。
package com.yone.driver;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import com.yone.mapreduce.MyMapper;
import com.yone.mapreduce.MyReducer;
public class MyDriver extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("mapred.job.tracker", "name.yone.com:9001");
conf.set("fs.default.name", "hdfs://name.yone.com:9000/");
Job job = new Job(conf, "WordCount");
job.setJarByClass(MyMapper.class);
job.setMapperClass(MyMapper.class);
job.setCombinerClass(MyReducer.class);
job.setReducerClass(MyReducer.class);
TextInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
return job.waitForCompletion(true) ? 0 : -1;
}
public static void main(String[] args) throws Exception {
ToolRunner.run(new MyDriver(),
new String[] {
"/user/hdpadmin/input",
"/user/yone/output" } // 入力ファイルと出力フォルダー
);
}
}
7．コードを解釈する
では、1つ1つコードを解釈していきましょう。サンプルコードを眺めながら読み進めてください。
Driverの構造
DriverはConfiguredを継承し、Toolを実装します。mainメソッドとrunメソッドでできています。Toolは、スレッドを起動するためのインターフェースであるRunnableのサブインターフェースなので、このrunはスレッドで実行されます。ToolRunner#runを実行することで、Driverインスタンスのrunメソッドがスレッドで実行されます。何度も実行すれば、その分だけジョブがサブミットされます。
runメソッドの中では、Configurationオブジェクト「conf」と、Jobオブジェクト「job」をインスタンス化します。この2つのオブジェクトでジョブがサブミットされ、監視されます。
このサンプルは、Hadoopクラスターの外からMapReduceジョブをサブミットする前提でコーディングされているので、conf.setを使って、HDFSネームノードとジョブトラッカーノードの場所を教えています（runメソッドの冒頭）。
setJarByClassで指定されたクラスが含まれているJARがサブミット対象となります。Mapper、Reducerは、setMapperClass、.setCombinerClass、.setReducerClassで設定されます。
すべての設定が終わったら、jobオブジェクトに対して「waitForCompletion」を実行することでジョブがサブミットされ、実行状況が監視されます。
入力ファイルの設定
入力ファイルの設定Driverのrunメソッド中で、InputFormatに対して行います。今回は、デフォルトのTextInputFormatを使うので、そのクラスメソッドのaddInputPaths()メソッドでセットしています。main関数から渡すようにしている理由は、こうすることで、ToolRunner#runを違う引数で2回呼べば、同じrunメソッドで違うファイルを対象に平行してジョブを実行することもできます。
ここでは/user/hdpadmin/inputというパスを指定しています。固有のファイルを指定することもできますが、このようにパスを指定すると、そのフォルダにあるすべてのファイルが入力として使われます。
Mapに渡される<Key, Value>
MapperはHadoopが提供するMapperクラスを継承して作ります。ジェネリククラスとなっているため、型宣言が必要です。
デフォルトであるTextInputFormatを使うため、キーとバリューのペアは必然的にLongWritableとText型となります。Mapperクラスの赤字で示した箇所（ジェネリクスの型宣言）がこれを示しています。これは必然なので、このとおりに指定しなければいけなく、整合していなければなりません。
シャッフルに使われる<Key, Value>
Reducerも同様にReducerクラスを継承して作ります。シャッフルに使われるWritableの型は、MapperとReducerの間で整合性がとれていれば良いので、Driverにはなんの記述もありません。Mapperが出力するときのwriteメソッドで出力されるオブジェクトの型と、Reducerのジェネリクス型宣言が一致していればいいです。これが緑でしめされている箇所です。
なお、シャッフルに使われるKeyは、ソートのための比較メソッド（WritableComarator）を持つ、WritableComparableインターフェースを実装する必要があります。これは、Writableのサブインターフェースです。
出力に使われる<Key, Value>
出力に使われるキー・バリューは、Reducerの型宣言、出力（writeメソッド）、Driverの出力設定がすべて一致している必要があります。青字の箇所です。setOutputKeyClass、setOutputValueClassといったメソッドで、Driver側でも予め設定している点にも注意してください。
出力先（ファイル）の設定
入力ファイルの設定とほぼ同じで、Driverのrunメソッド内で、FileInputFormatクラスメソッドで設定します。実は、入力ファイルと同様にTextOutputFormatに対してセットしてもよいのですが、スーパークラスのFileInputFormatに設定してもよいことを示すためにあえてFileInputFormatを使っています。
また、出力先は「add」ではなく、「set」です。複数の出力先は指定できません。設定の値はフォルダ名となります。出力データにはシーケンス番号がついて、自動的に分割されて出力されます。
