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

Spark : Tackle performance intensive commands like collect(), groupByKey(), reduceByKey()

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

已在documentation

中引用

To print all elements on the driver, one can use the collect() method to first bring the RDD to the driver node thus:rdd.collect().foreach(println). This can cause the driver to run out of memory, though,

因为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()

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

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

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

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

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

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

关于缓存，您可能会发现 很有用。