hadoop - MapReduce 连续执行

Question

我正在使用 Hadoop 将现有的时间序列数据库系统转换为 MapReduce 模型。数据库系统兼具历史和实时处理能力。到目前为止，我已经能够将批处理功能转换为 Hadoop。

不幸的是，当谈到实时处理时，我发现与 MapReduce 模型存在一些概念上的不一致。

我可以编写自己的 Hadoop InputFormat 接口(interface)实现，它将连续为映射器提供新数据，以便映射器可以处理并连续发出数据。但是，因为在所有映射器都完成执行之前不会调用任何 reduce() 方法，所以我的计算必然会卡在映射阶段。

我看过一些提到 mapred.reduce.slowstart.completed.maps 的帖子，但是，据我所知，这只控制 reducer 何时开始将数据拉到它们的本地目的地(洗牌)- - 实际的 reduce 方法只有在所有映射器都完成执行后才会被调用。

当然，可以选择通过使用独立 MR 作业的连续流在小时间间隔内处理小批量数据来模拟连续执行，但这会引入额外的延迟，这在我的情况下是 Not Acceptable 。

我也考虑过使用 Storm或 S4 ，但在进一步讨论之前，我需要确保这超出了 Hadoop 的范围。

总而言之，人们似乎已经能够开发实时 Hadoop 应用程序(例如 Impala)或构建在 Hadoop 之上的实时处理解决方案。问题是如何？

Answer 1

你是对的，如果 InputFormat/mappers 连续发出数据，reduce 方法将永远不会被调用。这样做的原因是 reduce 方法必须迭代键的所有值，并且在 map 阶段完成之前，整组值是未知的，因为要给那个 reduce 的值方法可能随时来自任何映射器。

reduce 方法通过迭代器访问值，因此从 API 的角度来看，理论上可以预先调用 reduce() 并使其在迭代器上永久阻塞运行，直到值可用为止。 Hadoop 没有此功能的原因是它需要在内存中保留每个键的上下文，这对于大型数据集的批处理没有意义。

在 Hadoop MapReduce 编程模型中实现对数据流的连续分析的一种方法是提交连续的小型 MR 作业流，每个作业分析一个数据 block 。在这种情况下处理额外延迟的方法是使用许多可用的 Hadoop 加速器之一(免责声明:我在一家公司工作，ScaleOut Software，该公司提供这样的加速器:ScaleOut hServer — 在免费社区版中可用)。 ScaleOut hServer是内存中的 MapReduce 引擎，可以在几毫秒内运行 MR 作业。它在作业之间重用 JVM，因此与 Hadoop 相比，作业启动延迟最小。这非常适合在数据 block 上连续运行 MapReduce 作业，因为它针对适合内存的数据集的实时性能进行了优化。

上述所有情况的一个异常(exception)是如果分析不需要 reduce 阶段(即，reducer 的数量设置为 0):如果算法可以表示为 map-only，则可以连续完成一个 Hadoop 批处理作业。