マッパーは、複数のディレクトリからファイルを読み取ることができませんでした。誰でも助けてもらえますか?各マッパーで 1 つのファイルを読み取る必要があります。複数の入力パスを追加し、カスタムの WholeFileInputFormat、WholeFileRecordReader を実装しました。map メソッドでは、入力キーは必要ありません。各マップがファイル全体を読み取れるようにします。
コマンド ライン: hadoop jar AutoProduce.jar Autoproduce /input_a /input_b /output 2 つの入力パスを指定しました----1.input_a; 2.input_b;
メソッドのスニペットを実行します。
Job job = new Job(getConf());
job.setInputFormatClass(WholeFileInputFormat.class);
FileInputFormat.setInputPaths(job, new Path(args[0]), new Path(args[1]));
FileOutputFormat.setOutputPath(job, new Path(args[2]));
map メソッドのスニペット:
public void map(NullWritable key, BytesWritable value, Context context){
FileSplit fileSplit = (FileSplit) context.getInputSplit();
System.out.println("Directory :" + fileSplit.getPath().toString());
......
}
カスタム WholeFileInputFormat:
class WholeFileInputFormat extends FileInputFormat<NullWritable, BytesWritable> {
@Override
protected boolean isSplitable(JobContext context, Path file) {
return false;
}
@Override
public RecordReader<NullWritable, BytesWritable> createRecordReader(
InputSplit split, TaskAttemptContext context) throws IOException,
InterruptedException {
WholeFileRecordReader reader = new WholeFileRecordReader();
reader.initialize(split, context);
return reader;
}
}
カスタム WholeFileRecordReader:
class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> {
private FileSplit fileSplit;
private Configuration conf;
private BytesWritable value = new BytesWritable();
private boolean processed = false;
@Override
public void initialize(InputSplit split, TaskAttemptContext context)
throws IOException, InterruptedException {
this.fileSplit = (FileSplit) split;
this.conf = context.getConfiguration();
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
if (!processed) {
byte[] contents = new byte[(int) fileSplit.getLength()];
Path file = fileSplit.getPath();
FileSystem fs = file.getFileSystem(conf);
FSDataInputStream in = null;
try {
in = fs.open(file);
IOUtils.readFully(in, contents, 0, contents.length);
value.set(contents, 0, contents.length);
} finally {
IOUtils.closeStream(in);
}
processed = true;
return true;
}
return false;
}
@Override
public NullWritable getCurrentKey() throws IOException,InterruptedException {
return NullWritable.get();
}
@Override
public BytesWritable getCurrentValue() throws IOException,InterruptedException {
return value;
}
@Override
public float getProgress() throws IOException {
return processed ? 1.0f : 0.0f;
}
@Override
public void close() throws IOException {
// do nothing
}
}
問題:
2 つの入力パスを設定した後、すべてのマップ タスクは 1 つのディレクトリからのみファイルを読み取ります。
前もって感謝します。