java - 具有推测执行的 Hadoop 多输出

Question

我的任务是将 avro 输出写入由输入记录的几个字段组织的多个目录中。

For example : Process records of countries across years and write in a directory structure of country/year eg:outputs/usa/2015/outputs_usa_2015.avro outputs/uk/2014/outputs_uk_2014.avro

AvroMultipleOutputs multipleOutputs=new AvroMultipleOutputs(context);
....
....
     multipleOutputs.write("output", avroKey, NullWritable.get(), 
            OUTPUT_DIR + "/" + record.getCountry() + "/" + record.getYear() + "/outputs_" +record.getCountry()+"_"+ record.getYear());

以下代码将使用什么输出提交器来编写输出。与推测执行一起使用是否不安全？ 通过推测执行，这会导致(可能会导致)org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException

在这篇文章中 Hadoop Reducer: How can I output to multiple directories using speculative execution?建议使用自定义输出提交器

来自 hadoop AvroMultipleOutputs 的以下代码没有说明推测执行有任何问题

 private synchronized RecordWriter getRecordWriter(TaskAttemptContext taskContext,
          String baseFileName) throws IOException, InterruptedException {

    writer =
                ((OutputFormat) ReflectionUtils.newInstance(taskContext.getOutputFormatClass(),
                    taskContext.getConfiguration())).getRecordWriter(taskContext);
...
}

如果基本输出路径在作业目录之外，写入方法也不会记录任何问题

public void write(String namedOutput, Object key, Object value, String baseOutputPath)

在作业目录外写入时，AvroMultipleOutputs(其他输出)是否存在预测执行的真正问题？如果，那么我如何重写 AvroMultipleOutputs 使其拥有自己的输出提交器。我在 AvroMultipleOutputs 中看不到任何输出格式，它使用其输出提交器

Answer 1

AvroMultipleOutputs将使用 OutputFormat您在添加命名输出时已注册到作业配置，例如使用 addNamedOutput API 来自 AvroMultipleOutputs (例如 AvroKeyValueOutputFormat )。

与 AvroMultipleOutputs ，您可能无法使用推测任务执行功能。即使覆盖它也无济于事或并不简单。

相反，您应该编写自己的 OutputFormat (很可能扩展一种可用的 Avro 输出格式，例如 AvroKeyValueOutputFormat )，并覆盖/实现其 getRecordWriter API，它将返回一个 RecordWriter实例说MainRecordWriter (仅供引用)。

这MainRecordWriter将维护 RecordWriter 的 map (例如 AvroKeyValueRecordWriter )实例。这些中的每一个RecordWriter实例将属于输出文件之一。在 write API MainRecordWriter ，你会得到实际的 RecordWriter从 map 实例(基于您要写入的记录)，并使用此记录编写器写入记录。所以MainRecordWriter将只是作为多个 RecordWriter 的包装器工作实例。

对于一些类似的实现，你可能会喜欢研究MultiStorage的代码来自 piggybank 的类(class)图书馆。