我有一个 JavaPairRDD<Integer, Integer[]>我想在其上执行 groupByKey行动。

groupByKey行动给我一个:

org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle

如果我没记错的话，这实际上是一个 OutOfMemory 错误。这只发生在大数据集中(在我的例子中，Web UI 中显示的“Shuffle Write”约为 96GB)。

我已经设置:

spark.serializer org.apache.spark.serializer.KryoSerializer

在$SPARK_HOME/conf/spark-defaults.conf ，但我不确定 Kryo 是否用于序列化我的 JavaPairRDD。

除了设置此 conf 参数之外，我还应该做些什么来使用 Kryo 来序列化我的 RDD？我可以在 serialization instructions 中看到那:

Spark automatically includes Kryo serializers for the many commonly-used core Scala classes covered in the AllScalaRegistrar from the Twitter chill library.

还有:

Since Spark 2.0.0, we internally use Kryo serializer when shuffling RDDs with simple types, arrays of simple types, or string type.

我还注意到，当我将 spark.serializer 设置为 Kryo 时，Web UI 中的 Shuffle Write 从 ~96GB(使用默认序列化器)增加到 243GB!

编辑: 在评论中，有人问我程序的逻辑，以防 groupByKey 可以替换为 reduceByKey。我不认为这是可能的，但无论如何:

• 输入的形式为:

  • key:索引桶id，
  • 值:此存储桶中实体 ID 的整数数组

• 随机写入操作产生以下形式的对:

  • 实体编号
  • 同一桶中所有实体 ID 的整数数组(称它们为邻居)

• groupByKey操作收集每个实体的所有邻居数组，有些可能出现不止一次(在许多桶中)。

• groupByKey之后在操作中，我为每个桶保留一个权重(基于它包含的负实体 ID 的数量)，并为每个邻居 ID 汇总它所属的桶的权重。

• 我用另一个值(假设它是给定的)标准化每个邻居 ID 的分数，并发出每个实体的前 3 个邻居。

我获得的不同键的数量约为 1000 万(大约 500 万个正实体 ID 和 500 万个负实体 ID)。

EDIT2:我尝试分别使用 Hadoop 的 Writables(VIntWritable 和 VIntArrayWritable 扩展 ArrayWritable)而不是 Integer 和 Integer[]，但是 shuffle 大小仍然大于默认的 JavaSerializer。

然后我增加了spark.shuffle.memoryFraction从 0.2 到 0.4(即使在 2.1.0 版中已弃用，也没有说明应该使用什么来代替)并启用了 offHeap 内存，并且 shuffle 大小减少了约 20GB。即使这符合标题的要求，我还是更喜欢一种算法性更强的解决方案，或者包含更好压缩的解决方案。

最佳答案

简答:使用fastutil并可能增加 spark.shuffle.memoryFraction。

更多详情: 这个 RDD 的问题是 Java 需要存储 Object 引用，这比原始类型消耗更多的空间。在此示例中，我需要存储 Integer，而不是 int 值。 Java Integer 占用 16 个字节，而原始 Java int 占用 4 个字节。另一方面，Scala 的 Int 类型是 32 位(4 字节)类型，就像 Java 的 int 一样，这就是为什么使用 Scala 的人可能没有遇到过一些问题相似的。

除了将 spark.shuffle.memoryFraction 增加到 0.4 之外，另一个不错的解决方案是使用 fastutil library , 如 Spark's tuning documentation 中所建议:

The first way to reduce memory consumption is to avoid the Java features that add overhead, such as pointer-based data structures and wrapper objects. There are several ways to do this: Design your data structures to prefer arrays of objects, and primitive types, instead of the standard Java or Scala collection classes (e.g. HashMap). The fastutil library provides convenient collection classes for primitive types that are compatible with the Java standard library.

这可以将我的 RDD 对的 int 数组中的每个元素存储为 int 类型(即，对数组的每个元素使用 4 个字节而不是 16 个字节)。就我而言，我使用了 IntArrayList而不是 Integer[]。这使得 shuffle 大小显着下降，并允许我的程序在集群中运行。我还在代码的其他部分使用了这个库，在那里我制作了一些临时的 Map 结构。总的来说，通过将 spark.shuffle.memoryFraction 增加到 0.4 并使用 fastutil 库，shuffle 大小从 96GB 下降到 50GB (!) 使用默认的 Java 序列化程序(不是 Kryo)。

备选方案:我还尝试对 rdd 对的每个 int 数组进行排序，并使用 Hadoop 的 VIntArrayWritable 类型存储增量(较小的数字比较大的数字使用较少的空间)，但这也需要注册 VIntWritable和 Kryo 中的 VIntArrayWritable，毕竟它没有节省任何空间。总的来说，我认为 Kryo 只会让事情运行得更快，但不会减少所需的空间，但我仍然不确定。

我还没有将这个答案标记为已接受，因为其他人可能有更好的主意，而且因为我毕竟没有使用 Kryo，正如我的 OP 所要求的那样。我希望阅读它，能帮助其他人解决同样的问题。如果我设法进一步减小随机播放大小，我将更新此答案。

关于java - spark - 如何减少 JavaPairRDD<Integer, Integer[]> 的洗牌大小？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/42733558/