1. Spark 性能优化到底在优化什么1.1 从一次“跑不动”的作业说起我第一次真正意识到 Spark 性能优化的重要性是在处理一个看起来并不复杂的日志聚合任务时。数据量大概 800GB集群规模 20 个节点每个节点 64GB 内存、16 核。按道理说这个配置跑一个 group by 加 count 的作业应该很轻松但实际跑起来之后最后一个 reduce 阶段卡了将近 4 个小时中间还出现了两次 executor OOM。当时我盯着 Spark UI 上那个迟迟不动的进度条心里只有一个念头这玩意儿到底在干什么。后来我把 Spark UI 的各个页面翻了个遍才发现问题出在数据倾斜上——某个 key 对应的记录数占了全量的 40%所有分配到那个 key 的 task 都在一台机器上死磕其他 task 早就跑完了整个作业被这一个 task 拖死。这件事让我明白一个道理Spark 性能优化不是简单地调几个参数而是要理解作业在集群上到底是怎么执行的然后找到那个真正的瓶颈。Spark 性能优化的核心目标其实就三个让 CPU 别闲着、让内存别爆掉、让网络别堵死。听起来像废话但绝大多数优化手段最终都是围绕这三个目标展开的。CPU 闲着通常是因为数据倾斜或者分区不合理内存爆掉多半是缓存策略或者 shuffle 数据量的问题网络堵死则往往跟 shuffle 和广播变量有关。1.2 优化的三个层次代码、配置、架构我把 Spark 性能优化分成三个层次来看这样在排查问题的时候思路会更清晰。第一层是代码层面的优化。这是最容易被忽视但收益往往最大的地方。很多性能问题其实在写代码的时候就已经埋下了比如用了 groupByKey 而不是 reduceByKey比如在 map 里做了不必要的重复计算比如没有及时 filter 掉不需要的数据。代码层面的优化不需要改集群配置也不需要加机器纯粹靠改写逻辑就能拿到几倍甚至几十倍的提升。第二层是配置层面的调优。这包括 executor 的内存和核数设置、并行度调整、shuffle 参数、序列化方式等等。配置调优的特点是见效快但需要反复试验因为不同的作业、不同的数据量、不同的集群规模最优配置都不一样。我见过太多人直接抄网上的“最优配置”结果跑起来还不如默认参数。第三层是架构层面的设计。这涉及到数据存储格式的选择、分区策略的设计、是否使用缓存、是否广播小表等等。架构层面的优化往往在作业开发之前就要考虑清楚一旦作业上线之后再改成本会很高。这三个层次不是孤立的实际优化的时候往往需要结合起来看。但排查问题的顺序我建议是先看代码有没有明显的问题再看配置是否合理最后才考虑架构是否需要调整。因为代码层面的改动成本最低架构层面的改动成本最高。1.3 为什么 Spark UI 是你最好的朋友如果你只能记住一件事那就是优化之前先看 Spark UI。Spark UI 提供了作业执行的几乎所有关键信息包括每个 stage 的耗时、每个 task 的数据量、shuffle 的读写量、executor 的内存使用情况等等。我通常按这个顺序看 Spark UIJobs 页面先看哪个 job 耗时最长哪个 stage 卡住了。Stages 页面看每个 stage 的 task 耗时分布如果 max 和 median 差距很大基本可以确定是数据倾斜。Storage 页面看缓存的使用情况有没有缓存了但没用到或者缓存太大导致内存压力。Executors 页面看每个 executor 的 GC 时间、shuffle 读写量GC 时间占比超过 10% 就要注意了。SQL 页面如果是 DataFrame/Dataset API看物理执行计划有没有出现 CartesianProduct、BroadcastNestedLoopJoin 这种危险的算子。提示Spark UI 的 Stages 页面里task 的 Duration 分布图非常有用。如果看到某个 task 的耗时远远超过其他 task不用怀疑就是数据倾斜。2. 代码层面的优化从源头减少计算量2.1 算子选择reduceByKey 和 groupByKey 差在哪里很多人写 Spark 代码的时候习惯性地用 groupByKey 来做聚合觉得反正结果一样。但这两个算子在底层执行上差别巨大。groupByKey 会把所有相同 key 的 value 都 shuffle 到同一个节点上然后在内存里形成一个 Iterable。这意味着如果某个 key 有 100 万条记录这 100 万条记录都要通过网络传输到同一个节点并且在内存里同时存在。而 reduceByKey 会先在 map 端做本地聚合combine只把聚合后的结果 shuffle 到 reduce 端。举个例子假设有 10 亿条记录key 的数量是 100 万每个 key 平均有 1000 条记录。用 groupByKey 的话shuffle 的数据量是 10 亿条用 reduceByKey 的话map 端先聚合shuffle 的数据量只有 100 万条。差了三个数量级。类似的算子选择还有不推荐推荐原因groupByKeyreduceByKeymap 端预聚合减少 shufflereduceByKeyaggregateByKey更灵活可以控制 map 端和 reduce 端的聚合逻辑joinbroadcast join小表广播避免 shufflecollecttake / foreachPartition避免把大量数据拉到 drivercountByKeyreduceByKey count避免全量数据拉到 driver2.2 尽早过滤把不需要的数据挡在门外我见过一个作业从 Hive 表里读了一天的数据然后 filter 出某个特定渠道的记录再做聚合。问题是这个 filter 是在读完之后才做的也就是说 Spark 先把一天的数据全部读进来然后在内存里过滤。正确的做法是在读取的时候就做过滤让数据源只返回需要的数据。如果是 Parquet 或者 ORC 格式Spark 支持谓词下推predicate pushdownfilter 条件会直接下推到存储层只读取满足条件的数据块。如果是 Hive 表还可以利用分区裁剪partition pruning只读取特定分区的数据。# 不推荐先读全量再过滤 df spark.sql(SELECT * FROM log_table WHERE dt 2024-01-01) df_filtered df.filter(df.channel specific_channel) # 推荐在 SQL 里就过滤掉 df spark.sql( SELECT * FROM log_table WHERE dt 2024-01-01 AND channel specific_channel )这个改动看起来很小但效果可能非常明显。如果特定渠道的数据只占全量的 5%那么读取的数据量就减少了 95%后续所有的计算、shuffle、内存占用都会相应减少。2.3 避免重复计算cache 的正确用法Spark 是惰性求值的这意味着你写了一个 DataFrame它并不会立即计算而是在遇到 action 操作比如 count、show、write的时候才真正执行。如果同一个 DataFrame 被多个 action 使用那么每次 action 都会重新计算一遍。# 不推荐df 被计算了两次 df spark.sql(SELECT * FROM large_table WHERE condition x) count df.count() # 第一次计算 df.write.parquet(/output) # 第二次计算 # 推荐cache 之后只计算一次 df spark.sql(SELECT * FROM large_table WHERE condition x) df.cache() count df.count() # 第一次计算结果缓存 df.write.parquet(/output) # 直接使用缓存但 cache 不是万能的用不好反而会拖慢作业。cache 会把数据放在内存里或者内存加磁盘如果缓存的数据量太大会导致内存压力增加GC 时间变长甚至 OOM。所以 cache 的原则是只缓存会被多次使用、且数据量可控的 DataFrame。注意cache 之后记得在不需要的时候 unpersist释放内存。特别是在循环里 cache 的情况如果不及时释放内存会很快被撑爆。2.4 广播小表让 join 不再 shuffle两个表做 join 的时候默认情况下 Spark 会把两个表的数据都 shuffle 到相同的节点上然后做 join。这个过程涉及大量的网络传输如果其中一个表很小这个 shuffle 就很不划算。广播 joinbroadcast join的思路是把小表复制到每个 executor 上然后大表不需要 shuffle直接在本地和小表做 join。这样就避免了 shuffle 的开销。from pyspark.sql.functions import broadcast # 小表大约 10MB大表 1TB result large_df.join(broadcast(small_df), onkey, howinner)广播 join 的阈值默认是 10MB可以通过spark.sql.autoBroadcastJoinThreshold参数调整。但要注意广播的表不能太大否则每个 executor 都要复制一份网络传输和内存压力都会很大。一般来说广播表的大小不要超过 100MB。2.5 用 DataFrame API 而不是 RDD APISpark 的 DataFrame API 比 RDD API 快很多原因在于 DataFrame 有 schema 信息Spark 的 Catalyst 优化器可以对执行计划做各种优化比如谓词下推、列裁剪、常量折叠等等。而 RDD API 是黑盒Spark 不知道你在做什么只能按你写的逻辑一步步执行。# RDD APISpark 不知道你在做什么 rdd sc.textFile(/data/log.txt) result rdd.map(lambda line: line.split(,)) \ .filter(lambda fields: fields[2] error) \ .map(lambda fields: (fields[0], 1)) \ .reduceByKey(lambda a, b: a b) # DataFrame APISpark 可以优化执行计划 df spark.read.csv(/data/log.txt) result df.filter(df._c2 error) \ .groupBy(_c0) \ .count()DataFrame API 不仅写起来更简洁而且执行效率通常更高。如果必须用 RDD API也尽量用 DataFrame 和 RDD 混合的方式只在必要的地方用 RDD。3. 配置层面的调优让集群跑在最佳状态3.1 executor 内存和核数怎么设executor 的内存和核数设置是 Spark 调优里最基础但也最容易搞错的部分。我见过很多人直接把 executor 内存设成 32GB 甚至 64GB觉得内存越大越好。但实际上executor 内存太大反而会带来问题。首先executor 内存太大会导致 GC 时间变长。JVM 的 GC 在堆内存很大的时候一次 full GC 可能要几十秒甚至几分钟这段时间整个 executor 是停止工作的。其次内存太大的 executor 一旦 OOM损失也更大。我通常的建议是executor 内存8GB 到 16GB 之间比较合适。如果数据量特别大可以适当增加到 24GB但一般不超过 32GB。executor 核数4 到 6 个核比较合适。核数太多会导致每个核分到的内存减少容易 OOM核数太少则无法充分利用 CPU。每个 executor 的内存 核数 × 每个核的内存。比如 5 个核每个核 3GB那么 executor 内存就是 15GB。还有一个重要的参数是spark.executor.memoryOverhead这是 executor 的堆外内存默认是 executor 内存的 10%。如果作业里有大量的 Python 调用或者 JNI 调用这个值需要调大。# 一个典型的配置示例 spark-submit \ --executor-memory 12g \ --executor-cores 4 \ --num-executors 20 \ --conf spark.executor.memoryOverhead2g \ --conf spark.driver.memory4g \ your_app.py3.2 并行度task 数量决定一切并行度是 Spark 性能优化里最容易被忽视但影响巨大的参数。并行度决定了有多少个 task 可以同时执行如果并行度太低集群的资源就浪费了如果并行度太高task 的调度开销又会变大。并行度的设置原则是task 数量应该是集群总核数的 2 到 3 倍。比如集群有 100 个核那么并行度设置在 200 到 300 之间比较合适。并行度可以通过以下几个参数控制spark.default.parallelismRDD 操作的默认并行度通常设置为集群总核数的 2 到 3 倍。spark.sql.shuffle.partitionsDataFrame shuffle 操作的默认分区数默认是 200。对于大数据量的作业200 往往太小需要调大。spark.sql.files.maxPartitionBytes读取文件时每个分区的最大字节数默认 128MB。# 在代码里设置并行度 spark.conf.set(spark.sql.shuffle.partitions, 600) # 或者对单个 DataFrame 设置 df.repartition(600)提示spark.sql.shuffle.partitions的设置需要根据数据量来定。一个经验法则是每个分区的数据量在 100MB 到 200MB 之间比较合适。如果 shuffle 的数据量是 60GB那么分区数设置在 300 到 600 之间比较合理。3.3 序列化Kryo 比 Java 快在哪里Spark 默认使用 Java 序列化但 Java 序列化的效率比较低序列化后的数据量也大。Kryo 序列化比 Java 序列化快 10 倍左右序列化后的数据量也小很多。启用 Kryo 序列化spark.conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer) spark.conf.set(spark.kryo.registrationRequired, false)如果自定义了类最好注册一下这样 Kryo 不需要在运行时写入类名序列化后的数据量会更小spark.conf.set(spark.kryo.classesToRegister, com.example.MyClass1,com.example.MyClass2)Kryo 序列化对 shuffle 和缓存都有影响。shuffle 的数据量变小了网络传输就少了缓存的数据量变小了内存压力就小了。所以启用 Kryo 序列化通常能带来比较明显的性能提升。3.4 shuffle 调优减少磁盘和网络开销shuffle 是 Spark 作业里最耗时的部分因为它涉及大量的磁盘 I/O 和网络传输。shuffle 调优的目标是减少 shuffle 的数据量和提高 shuffle 的效率。几个关键的 shuffle 参数spark.shuffle.file.buffershuffle 写磁盘时的缓冲区大小默认 32KB。可以增加到 64KB 或 128KB减少磁盘 I/O 次数。spark.reducer.maxSizeInFlightreduce 端每次拉取的数据量默认 48MB。可以增加到 96MB减少网络请求次数。spark.shuffle.io.maxRetriesshuffle 拉取失败的重试次数默认 3。可以增加到 10避免因为网络抖动导致 task 失败。spark.shuffle.io.retryWait重试的等待时间默认 5 秒。可以适当增加。--conf spark.shuffle.file.buffer128k \ --conf spark.reducer.maxSizeInFlight96m \ --conf spark.shuffle.io.maxRetries10 \ --conf spark.shuffle.io.retryWait10s另外如果 shuffle 的数据量特别大可以考虑使用spark.shuffle.consolidateFilestrue对于 Hash Shuffle或者使用 Sort Shuffle默认。Sort Shuffle 会把多个 map task 的输出合并到一个文件里减少文件数量提高 I/O 效率。3.5 内存管理execution 和 storage 怎么分Spark 的内存分为两块execution 内存和 storage 内存。execution 内存用于 shuffle、join、sort 等操作storage 内存用于缓存 RDD 和 DataFrame。默认情况下execution 和 storage 各占 50%但可以根据作业的特点调整。如果作业里有很多缓存操作可以增加 storage 的比例如果作业里有很多 shuffle 和 join可以增加 execution 的比例。--conf spark.memory.fraction0.6 \ --conf spark.memory.storageFraction0.5spark.memory.fraction是 execution 和 storage 总共占 JVM 堆内存的比例默认 0.6。spark.memory.storageFraction是 storage 在 execution 和 storage 总内存中的比例默认 0.5。注意execution 和 storage 之间的内存是可以互相借用的。如果 execution 内存不够可以借用 storage 的内存反之亦然。但如果 storage 的内存被借用后需要归还而 execution 又不够就会导致缓存的数据被驱逐。4. 数据倾斜最让人头疼的问题4.1 怎么判断是不是数据倾斜数据倾斜是 Spark 作业里最常见也最让人头疼的问题。它的表现是大部分 task 很快就跑完了但少数几个 task 跑得特别慢甚至卡死。判断数据倾斜的方法看 Spark UI 的 Stages 页面如果某个 stage 的 task 耗时分布图中max 远远大于 median比如 max 是 1 小时median 是 2 分钟基本可以确定是数据倾斜。看 task 的输入数据量在 Stages 页面点击某个 stage可以看到每个 task 的 Input Size 和 Shuffle Read Size。如果某些 task 的数据量远远大于其他 task就是数据倾斜。看 executor 的负载在 Executors 页面如果某些 executor 的 task 数量远远多于其他 executor也可能是数据倾斜。4.2 数据倾斜的常见原因数据倾斜的根本原因是 key 的分布不均匀。常见的情况包括某些 key 的记录数特别多比如某个热门商品的点击日志某个大 V 的微博转发记录。join 时某个 key 特别多比如用户表和订单表 join某个测试账号的订单特别多。空值或默认值太多比如很多记录的 key 是 null 或者空字符串这些记录都会被分到同一个 task。4.3 解决数据倾斜的几种方案方案一过滤掉异常 key如果倾斜的 key 是异常数据比如 null、空字符串、测试账号直接过滤掉就行。df_filtered df.filter(df.key.isNotNull() (df.key ! ))方案二给倾斜的 key 加随机前缀如果倾斜的 key 是正常数据不能过滤可以给这些 key 加上随机前缀让它们分散到不同的 task 上。import random # 给倾斜的 key 加上 0 到 9 的随机前缀 def add_random_prefix(key): if key skewed_key: return f{random.randint(0, 9)}_{key} return key df_skewed df.rdd.map(lambda row: (add_random_prefix(row.key), row.value)).toDF([key, value])方案三拆分倾斜的 key 单独处理把倾斜的 key 和非倾斜的 key 分开处理倾斜的 key 单独用另一种方式处理。# 非倾斜的 key 正常处理 df_normal df.filter(df.key ! skewed_key) result_normal df_normal.groupBy(key).count() # 倾斜的 key 单独处理 df_skewed df.filter(df.key skewed_key) result_skewed df_skewed.groupBy(key).count() # 合并结果 result result_normal.union(result_skewed)方案四使用 map-side join如果倾斜发生在 join 操作上且其中一个表比较小可以用广播 join 来避免 shuffle。from pyspark.sql.functions import broadcast result large_df.join(broadcast(small_df), onkey, howinner)方案五使用 salting 技术Salting 是给 key 加上随机后缀让原本相同的 key 变成不同的 key从而分散到不同的 task 上。处理完之后再去掉后缀做第二次聚合。# 第一步加盐做第一次聚合 df_salted df.withColumn(salted_key, concat(col(key), lit(_), (rand() * 10).cast(int))) result_partial df_salted.groupBy(salted_key).agg(sum(value).alias(partial_sum)) # 第二步去盐做第二次聚合 result_final result_partial.withColumn(key, split(col(salted_key), _)[0]) \ .groupBy(key).agg(sum(partial_sum).alias(total_sum))4.4 数据倾斜排查速查表现象可能原因解决方案少数 task 耗时特别长key 分布不均匀加随机前缀、saltingjoin 时某个 task 卡住某个 key 的记录特别多广播 join、拆分 key大量 task 数据量为 0分区数太多减少 shuffle partitions所有 task 都很慢并行度太低增加并行度executor OOM单个分区数据量太大增加分区数、减少 executor 内存5. 实操案例一个日志聚合作业的优化过程5.1 原始作业和问题假设有一个日志聚合作业从 Hive 表里读取一天的日志按渠道和小时聚合统计每个渠道每小时的访问量。原始代码如下# 原始作业 df spark.sql(SELECT * FROM log_table WHERE dt 2024-01-01) result df.groupBy(channel, hour).count() result.write.parquet(/output/log_agg)这个作业跑起来之后发现最后一个 stage 卡了将近 3 个小时而且有 executor OOM。5.2 排查过程第一步看 Spark UI打开 Spark UI发现最后一个 stage 有 200 个 task其中 198 个 task 在 5 分钟内跑完了但有 2 个 task 跑了将近 3 个小时。这 2 个 task 的 Shuffle Read Size 分别是 12GB 和 8GB而其他 task 只有 100MB 左右。很明显这是数据倾斜。第二步分析倾斜的 key把倾斜的 key 找出来# 找出记录数最多的 key top_keys df.groupBy(channel, hour).count().orderBy(desc(count)).take(10) for key in top_keys: print(key)发现某个渠道的某个小时有 8 亿条记录而其他渠道每小时只有几百万条。这个渠道是一个默认渠道很多日志没有正确上报渠道信息都被归到了这个默认渠道。第三步确认数据量# 统计总记录数和倾斜 key 的记录数 total df.count() skewed df.filter((df.channel default) (df.hour 12)).count() print(fTotal: {total}, Skewed: {skewed}, Ratio: {skewed/total})发现倾斜的 key 占了总记录数的 35%。5.3 优化方案方案一过滤掉默认渠道如果默认渠道的数据不需要直接过滤掉df_filtered df.filter(df.channel ! default) result df_filtered.groupBy(channel, hour).count()方案二给默认渠道加随机前缀如果默认渠道的数据需要保留可以给它加随机前缀from pyspark.sql.functions import concat, lit, rand, split, sum as spark_sum # 给默认渠道加随机前缀 df_salted df.withColumn( salted_channel, when(df.channel default, concat(lit(default_), (rand() * 20).cast(int))) .otherwise(df.channel) ) # 第一次聚合 result_partial df_salted.groupBy(salted_channel, hour).count() # 去掉前缀第二次聚合 result_final result_partial.withColumn( channel, when(col(salted_channel).startswith(default_), lit(default)) .otherwise(col(salted_channel)) ).groupBy(channel, hour).agg(spark_sum(count).alias(count))方案三增加 shuffle partitions把spark.sql.shuffle.partitions从 200 增加到 600让数据分散到更多的 task 上spark.conf.set(spark.sql.shuffle.partitions, 600)5.4 优化效果指标优化前优化后总耗时3 小时 20 分钟25 分钟最长 task 耗时3 小时8 分钟executor OOM有无shuffle 数据量45GB18GB这个案例说明数据倾斜的优化效果往往非常明显。但需要注意的是不同的作业倾斜的原因不同解决方案也不同不能生搬硬套。6. 常见问题与排查技巧实录6.1 executor OOM 怎么排查executor OOM 是 Spark 作业里最常见的问题之一。排查思路看 Spark UI 的 Executors 页面看是哪个 executor OOM 了它的内存使用情况如何。看 Stages 页面看 OOM 发生在哪个 stage这个 stage 的 task 数据量有多大。看日志OOM 的日志里通常会显示是哪个操作导致的比如 shuffle、join、cache。常见的 OOM 原因和解决方案原因解决方案单个分区数据量太大增加分区数减少每个分区的数据量cache 的数据量太大减少 cache 的数据量或者用 MEMORY_AND_DISK 级别shuffle 数据量太大增加 shuffle partitions减少每个 task 的数据量executor 内存太小增加 executor 内存数据倾斜解决数据倾斜问题6.2 作业跑得慢但看不出哪里慢有时候作业跑得慢但 Spark UI 上每个 stage 的耗时都不长看不出明显的瓶颈。这种情况通常是driver 端瓶颈比如 collect 了大量数据到 driver或者 driver 做了复杂的计算。小文件太多读取大量小文件会导致 task 数量过多调度开销变大。GC 时间太长在 Executors 页面看 GC 时间如果占比超过 10%说明 GC 是瓶颈。# 检查 GC 时间 # 在 Spark UI 的 Executors 页面看每个 executor 的 GC Time # 如果 GC Time 占比超过 10%需要调整内存配置6.3 小文件问题怎么解决小文件问题是 Spark 作业里另一个常见的问题。小文件太多会导致task 数量过多调度开销大。每个 task 处理的数据量很小CPU 利用率低。文件系统的元数据压力大。解决小文件问题的方法# 方法一在写入时合并小文件 df.repartition(100).write.parquet(/output) # 方法二使用 coalesce 减少分区数 df.coalesce(100).write.parquet(/output) # 方法三设置合并小文件的参数 spark.conf.set(spark.sql.adaptive.coalescePartitions.enabled, true) spark.conf.set(spark.sql.adaptive.advisoryPartitionSizeInBytes, 128MB)提示repartition会做全量 shufflecoalesce不会。如果只是减少分区数用coalesce更高效如果需要重新分布数据用repartition。6.4 我的实操心得心得一不要迷信“最优配置”网上有很多所谓的“Spark 最优配置”但那些配置是针对特定场景的。你的数据量、集群规模、作业逻辑都不一样直接抄配置往往效果不好。最好的方法是先理解每个参数的含义然后根据自己的情况做实验找到最适合自己的配置。心得二优化是一个迭代的过程不要指望一次优化就能解决所有问题。我通常的做法是先解决最明显的瓶颈比如数据倾斜然后看效果再解决下一个瓶颈。每次只改一个地方这样才能知道哪个改动有效。心得三监控和日志很重要优化之前一定要有监控和日志否则你连问题在哪里都不知道。Spark UI 是最基本的但还不够。我通常还会加一些自定义的日志比如每个 stage 的输入输出数据量、每个 task 的耗时等等。心得四不要过度优化优化是有成本的包括时间成本和维护成本。如果一个作业已经能在可接受的时间内跑完就不要再去折腾了。把精力放在真正有问题的作业上。心得五理解业务比理解技术更重要很多时候性能问题不是技术问题而是业务问题。比如数据倾斜可能是因为业务逻辑不合理导致某些 key 的数据特别多。这时候改业务逻辑比改技术方案更有效。6.5 常见问题速查表问题可能原因排查方法解决方案作业跑得慢数据倾斜看 Stages 页面的 task 耗时分布加随机前缀、saltingexecutor OOM分区数据量太大看 Stages 页面的 task 数据量增加分区数GC 时间长内存不足看 Executors 页面的 GC Time增加 executor 内存shuffle 数据量大分区数太少看 Stages 页面的 Shuffle Write增加 shuffle partitions小文件太多分区数太多看输出目录的文件数量coalesce 或 repartition作业卡在某个 stage数据倾斜或资源不足看 Spark UI 的 stage 详情根据具体情况处理7. 一些容易被忽视的细节7.1 数据格式的选择数据格式对性能的影响很大。Parquet 和 ORC 是列式存储格式支持谓词下推和列裁剪读取效率比 CSV 和 JSON 高很多。如果数据量很大建议用 Parquet 或 ORC。# 写入 Parquet 格式 df.write.parquet(/output/data.parquet) # 读取 Parquet 格式 df spark.read.parquet(/output/data.parquet)7.2 分区策略的设计如果数据是按时间分区的查询的时候可以利用分区裁剪只读取特定分区的数据。分区字段的选择很重要通常选择经常用于过滤的字段。# 按日期和渠道分区 df.write.partitionBy(dt, channel).parquet(/output/data)7.3 动态资源分配Spark 支持动态资源分配可以根据作业的负载自动增加或减少 executor。这对于多个作业共享集群的场景很有用。--conf spark.dynamicAllocation.enabledtrue \ --conf spark.dynamicAllocation.minExecutors5 \ --conf spark.dynamicAllocation.maxExecutors50 \ --conf spark.dynamicAllocation.initialExecutors107.4 推测执行推测执行speculative execution是 Spark 的一个特性当某个 task 跑得特别慢时Spark 会在另一个节点上启动一个相同的 task谁先跑完就用谁的结果。这对于数据倾斜或者节点故障的场景很有用。--conf spark.speculationtrue \ --conf spark.speculation.interval100ms \ --conf spark.speculation.multiplier1.5但推测执行也有副作用它会消耗额外的资源。如果集群资源紧张建议关闭推测执行。7.5 自适应查询执行Spark 3.0 引入了自适应查询执行AQE可以在运行时根据统计信息动态调整执行计划。AQE 包括三个主要功能动态合并 shuffle 分区、动态调整 join 策略、动态处理数据倾斜。--conf spark.sql.adaptive.enabledtrue \ --conf spark.sql.adaptive.coalescePartitions.enabledtrue \ --conf spark.sql.adaptive.skewJoin.enabledtrueAQE 对于数据倾斜和分区数不合理的场景非常有用建议在 Spark 3.0 及以上版本中启用。8. 最后再分享几个小技巧技巧一用 explain 看执行计划在写代码的时候可以用explain看 Spark 的执行计划确认有没有出现 CartesianProduct、BroadcastNestedLoopJoin 这种危险的算子。df.explain(True)技巧二用 Spark UI 的 SQL 页面看物理计划如果是 DataFrame/Dataset APISpark UI 的 SQL 页面会显示物理执行计划可以看到每个算子的实际执行情况。技巧三用 accumulators 做调试Accumulators 可以在不破坏作业逻辑的情况下收集统计信息比如统计某个条件的记录数。accum spark.sparkContext.accumulator(0) def count_errors(row): global accum if row.status error: accum 1 return row df.foreach(count_errors) print(fError count: {accum.value})技巧四用 checkpoint 切断 lineage如果 lineage 太长会导致 driver 内存压力大而且一旦某个 partition 丢失需要重新计算整个 lineage。这时候可以用 checkpoint 把中间结果写到可靠的存储上切断 lineage。spark.sparkContext.setCheckpointDir(/checkpoint) df.checkpoint()技巧五合理使用 repartition 和 coalescerepartition会做全量 shuffle可以增加或减少分区数coalesce不会 shuffle只能减少分区数。如果需要增加分区数或者需要重新分布数据用repartition如果只是减少分区数用coalesce。# 增加分区数 df.repartition(600) # 减少分区数 df.coalesce(100)这些技巧看起来简单但在实际优化中往往能起到意想不到的效果。我在实际使用中发现很多性能问题其实不需要复杂的调优只需要把一些基本的细节做好就能拿到不错的提升。