用于大型数据集(10 亿条记录)的 Hadoop UniqValueCount 映射和聚合缩减器

Question

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

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

我正在使用一种相当简单的算法来提取这些唯一值:我正在解析 map 中的 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)，作业也会因“内存不足”错误而失败。

但我能够使用带有 Set 数据结构(哈希表)的 Python 代码在一个 m1.large(一个具有 8 GB 内存的小得多的盒子)上提取唯一值。我对 Hadoop 作业失败感到困惑，因为 4600 万个唯一身份不应该占用那么多内存。

可能出了什么问题？我使用的 UniqValueCount 有误吗？

Answer 1

您可能在随机播放中遇到内存错误，请记住 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

这将有效地减少网络带宽，并且洗牌将更加简单，因为要排序的键会更少。

你可以通过两种方式做到这一点:

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