Spark groupBy OutOfMemory woes

Question

我正在一个相当小的数据集（HDFS 中的 80 个文件，总共几场演出）上做一个简单的 groupBy。我在纱线集群中的 8 台低内存机器上运行 Spark，即类似于以下内容：

spark-submit ... --master yarn-client --num-executors 8 --executor-memory 3000m --executor-cores 1

数据集由长度为 500-2000 的字符串组成。

我正在尝试做一个简单的groupByKey（见下文），但它因java.lang.OutOfMemoryError: GC overhead limit exceeded 异常而失败

val keyvals = sc.newAPIHadoopFile("hdfs://...")
  .map( someobj.produceKeyValTuple )
keyvals.groupByKey().count()

我可以毫无问题地使用reduceByKey 计算组大小，确保我自己的问题不是由单个过大的组引起的，也不是由过多的组引起的：

keyvals.map(s => (s._1, 1)).reduceByKey((a,b) => a+b).collect().foreach(println)
// produces:
//  (key1,139368)
//  (key2,35335)
//  (key3,392744)
//  ...
//  (key13,197941)

我尝试过重新格式化、重新洗牌和增加 groupBy 的并行度：

keyvals.groupByKey(24).count // fails
keyvals.groupByKey(3000).count // fails
keyvals.coalesce(24, true).groupByKey(24).count // fails
keyvals.coalesce(3000, true).groupByKey(3000).count // fails
keyvals.coalesce(24, false).groupByKey(24).count // fails
keyvals.coalesce(3000, false).groupByKey(3000).count // fails

我尝试过使用spark.default.parallelism，并将spark.shuffle.memoryFraction 增加到0.8，同时将spark.storage.memoryFraction 降低到0.1

失败的阶段（计数）将在 3000 个任务 2999 上失败。

我似乎找不到任何表明 groupBy 不应该只是溢出到磁盘而不是把东西保存在内存中的东西，但我就是无法让它正常工作，即使在相当小的数据集上也是如此。这显然不是这种情况，我一定做错了什么，但我不知道从哪里开始调试这个！

Answer 1

Patrick Wendell 阐明了 groupBy 运算符 on the mailing list 的一些细节。外卖信息如下：

在一个分区内，事情会溢出 [...] 这种溢出目前只能跨键发生。目前，密钥内不会发生溢出。 [...] GroupBy 的一个键内溢出可能最终会出现在 Spark 的下一版本 Spark 1.2 中。 [...] 如果目标只是将与每个组关联的所有值写入磁盘，并且与单个组关联的值大于内存中的容量，那么现在无法使用 groupBy 运算符完成。

他进一步提出了一种解决方法：

解决此问题的最佳方法在一定程度上取决于您尝试对下游数据执行的操作。通常方法涉及细分任何非常大的组，例如，将小范围 (1-10) 中的散列值附加到大键。然后，您的下游代码必须处理聚合每个组的部分值。如果您的目标只是将每个组按顺序排列在磁盘上的一个大文件上，您也可以使用散列后缀调用sortByKey。排序功能在 Spark 1.1（预发布）中被外部化。