hadoop - Mapreduce 自定义格式。现实世界中是否有很多情况可以编写自定义输入和输出格式？

Question

我知道 Hadoop 支持各种输入和输出格式，如 Text、Json、Avo、Sequence 文件等。

在 Hadoop 的现实世界中，我们是否遇到过开发人员必须开发自定义输入和输出格式的情况。对不起，Hadoop 世界的新手。

非常感谢纳特

Answer 1

CustomInputFormat can be written when you need to customize inputrecord reading..

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);
}