hadoop - mapreduce成功后如何删除输入文件

Question

我们有一个系统可以接收指定目录中的文件，并定期启动一个 mapreduce 作业来打开文件并处理其中的文件。为了避免下次重新处理相同的文件，我们连接到 RecordReader 上的 close() 方法，以便在读取最后一个条目后将其删除。

这种方法的问题(我们认为)是，如果一个特定的映射失败，下一个再次尝试它的映射器会发现原始文件已被记录读取器从第一个读取器中删除，并且它会爆炸。我们认为要走的路是等到所有映射和归约完成后再删除输入文件。

这是最好的方法吗？

如果是这样，我们如何从主程序中获取系统找到的所有输入文件的列表？ (我们不能只删除整个输入目录，新文件可能存在)

即:

   . . .

   job.waitForCompletion(true);

   (we're done, delete input files, how?)

   return 0;
}

Answer 1

一些评论。

1. 我认为这种设计容易让人心痛。当您发现有人将困惑的算法部署到您的 MR 集群并且您必须回填一个月的文件时会发生什么？他们现在走了。如果处理时间比预期的长，并且需要在旧作业完全完成之前开始新作业，会发生什么情况？文件太多，有些文件需要重新处理。当文件仍在运行中时工作何时开始呢？等等

2. 摆脱这个陷阱的一个方法是让文件根据时间转到一个轮换位置，然后自己清除记录或(在 S3 之类的情况下)建立一个保留策略，允许特定的操作窗口。此外，无论后端 map reduce 处理在做什么，都可以是幂等的:处理同一条记录两次与处理一次应该没有任何不同。有些事情告诉我，如果你正在减少你的数据集，那么该属性将很难保证。

3. 至少您可以重命名您处理的文件而不是立即删除它们，并使用 glob 表达式来定义不包括重命名文件的输入。正如我上面提到的，仍然存在竞争条件。

4. 您可以使用 Amazon SQS 等队列来记录存档的传送，并且您的 InputFormat 可以提取这些条目，而不是在确定输入拆分时列出存档文件夹。但是，如果没有额外的基础设施，再处理或回填就会成为问题。

5. 综上所述，拆分列表是由 InputFormat 生成的。围绕它编写一个装饰器，你可以将拆分列表存放在任何你想在工作完成后供主人使用的地方。