如何再次覆盖/重用 Hadoop 作业的现有输出路径

Question

当我每天运行 Hadoop 作业时，我想覆盖/重用现有的输出目录。 实际上输出目录将存储每天作业运行结果的汇总输出。 如果我指定相同的输出目录，则会出现错误“输出目录已存在”。

如何绕过这个验证？

Answer 1

在运行作业之前删除目录怎么样？

执行此操作：

hadoop fs -rmr /path/to/your/output/

您可以通过 shell或通过 Java API

// configuration should contain reference to your namenode
FileSystem fs = FileSystem.get(new Configuration());
// true stands for recursively deleting the folder you gave
fs.delete(new Path("/path/to/your/output"), true);

Answer 2

Jungblut 的答案是您的直接解决方案。由于我从不相信自动化流程会删除内容（我个人），因此我会建议一种替代方案：

我建议您不要尝试覆盖，而是使作业的输出名称动态化，包括它运行的时间。

类似于“/path/to/your/output-2011-10-09-23-04/”。这样，您就可以保留旧的作业输出，以防您需要重新访问。在我的系统中，每天运行 10 多个作业，我们将输出结构为：/output/job1/2011/10/09 /job1out/part-r-xxxxx、/output/job1/2011/10/10/job1out/part-r-xxxxx 等

Answer 3

Hadoop 的 TextInputFormat （我猜您正在使用它）不允许覆盖现有目录。可能是为了原谅你发现自己错误地删除了你（和你的集群）努力工作的东西所带来的痛苦。

但是，如果您确定希望输出文件夹被作业覆盖，我相信最干净的方法是像这样更改 TextOutputFormat ：

public class OverwriteTextOutputFormat<K, V> extends TextOutputFormat<K, V>
{
      public RecordWriter<K, V> 
      getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException 
      {
          Configuration conf = job.getConfiguration();
          boolean isCompressed = getCompressOutput(job);
          String keyValueSeparator= conf.get("mapred.textoutputformat.separator","\t");
          CompressionCodec codec = null;
          String extension = "";
          if (isCompressed) 
          {
              Class<? extends CompressionCodec> codecClass = 
                      getOutputCompressorClass(job, GzipCodec.class);
              codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
              extension = codec.getDefaultExtension();
          }
          Path file = getDefaultWorkFile(job, extension);
          FileSystem fs = file.getFileSystem(conf);
          FSDataOutputStream fileOut = fs.create(file, true);
          if (!isCompressed) 
          {
              return new LineRecordWriter<K, V>(fileOut, keyValueSeparator);
          } 
          else 
          {
              return new LineRecordWriter<K, V>(new DataOutputStream(codec.createOutputStream(fileOut)),keyValueSeparator);
          }
      }
}

现在您正在创建 FSDataOutputStream code> (fs.create(file, true)) 并覆盖=true。

Answer 4

Hadoop 已经支持您试图通过允许作业的多个输入路径来实现的效果。不要尝试使用单个文件目录来添加更多文件，而是使用一个目录来添加新目录。要将聚合结果用作输入，只需将输入 glob 指定为子目录上的通配符（例如，my-aggregate-output/*）。要将新数据“追加”到聚合作为输出，只需指定聚合的一个新的唯一子目录作为输出目录，通常使用从输入数据派生的时间戳或某些序列号（例如 my-aggregate-output/ 20140415154424）。

Answer 5

如果将输入文件（例如带有附加条目）从本地文件系统加载到 hadoop 分布式文件系统，如下所示：

hdfs dfs -put  /mylocalfile /user/cloudera/purchase

那么还可以使用 -f 覆盖/重用现有的输出目录。无需删除或重新创建文件夹

hdfs dfs -put -f  /updated_mylocalfile /user/cloudera/purchase

Answer 6

Hadoop 遵循“一次写入，多次读取”的理念。因此，当您尝试再次写入目录时，它会假设必须创建一个新目录（写入一次），但它已经存在，因此它抱怨。您可以通过hadoop fs -rmr /path/to/your/output/删除它。最好创建一个动态目录（例如，基于时间戳或哈希值）以保存数据。

Answer 7

您可以按时间为每次执行创建一个输出子目录。例如，假设您希望来自用户的输出目录，然后将其设置如下：

FileOutputFormat.setOutputPath(job, new Path(args[1]);

通过以下行更改此设置：

String timeStamp = new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss", Locale.US).format(new Timestamp(System.currentTimeMillis()));
FileOutputFormat.setOutputPath(job, new Path(args[1] + "/" + timeStamp));

Answer 8

我有一个类似的用例，我使用 MultipleOutputs 来解决这个问题。

例如，如果我希望不同的 MapReduce 作业写入同一目录 /outputDir/。作业 1 写入 /outputDir/job1-part1.txt，作业 2 写入 /outputDir/job1-part2.txt（不删除现有文件）。

在 main 中，将输出目录设置为随机目录（可以在新作业运行之前将其删除）。

FileInputFormat.addInputPath(job, new Path("/randomPath"));

在 reducer/mapper 中，使用 MultipleOutputs 并将 writer 设置为写入所需的目录

public void setup(Context context) {
    MultipleOutputs mos = new MultipleOutputs(context);
}

： ：

mos.write(key, value, "/outputDir/fileOfJobX.txt")

---

但是，我的用例比这有点复杂。如果只是写入同一个平面目录，您可以写入不同的目录并运行脚本来迁移文件，例如：hadoop fs -mv /tmp/* /outputDir

在我的用例中，每个 MapReduce 作业根据正在写入的消息的值写入不同的子目录。目录结构可以是多层的，例如：

/outputDir/
    messageTypeA/
        messageSubTypeA1/
            job1Output/
                job1-part1.txt
                job1-part2.txt
                ...
            job2Output/
                job2-part1.txt
                ...

        messageSubTypeA2/
        ...
    messageTypeB/
    ...

每个 Mapreduce 作业可以写入数千个子目录。写入 tmp 目录并将每个文件移动到正确目录的成本很高。

Answer 9

我遇到了这个确切的问题，它源于类 FileOutputFormat 中的 checkOutputSpecs 中引发的异常。就我而言，我希望有许多作业将文件添加到已经存在的目录中，并且我保证这些文件具有唯一的名称。

我通过创建一个输出格式类来解决这个问题，该类仅覆盖 checkOutputSpecs 方法，并抑制（忽略）在检查目录是否已存在时抛出的 FileAlreadyExistsException 。

public class OverwriteTextOutputFormat<K, V> extends TextOutputFormat<K, V> {
    @Override
    public void checkOutputSpecs(JobContext job) throws IOException {
        try {
            super.checkOutputSpecs(job);
        }catch (FileAlreadyExistsException ignored){
            // Suffocate the exception
        }
    }
}

在作业配置中，我使用了 LazyOutputFormat 和 MultipleOutputs。

LazyOutputFormat.setOutputFormatClass(job, OverwriteTextOutputFormat.class);

Answer 10

您需要在主类中添加设置：

//Configuring the output path from the filesystem into the job
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//auto_delete output dir
OutputPath.getFileSystem(conf).delete(OutputPath);