Hadoopの出力ファイルにヘッダを出力する

MapReduce処理を行った後の出力ファイルにヘッダを出力したい場合はTextOutputFormatを拡張する。

Mapperの実装でJobConfにヘッダを設定する。元データからヘッダを取る場合は以下のように設定

public class MapperSample extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {

	private JobConf conf;

	public void map(LongWritable key, Text value,
			OutputCollector<Text, Text> output, Reporter reporter)
			throws IOException {
		if (isHeader()) {
			// ヘッダの場合はJobConfに設定
			conf.setStrings("jp.yustam.hadoop.header", value.toString());
		} else {
			// ヘッダ以外の場合は出力
			output.collect(new Text("output"), value);
		}
	}

	@Override
	public void configure(JobConf conf) {
		this.conf = conf;
	}
}

入力データの中からヘッダを抽出する条件が複雑になる場合はコマンドライン引数から指定しても良いかも

TextOutputFormatを継承したクラスを作成しJobConfのOutputFormatのクラスに設定

public class TextOutputFormatWithHeader extends TextOutputFormat<NullWritable, Text> {
	@Override
	public RecordWriter<NullWritable, Text> getRecordWriter(FileSystem ignored,
			JobConf job, String name, Progressable progress) throws IOException {
		RecordWriter<NullWritable, Text> writer = super.getRecordWriter(ignored, job, name, progress);
		// ヘッダを出力
		String header = job.get("jp.yustam.hadoop.header");
		if (header != null) {
			writer.write(NullWritable.get(), new Text(header));
		}
		return writer;
	}
}