乍一看编写 Spark 程序似乎很容易，如果了解数据的要求和足够的知识，那就是读取数据集、根据键进行连接、然后做一些转换，最后便准备好了一个交付的新数据集！如果您正在使用的数据非常小并且整个转换过程只需要几分钟，则可能会出现这种情况。然而从定义上看，Spark 具有高度可扩展性，当数据量越来越大时，事情可能会失控。相反如果转换可能需要数小时，而需要尝试尽可能缩短执行时间，则可以尝试扩展集群并为此支付更多费用，或者可以尝试优化代码。知道如何解决性能问题并能够优化 Spark 代码（或者只是让它运行得更快）是一项非常有价值的技能，它可以区分业余爱好者和精通 Spark 开发人员。在这篇文章中，我将介绍最常见的性能问题以及如何尝试解决它们。考虑到将这篇文章用作何时需要优化 Spark 代码的参考指南。由于优化是一个高级主题，因此深入了解 Spark 在后台的工作原理以及对 Spark UI 的一些了解肯定有助于充分利用这些内容。

1. 对代码进行基准测试

在继续之前，我们需要找到一种方法来对 Spark 代码进行基准测试。通过这种方式我们可以实际衡量我们是否对引入的更改进行了任何改进。这里的想法是测量 Spark 执行需要完整计算转换逻辑的操作所需的时间。在这种情况下， df.show() 、 df.head() 或 display(df) 立即被排除在外，因为通过此操作，Spark 将尝试计算尽可能少的数据，写入数据是一种选择，但在这个探索阶段将输出持久化到磁盘没有意义，相关的基准测试命令是：

df.count // 动作df.foreach // “什么都不做”的 lambda 函数的操作df.write.format(“noop”) // 伪造一个写操作

当然如果此阶段的数据集太大，请先获取其中的相关数据子集，然后再尝试整个数据集。

2. 最常见的性能问题

Spark 中的性能问题主要与以下主题相关：

• Skew：在数据分区大小不平衡的情况下会发生什么。

• Spill：由于内存不足而将临时文件写入磁盘。

• Shuffle：由于广泛的转换而在Executor之间移动数据。

• Storage：数据在磁盘上的存储方式实际上很重要。

• Serialization：跨集群的代码分布（UDF 是邪恶的）。

尽管找到问题的根源可能非常困难，因为一个问题实际上可能导致另一个问题：倾斜会导致溢出，存储问题会导致过度Shuffle，错误地解决Shuffle的方法会增加偏斜，有时其中很多原因可能同时存在！

2.1 Skew

在 Spark 中，数据通常在均匀分布在集群Executor中的分区中读取。只要我们对数据进行转换，一些分区最终可能会比其他分区拥有更多的记录，这种数据分区大小的不平衡就是我们所说的 Skew。少量的偏斜不是问题，可以忽略。虽然数据中的大偏差可能会导致分区太大，无法放入Worker的 RAM。这会导致溢出，有时甚至会导致难以诊断的 OOM 错误。

2.1.1. 检测Skew

我们要深入Spark UI，查看join作业，注意以下几点：

• 检查 Event Timeline：当我们看到非常不平衡的任务时，我们认为它是不健康的，这意味着某些任务计算的数据比其他任务多或少，换句话说分区是不平衡的，因为正常情况下它们都应该具有差不多相同的持续时间。

• 检查 Summary Metrics：注意 Shuffle Read 的值（min/25th p./median/75th p./max），如果第 75p（意味着大多数任务）的数字很高，而其他的非常低，则意味着大多数的任务正在Shuffle大量数据，因为所需的数据在其他分区中。如果这些 Shuffle Read 值彼此相似，则表明所有任务都在 Shuffle 大约相同数量的数据，因此分区平衡良好。

• 检查 Aggregated Metrics by Executor：如果 Spill 值（内存/磁盘）非常高，这是由于 skew 引起的 shuffle 造成的，由于内存不足，太大的分区可能需要将数据存储在磁盘中的临时文件中。

• 检查数据：一旦知道可能面临Skew问题，就可以通过检查数据来验证猜想，通过对”group by”或”join key”的每个类别执行记录计数，并检查某些类别中的记录是否比其他类别多得多，如果存在很大的不平衡，最大的类别（分区）将花费更长的时间，从而导致大量的Shuffle和Spill。

2.1.2. 我们可以做些什么来减轻Skew？

如果出现 OOM 问题，则可能会想增加集群Worker的内存，这可能会解决问题，但不会彻底解决问题，并且可能会将问题推到以后。如果检测到Skew，首先要解决不均匀分区的问题，为此我们可以：

• 如果在 Spark 3 中，启用 AQE（自适应查询执行）。

• 如果在 Databricks 中，特定的倾斜提示（倾斜连接优化）。

• 否则应用 Key Salting。用随机数对倾斜的列进行加盐以在每个分区之间创建更好的分布，但代价是额外的处理。

请记住，在应用这些解决方案时，作业持续时间甚至会增加，但请记住比持续时间更重要的是减轻Skew可以消除潜在的 OOM 错误。

2.1.3. 如何为倾斜连接实施 Key Salting？

Key Salting 的一般思想在于通过增加连接键的数量来减少分区的数量。为此我们可以创建一个新列，其值基于连接键加上一个范围内的随机数，这需要应用于由受影响的键连接的所有涉及的DataFrame。在此之后我们可以使用新Key执行连接操作，最终会得到更多更小的分区和任务。另外随机数范围必须经过实验选择，如果我们选择一个很大的数字，我们最终会得到太多的小分区，如果太少，我们会继续出现倾斜问题。

2.2 Spill

在 Spark 中 Spill 被定义为在作业期间将数据从内存移动到磁盘的行为，反之亦然。这是 Spark 的一种防御措施，目的是在分区太大而无法放入内存时释放 Worker 的内存并避免 OOM 错误。这样Spark 作业就可以通过溢出数据，增加读写开销的计算时间代价而使得作业成功运行完成。有几种方法可以解决这个问题：

• 将 spark.sql.filesMaxPartitionBytes设置得太高（默认为 128MB），从而摄取可能无法放入内存的大分区。

• 即使是一个小数组的explode() 操作。每个分区最终将包含与数组中的项目一样多的行，因此生成的分区大小可能不适合内存。

• 两个表的 join()或 crossJoin()。

通过倾斜键聚合结果。正如我们之前看到的，拥有不平衡的数据集会导致比其他数据集更大的分区，并且在某些情况下这个更大的分区可能不适合内存。

2.2.1 检测Spill

在 Spark UI 中，这表示为：

• 溢出（内存）：溢出分区的内存大小

• 溢出（磁盘）：溢出分区的磁盘大小（由于压缩，总是小于内存）。

此值仅在单个阶段的详细信息页面（summary metrics, aggregated metrics by executor, task表）或 SQL 查询详细信息中表示。这使得它很难识别，因为必须手动搜索，请注意如果没有溢出将找不到这些值。手动搜索溢出的替代方法是实现一个 SpillListener 以在阶段溢出时自动跟踪，不幸的是这仅在 Scala 中可用。

2.2.2 我们可以做些什么来减少Spill？

一旦我们发现我们的阶段正在Spill，我们有一些选择进行缓解：

• 检查Spill是否是由数据倾斜引起的，在这种情况下首先缓解该问题。

• 如果可能，增加集群Worker的内存。通过这种方式更大的分区将适合内存，Spark 不需要向磁盘写入那么多。

• 通过增加分区数来减小每个分区的大小。我们可以通过调整spark.sql.shuffle.partitions 和 spark.sql.maxPartitionBytes或显式重新分区来做到这一点。

减少Spill并不总是值得的，所以首先需要检查它是否有意义。

2.3 Shuffle

Shuffle 是 Spark 的一种自然操作，其是Join、groupBy或Sort等广泛转换的副作用，在这些情况下需要对数据进行Shuffle，以便将具有相同键的记录分组到相同的分区下，以便以后能够通过这些键执行聚合。当发生广泛的转换时，分区被写入磁盘，因此下一阶段的Executor可以读取数据并继续工作。因为数据需要在Worker之间移动，所以这种行为会导致大量的网络 IO。正如我之前提到的，Shuffle是不可避免的，只需将重点放在最昂贵的操作上，同时请注意针对其他问题（如倾斜、溢出或小文件问题）通常更有效。

2.3.1 我们可以做些什么来减轻Shuffle的影响？

Shuffle 的最大痛点是需要跨集群移动的数据量。为了减少这个问题，我们可以应用以下策略：

• 使用更少和更大的Worker来减少网络流量。在更少的Worker中拥有相同数量的Executors（CPU），将减少需要通过网络传输到其他Worker的数据量。

• 减少被Shuffle的数据量。有时对最终结果不需要的数据执行这些广泛的转换，从而不必要地增加了成本。因此我们应该在执行宽转换之前过滤掉那些不需要的列和行。

• 非规范化数据集。如果数据用户经常执行导致昂贵Shuffle的查询，则该查询的输出可以持久化在数据湖中并支持直接查询。

• 广播较小的表。当连接中涉及的一个表比其他表小得多时，我们可以将它广播给所有Executors，Spark 将能够将它连接到其他表分区而无需对其进行Shuffle，这称为 BroadcastHashJoi，可以使用 .broadcast(df)应用，但是默认表大小阈值为 10MB，我们可以通过调整spark.sql.autoBroadcastJoinThreshold来增加阈值，但不要太大，因为它会使Driver承受压力并且可能导致 OOM 错误，而且这种方式增加了Driver和Executor之间的IO，在很多空分区的情况下效果不佳，并且需要Driver和Executor有足够的内存。

• 分桶数据集。对于Join，数据可以预先打乱并按桶存储，并且可以选择按桶排序。在处理 TB 大小、经常Join表且未应用过滤器时是值得的，这要求所涉及的所有表都通过具有相同数量的桶（通常每个Core一个）的键进行分桶，但是生产和维护这种方法的成本非常高，而且必须是合理的。

2.4 Storage

在数据湖中摄取初始数据的方式可能会导致通常与以下相关的问题：小文件、目录扫描或Schema。

2.4.1 小文件

我们认为那些比底层文件系统的默认块大小（128MB）小得多的小文件。将数据集划分为太多的小文件意味着打开和关闭文件的总时间更长，并且会导致性能非常差，它通常与摄取数据的高开销有关，或者是 Spark 作业的结果。我们可以使用 Spark UI 查看 SQL 选项卡下读取了多少文件并检查读取操作，这个问题可以通过以下方式缓解：

• 将现有的小文件压缩为相当于块大小或使用的有效分区大小的较大文件。

• 制作摄取工具来编写更大的文件。

当作为 Spark 作业的结果生成时，Spark 对数据的分区方式超出了其大小所需的程度，并且在写入时会反映出来。我们可以通过以下方式缓解这种情况：

• 更改默认分区数。调整 spark.sql.shuffle.partitions（默认为 200）。

• 在写入之前显式重新分区数据。应用 repartition() 或 coalesce() 函数来减少分区数量，或者在启用 AQE 的 Spark 3.0+ 的情况下，将 spark.sql.adaptive.coalescePartitions.enabled设置为 true。

2.4.2 目录扫描

某些数据集的目录过多（由于数据分区）会导致扫描时出现性能问题，没有太多数据的分区数据集也会导致小文件问题，我们可以通过关注SQL选项卡读取操作下的扫描时间来检测到这一点，也可以通过以下方式缓解这个问题：

• 以更智能的方式对存储的数据进行分区。

• 将数据集注册为表。这样做时，诸如在哪里可以找到属于该数据集的文件之类的元数据存储在 Hive Metastore 中，因此不再需要扫描目录。但是我们第一次注册表时需要一些时间通过首先扫描目录来检索元数据。

2.4.3 Schema

推断Schema需要完整读取文件以确定每列的数据类型，这涉及打开和扫描文件的时间。读取 parquet 文件需要一次性读取Schema，因为Schema包含在文件本身中，另一方面如果有成百上千个零件文件，则支持Schema演变可能会很昂贵，每个Schema都必须被读入然后进行合并，这可能真的很昂贵，可以通过spark.sql.parquet.mergeSchema启用Schema合并。我们可以通过以下方式缓解模式问题：

• 每次都显示提供Schema。

• 将数据集注册为表。这样Schema也将存储在 Hive Metastore 中。

• 使用数据湖格式，自动合并Schema以支持Schema演变。

2.5 序列化

当我们需要应用非原生 API 转换（称为 UDF）时，就会发生这种情况。这意味着将数据序列化为 JVM 对象，以便在 Spark 外部进行修改。这在 Python 中的影响要严重得多，因为 JVM 对象不是 Python 的原生对象，因此需要先对它们进行转换，在其上执行代码，然后将结果序列化回 JVM 对象，从而造成很大的开销。另一方面，Scala 不需要这部分转化，因为它是 JVM 原生语言。在任何情况下，UDF 都是 Catalyst Optimizer 的障碍，因为它无法在应用 UDF 之前和之后连接代码，因为它不可能知道 UDF 在做什么，以及如何优化整体作业执行。我们可以通过以下方式缓解序列化问题：

• 尽可能避免使用 UDF、Pandas UDF 或类型转换，而是使用原生 Spark 高阶函数。

• 如果没有其他选择：

• Python：在”标准”Python 代码上使用 Pandas UDF。Pandas UDF 使用 PyArrow 序列化成批的记录（被视为 Pandas 系列或DataFrame），以便稍后将 UDF 应用于 Python 中的每条记录。另一方面，常规Python UDF 会单独序列化每条记录，并在其上执行 UDF。

• Scala：在”标准”Scala 代码上使用类型转换。

如果数据转换需要应用许多 UDF，请考虑使用 Scala 编程语言。