hadoop - 使用启用了推测执行的 Hadoop 多输出写入多个输出

Question

我已经编写了一个 mapreduce 程序来处理日志。除了实际输出之外，该作业还将边数据写入驱动程序代码中设置的输出路径之外的不同位置。但是启用推测执行后，终止任务尝试的输出不会已删除。有办法避免这个问题吗？除了写入正常输出位置并在作业完成后复制到外部位置之外，是否有可能解决该问题？

是否可以使用“OutputCommitter”来解决这个问题？

有人试过吗？任何帮助将不胜感激。

Answer 1

是的，可以使用 FileOutputCommitter，当任务成功时将临时任务目录的内容移动到最终输出目录，并删除原始任务目录。

我相信大多数在 Hadoop 中扩展 FileOutputFormat 的内置输出格式都使用 OutputCommitter，默认情况下是 FileOutputCommitter。

这是 FileOutputFormat 的代码

public synchronized 
     OutputCommitter getOutputCommitter(TaskAttemptContext context
                                        ) throws IOException {
    if (committer == null) {
      Path output = getOutputPath(context);
      committer = new FileOutputCommitter(output, context);
    }
    return committer;
  }

要写入多个路径，您可以查看 MultipleOutputs ，默认情况下使用 OutputCommitter。

或者您可以创建自己的输出格式并扩展 FileOutputFomat 并在 FileOutputFormat 中覆盖上述函数，查看 FileOutputCommitter 代码创建您自己的 OutputCommitter 实现。

在 FileOoutputcommitter 代码中，您将找到您可能感兴趣的函数 -

    /**
   * Delete the work directory
   */
  @Override
  public void abortTask(TaskAttemptContext context) {
    try {
      if (workPath != null) { 
        context.progress();
        outputFileSystem.delete(workPath, true);
      }
    } catch (IOException ie) {
      LOG.warn("Error discarding output" + StringUtils.stringifyException(ie));
    }
  }

If a task succeeds then commitTask()is called, which in the default implementation moves the temporaray task output directory (which has the task attempt ID in its name to avoid conflicts between task attempts) to the final output path, ${mapred.out put.dir}. Otherwise, the framework calls abortTask(), which deletes the temporary task output directory.