Hadoop UniqValueCount Map and Aggregate Reducer for Large Dataset（10 亿条记录）

Question

我有一个包含大约 10 亿个数据点的数据集。我想从中提取大约 4600 万个唯一数据点。

我想使用 Hadoop 来提取唯一值，但在 Hadoop 上不断出现“内存不足”和 Java 堆大小错误 - 同时，我可以使用 Python 在单个机器上相当轻松地运行它设置（哈希表，如果你愿意的话。）

我正在使用一种相当简单的算法来提取这些唯一值：我正在解析我的地图中的 10 亿行并输出如下所示的行：

UniqValueCount:I    a
UniqValueCount:I    a
UniqValueCount:I    b
UniqValueCount:I    c
UniqValueCount:I    c
UniqValueCount:I    d

然后运行“聚合”reducer 得到结果，对于上面的数据集应该是这样的：

I   4

这适用于一小组值，但是当我为 10 亿个数据点（如我所提到的有 4600 万个键）运行它时，作业失败了。

我在 Amazon 的 Elastic Map Reduce 上运行它，即使我使用六个 m2.4xlarge 节点（它们的最大内存节点为每个 68.4 GB），作业也会失败并出现“内存不足”错误。

但我可以使用 Python 代码提取唯一值，该代码在一个 m1.large（一个具有 8 GB 内存的小盒子）上具有 Set 数据结构（哈希表）。我很困惑 Hadoop 作业失败了，因为 4600 万个唯一值不应该占用那么多内存。

可能出了什么问题？我是不是用错了 UniqValueCount？

Answer 1

您可能会在 shuffle 中遇到内存错误，请记住 Hadoop 在启动 reducer 之前会对键进行排序。大多数应用程序不需要排序本身，但 Hadoop 使用它作为聚合属于某个键的所有值的一种方式。

对于您的示例，您的映射器最终会写入很多次相同的值，而您只关心给定键有多少唯一性。这是你现在正在做的事情：

Mapper output:
I -> a
I -> a
I -> a
I -> a
I -> b
I -> a
I -> b

Reducer input:
I -> [a, a, a, a, b, a, b]

Reducer output:
I -> 2

但是在这种情况下你真的不需要写 5*a 或 2*b，1 次就足够了，因为你只关心唯一性。因此，您可以通过确保只发送每个值一次来直接减少大量开销，而不是计算 reducer 中的唯一值：

Mapper output:
I -> a
I -> b

Reducer input:
I -> [a, b]

Reducer output:
I -> 2

这将有效地减少网络带宽，并且由于要排序的键更少，因此 shuffle 会简单得多。

您可以通过两种方式做到这一点：

• 在您的作业中添加一个组合器，该组合器将在映射器之后但在缩减器之前运行，并且只会在发送到缩减器之前保留唯一性。
• 修改您的映射器以保留您已发送的映射，如果您之前已经发送过此映射，则不要发送。