我们有一个系统，它接收指定目录中的档案，并定期启动mapreduce作业，打开档案并处理其中的文件。为了避免下次重新处理相同的归档文件，我们在recordreader上使用close（）方法，以便在读取最后一个条目后将其删除。
这种方法（我们认为）的问题是，如果某个特定的Map失败，下一个再次尝试它的Map程序会发现原始文件已被记录读取器从第一个文件中删除，它会爆炸出来。我们认为应该推迟，直到所有的Map和缩减都完成，然后删除输入档案。
这是最好的方法吗？
如果是这样，我们如何从主程序中获得系统找到的所有输入文件的列表(我们不能只删除整个输入目录，可能会出现新文件）
即。：
. . .
job.waitForCompletion(true);
(we're done, delete input files, how?)
return 0;
}
4条答案
按热度按时间krugob8w1#
几句评论。
我认为这种设计容易让人心痛。当您发现有人在您的mr集群中部署了一个乱七八糟的算法，而您必须回填一个月的归档文件时，会发生什么情况？他们走了。如果处理时间比预期的长，并且在旧作业完成之前需要启动新作业，会发生什么情况？存在太多的文件，有些文件被重新处理。当存档仍在运行时，作业何时开始呢？等。
走出这个陷阱的一个方法是让归档文件根据时间转到一个轮换的位置，或者自己清除记录，或者（在类似s3的情况下）建立一个允许特定操作窗口的保留策略。另外，无论后端map reduce处理在做什么，都可能是幂等的：处理同一条记录两次应该和处理一次没有任何区别。有些东西告诉我，如果你减少你的数据集，这个属性将很难保证。
至少可以重命名处理过的文件，而不是立即删除它们，并使用glob表达式定义不包含重命名文件的输入。正如我上面提到的，还有一些比赛条件。
您可以使用一个队列（如amazonsqs）来记录存档的传递，inputformat可以在确定输入拆分时提取这些条目，而不是列出存档文件夹。但如果没有额外的基础设施，再加工或回填就成了问题。
尽管如此，拆分列表是由inputformat生成的。写一个装饰器围绕着它，你可以把分割列表存放在任何你想让主人在工作完成后使用的地方。
rjee0c152#
我认为您应该使用apacheoozie来管理您的工作流。来自oozie的网站（粗体是我的）：
oozie是一个工作流调度系统，用于管理apachehadoop作业。
...
oozie协调器作业是由时间（频率）和数据可用性触发的循环oozie工作流作业。
t5fffqht3#
根据你正在解释的情况，我可以提出以下建议solution:- 1.the 数据监控的过程，即监控档案登陆的目录，应该由一个单独的过程来完成。这个独立的进程可以使用一些元数据表，比如mysql，根据对目录的监视来放置状态条目。元数据条目还可以检查重复性。2现在，基于元数据条目，一个单独的进程可以处理map reduce作业触发部分。某些状态可以签入元数据以触发作业。
whlutmcx4#
最简单的方法可能是执行多输入作业，在运行作业之前读取文件的目录，并将这些文件而不是目录传递给作业（然后在作业完成后删除列表中的文件）。