hadoop - 完全取消默认输出目录 - MapReduce

Question

我有一个使用 org.apache.hadoop.mapreduce.lib.output.MultipleOutputs 编写多个输出的代码。

Reducer 将结果写入预先创建的位置，因此我不需要默认的 o/p 目录(其中包含 _history 和 _SUCCESS 目录)。

每次再次运行我的工作之前，我都必须删除它们。

所以我删除了 TextOutputFormat.setOutputPath(job1,new Path(outputPath)); 行。但是，这给了我(预期)错误 org.apache.hadoop.mapred.InvalidJobConfException: Output directory not set

驱动类:

MultipleOutputs.addNamedOutput(job1, "path1", TextOutputFormat.class, Text.class,LongWritable.class);
MultipleOutputs.addNamedOutput(job1, "path2", TextOutputFormat.class, Text.class,LongWritable.class);
LazyOutputFormat.setOutputFormatClass(job1,TextOutputFormat.class);

reducer 类:

if(condition1)
    mos.write("path1", key, new LongWritable(value), path_list[0]);
else
    mos.write("path2", key, new LongWritable(value), path_list[1]);

是否有避免指定默认输出目录的解决方法？

Answer 1

我不认为 _SUCCESS 是一个目录，另一个 history 目录位于 _logs 目录中。

首先 TextOutputFormat.setOutputPath(job1,new Path(outputPath)); 很重要，因为当作业运行时，Hadoop 将此路径作为工作目录来创建临时文件等对于不同的任务(_temporary 目录)。这个 _temporary 目录和文件最终会在作业结束时被删除。文件 _SUCCESS 和历史目录实际上是保留在工作目录下并在作业成功完成后保留的内容。 _SUCCESS 文件是一种标志，表示作业实际运行成功。请看at this link .

您的文件 _SUCCESS 的创建是由 TextOutputFormat 类完成的，您实际使用的是 FileOutputComitter 类。 FileOutputCommiter 类定义了这样的函数 --

 public static final String SUCCEEDED_FILE_NAME = "_SUCCESS";
/**
   * Delete the temporary directory, including all of the work directories.
   * This is called for all jobs whose final run state is SUCCEEDED
   * @param context the job's context.
   */
  public void commitJob(JobContext context) throws IOException {
    // delete the _temporary folder
    cleanupJob(context);
    // check if the o/p dir should be marked
    if (shouldMarkOutputDir(context.getConfiguration())) {
      // create a _success file in the o/p folder
      markOutputDirSuccessful(context);
    }
  }

// Mark the output dir of the job for which the context is passed.
  private void markOutputDirSuccessful(JobContext context)
  throws IOException {
    if (outputPath != null) {
      FileSystem fileSys = outputPath.getFileSystem(context.getConfiguration());
      if (fileSys.exists(outputPath)) {
        // create a file in the folder to mark it
        Path filePath = new Path(outputPath, SUCCEEDED_FILE_NAME);
        fileSys.create(filePath).close();
      }
    }
  }

由于 markOutputDirSuccessful() 是私有(private)的，因此您必须改写 commitJob() 以绕过 SUCCEEDED_FILE_NAME 创建过程并实现您想要的。

下一个目录 _logs 非常重要，如果您想稍后使用 hadoop HistoryViewer 实际获取作业运行情况的报告。

我认为，当您使用相同的输出目录作为另一个作业的输入时，由于在 Hadoop 中设置了过滤器，文件 *_SUCCESS* 和目录 *_logs* 将被忽略。

此外，当您为 MultipleOutputs 定义命名输出时，您可以改为写入您在 TextOutputFormat.setOutputPath() 函数中提到的输出路径内的子目录，然后使用该路径作为下一个作业的输入正在运行。

我真的不知道 _SUCCESS 和 _logs 会如何打扰您？

谢谢