Spark：处理性能密集型命令，如 collect()、groupByKey()、reduceByKey()

Question

我知道collect() 等一些 Spark 操作会导致性能问题。

已在documentation引用

要打印驱动程序上的所有元素，可以使用 collect() 方法首先将 RDD 带到驱动程序节点：rdd.collect().foreach(println) 。这可能会导致驱动程序内存不足，但是，

因为collect() 将整个RDD 提取到一台机器上；如果只需要打印 RDD 的几个元素，更安全的方法是使用take():rdd.take(100).foreach(println)。

还有一个相关的 SE 问题：Spark runs out of memory when grouping by key

我知道groupByKey(), reduceByKey() 如果并行设置不正确，可能会导致内存不足。

其他Transformations和Action命令我没有得到足够的证据，必须谨慎使用。

这三个是唯一需要处理的命令吗？我对以下命令也有疑问

1. aggregateByKey()
2. sortByKey()
3. persist() / cache()

如果您提供有关密集命令（全局跨分区而不是单个分区或低性能命令）的信息，那就太好了，必须通过更好的保护来解决这些问题。

Answer 1

你必须考虑三种类型的操作：

• 仅使用mapPartitions(WithIndex) 实现的转换，如filter、map、flatMap 等。通常它将是最安全的组。您可能遇到的最大问题可能是大量溢出到磁盘。
• 需要随机播放的转换。它包括明显的嫌疑人，如combineByKey（groupByKey、reduceByKey、aggregateByKey）或join 的不同变体，以及不太明显的sortBy、distinct 或repartition。如果没有上下文（数据分布、精确的归约函数、分区器、资源），很难判断特定转换是否会出现问题。有两个主要因素：
  • 网络流量和磁盘 IO - 任何不在内存中执行的操作都将是at least an order of magnitude slower。
  • 偏斜的数据分布 - 如果分布高度偏斜，shuffle 可能会失败，或者后续操作可能会因资源分配不理想而受到影响

• 需要将数据传入和传出驱动程序的操作。通常，它涵盖collect 或take 之类的操作，并从本地 (parallelize) 创建分布式数据结构。

  此类别的其他成员是broadcasts（包括自动广播加入）和accumulators。总成本当然取决于特定操作和数据量。

虽然其中一些操作可能很昂贵，但没有一个操作本身特别糟糕（包括demonized groupByKey）。显然最好避免网络流量或额外的磁盘 IO，但实际上在任何复杂的应用程序中都无法避免它。

关于缓存，您可能会发现Spark: Why do i have to explicitly tell what to cache? 很有用。