カスタム出力フォーマットの基本概念
HadoopのOutputFormatは、MapReduceジョブの出力データの形式と書き込み方法を制御します。デフォルトではTextOutputFormatが使用され、キーと値のペアをテキストファイルに書き込みますが、特定の要件に対応するにはカスタム出力フォーマットの実装が必要になります。
実装手順
- FileOutputFormatを継承したクラスを作成
- getRecordWriterメソッドを実装してRecordWriterを返す
- RecordWriterを継承したクラスでwriteメソッドを定義
実装例: 複数ファイル出力
public class CustomOutputFormat extends FileOutputFormat<Text, Text> {
@Override
public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext context) {
Path outputPath = FileOutputFormat.getOutputPath(context);
Configuration conf = context.getConfiguration();
return new MultiFileRecordWriter(outputPath, conf);
}
}
class MultiFileRecordWriter extends RecordWriter<Text, Text> {
private Map<String, FSDataOutputStream> streams = new HashMap<>();
private Path basePath;
private Configuration conf;
public MultiFileRecordWriter(Path basePath, Configuration conf) {
this.basePath = basePath;
this.conf = conf;
}
@Override
public void write(Text key, Text value) throws IOException {
String filename = key.toString() + ".dat";
FSDataOutputStream out = streams.get(filename);
if (out == null) {
Path filePath = new Path(basePath, filename);
FileSystem fs = filePath.getFileSystem(conf);
out = fs.create(filePath);
streams.put(filename, out);
}
out.writeBytes(value.toString() + "\n");
}
@Override
public void close(TaskAttemptContext context) throws IOException {
for (FSDataOutputStream out : streams.values()) {
out.close();
}
}
}
MapperとReducerの実装例
public class CustomOutputJob {
public static class DataMapper extends Mapper<LongWritable, Text, Text, Text> {
public void map(LongWritable key, Text value, Context context) {
// データ処理ロジック
context.write(new Text(category), new Text(processedData));
}
}
public static class DataReducer extends Reducer<Text, Text, Text, Text> {
public void reduce(Text key, Iterable<Text> values, Context context) {
for (Text val : values) {
context.write(key, val);
}
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Custom Output Example");
job.setOutputFormatClass(CustomOutputFormat.class);
// その他のジョブ設定
}
}
実行と結果確認
ジョブ実行後、HDFS上で結果を確認します:
hdfs dfs -ls /output/path
hdfs dfs -cat /output/path/category1.dat