hadoop - 何时使用Map Reduce作业的自定义输入格式

Question

使用Map Reduce编程时，何时应使用自定义输入格式？

假设我有一个文件，需要逐行读取，并且有15列以管道分隔，我应该使用自定义输入格式吗？

在这种情况下，我可以使用TextInput格式以及“自定义输入格式”。

Answer 1

CustomInputFormat can be written when you need to customize inputrecord reading. But in your case you need not have such an implementation.

see below example of CustomInputFormat out of many such...

示例:读取段落作为输入记录
如果您正在使用Hadoop MapReduce或使用AWS EMR，则可能存在一个用例，其中输入文件将段落作为键值记录而不是单行一致(请考虑诸如分析新闻评论之类的方案)。因此，如果您需要一次将一条完整的段落作为一条记录处理，而不是将一行作为输入处理，则需要自定义 **TextInputFormat**的默认行为，即默认情况下将每一行读取为将一个完整的段落作为一个输入键读取值对，以便在MapReduce作业中进行进一步处理。
这要求我们创建一个自定义记录读取器，可以通过实现 class RecordReader来完成。 next()方法是告诉记录读取器取一个段落而不是一行的地方。请参阅以下实现，这是不言自明的:

public class ParagraphRecordReader implements RecordReader<LongWritable, Text> {
private LineRecordReader lineRecord;
private LongWritable lineKey;
private Text lineValue;
public ParagraphRecordReader(JobConf conf, FileSplit split) throws IOException {
lineRecord = new LineRecordReader(conf, split);
lineKey = lineRecord.createKey();
lineValue = lineRecord.createValue();
}
@Override
public void close() throws IOException {
lineRecord.close();
}
@Override
public LongWritable createKey() {
return new LongWritable();

}
@Override
public Text createValue() {
return new Text("");    
}
@Override
public float getProgress() throws IOException {
return lineRecord.getPos();    
}

@Override
public synchronized boolean next(LongWritable key, Text value) throws IOException {
boolean appended, isNextLineAvailable;
boolean retval;
byte space[] = {' '};
value.clear();
isNextLineAvailable = false;
do {
appended = false;
retval = lineRecord.next(lineKey, lineValue);
if (retval) {
if (lineValue.toString().length() > 0) {
byte[] rawline = lineValue.getBytes();
int rawlinelen = lineValue.getLength();
value.append(rawline, 0, rawlinelen);
value.append(space, 0, 1);
appended = true;
}
isNextLineAvailable = true;
}
} while (appended);

return isNextLineAvailable;
}

@Override
public long getPos() throws IOException {
return lineRecord.getPos();
}
}

使用ParagraphRecordReader实现，我们将需要扩展TextInputFormat来创建自定义InputFomat，方法是仅重写getRecordReader方法并返回ParagraphRecordReader的对象以覆盖默认行为。 ParagrapghInputFormat看起来像:

public class ParagrapghInputFormat extends TextInputFormat
{
@Override
public RecordReader<LongWritable, Text> getRecordReader(InputSplit split, JobConf conf, Reporter reporter)throws IOException {
reporter.setStatus(split.toString());
return new ParagraphRecordReader(conf, (FileSplit)split);
}
}

确保作业配置使用我们的自定义输入格式实现将数据读入MapReduce作业。只需将输入格式类型设置为ParagraphInputFormat即可，如下所示:
conf.setInputFormat(ParagraphInputFormat.class);
通过上述更改，我们可以将段落作为输入记录读取到MapReduce程序中。
我们假设输入文件如下所示:
一个简单的映射器代码如下所示:

@Override
public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter)
throws IOException {
System.out.println(key+" : "+value);
}