Hadoop Java API

HadoopのJava APIについて。

ToolRunner

チュートリアルのソースでは使っていないが、@ITのサンプルではToolRunnerというクラスを使って実行している。

ToolRunnerを使うと、Configurationの初期化や汎用（一般）オプション（generic option）の解釈を行ってくれる。
したがって、Configurationの初期化を自前で行い、汎用オプションも使わないのであれば、ToolRunnerを使う必要は無い。

ToolRunnerを使わない例

import org.apache.hadoop.conf.Configuration;

public class WordCount {

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "wordcount");
		job.setJarByClass(WordCount.class);
〜
		boolean success = job.waitForCompletion(true);
		System.exit(success ? 0 : 1);
	}

ToolRunnerを使う例

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;

import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class WordCount extends Configured implements Tool {

	public static void main(String[] args) throws Exception {
		int r = ToolRunner.run(new WordCount(), args);
		System.exit(r);
	}

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = getConf();
		Job job = new Job(conf, "wordcount");
		job.setJarByClass(getClass());
〜
		boolean success = job.waitForCompletion(true);
		return success ? 0 : 1;
	}

ToolRunner#run()にはToolインターフェースの実装クラスを渡す必要があるので、クラス本体（上の例ではWordCount）はToolを実装する必要がある。

ToolインターフェースはConfigurableというインターフェースを継承している。
これにはgetConf()等が宣言されており、これをConfiguredクラスが実装している為、クラス本体はConfiguredを親クラスとしている。

Toolインターフェースのメソッドの中で Configuredクラスで実装されていないのはrun()のみであり、これをクラス本体が実装する必要がある。
ToolRunnerはConfigurationの初期化を行った後でTool#run()を呼び出す。

ToolRunnerが解釈してくれる汎用オプション（汎用コマンドライン・一般オプション）の内容は、ToolRunner.printGenericCommandUsage()を実行すると表示される。

		ToolRunner.printGenericCommandUsage(System.out);

Generic options supported are
-conf <configuration file>     specify an application configuration file
-D <property=value>            use value for given property
-fs <local|namenode:port>      specify a namenode
-jt <local|jobtracker:port>    specify a job tracker
-files <comma separated list of files>    specify comma separated files to be copied to the map reduce cluster
-libjars <comma separated list of jars>    specify comma separated jar files to include in the classpath.
-archives <comma separated list of archives>    specify comma separated archives to be unarchived on the compute machines.

The general command line syntax is
bin/hadoop command [genericOptions] [commandOptions]

初期化メソッド

最初に、どのクラスを使ってMap/Reduceの処理を行うかを指定する。[2010-02-24/2010-03-19]

• TechBlogのHadoopを使いこなす(1)

Map/Reduceのコーディングでは、入力の型と出力の型に相関関係がある。
つまり、それぞれのステップの前後における入力のキー・データの型と出力のキー・データの型が一致していなければならない。

データの流れから考えて、その関係は以下のようになる。

public class MyInputFormat extends InputFormat<K, V> {
	〜
}

public class MyMapper   extends Mapper <KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

	@Override
	protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException {
		KEYOUT   outKey = 〜;
		VALUEOUT outVal = 〜;
		context.write(outKey, outVal);
	}
}

public class MyCombiner extends Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

	@Override
	protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context) throws IOException, InterruptedException {
		KEYOUT   outKey = 〜;
		VALUEOUT outVal = 〜;
		context.write(outKey, outVal);
	}
}

public class MyReducer  extends Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {

	@Override
	protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context) throws IOException, InterruptedException {
		KEYOUT   outKey = 〜;
		VALUEOUT outVal = 〜;
		context.write(outKey, outVal);
	}
}

public class MyOutputFormat extends OutputFormat<K, V> {
	〜
}

	public static void main(String[] args) {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "〜");
		job.setJarByClass(〜.class);

		job.setMapOutputKeyClass  (MyMapperのKEYOUT);
		job.setMapOutputValueClass(MyMapperのVALUEOUT);
		job.setOutputKeyClass  (MyReducerのKEYOUT);
		job.setOutputValueClass(MyReducerのVALUEOUT);

		job.setMapperClass  (MyMapper.class);
		job.setCombinerClass(MyCombiner.class);
		job.setReducerClass (MyReducer.class);

		job.setInputFormatClass (MyInputFormat.class);
		job.setOutputFormatClass(MyOutputFormat.class);
〜
	}