企业官网建设流程全解析

1. 为什么你的 Spark 作业跑得比隔壁慢三倍

刚接手一个数据处理任务时，我遇到过一件特别邪门的事：同样一份日志数据，同样的集群规模，隔壁组的作业跑完只要十二分钟，我这边跑了四十分钟还没出结果。更气人的是，资源监控面板上 CPU 和内存都没跑满，看起来一切正常，但任务就是慢。后来花了整整两天时间排查，才发现问题出在几个非常隐蔽的地方——数据倾斜、序列化方式、以及 shuffle 分区数设置不合理。

这件事让我意识到，Spark 性能优化这件事，很多人以为就是调几个参数的事，实际上它是一套从代码逻辑到资源配置再到运行时调优的完整方法论。你光知道spark.sql.shuffle.partitions要调大调小，但不知道为什么要调、什么时候调、调完会有什么副作用，那跟瞎蒙没什么区别。

这篇文章想聊的就是这套方法论。我会从最常见的性能瓶颈入手，把每个优化手段背后的原理讲清楚，再给出可以直接抄的配置和代码示例。不管你是刚接触 Spark 的新手，还是已经写过一些作业但总觉得跑得不够快的开发者，应该都能从中找到对自己有用的东西。核心关键词就一个：Spark 性能优化，但我会把它拆成数据倾斜处理、序列化调优、内存管理、shuffle 优化、SQL 执行计划分析这几个具体方向来展开。

先说一个我自己的判断标准：一个 Spark 作业如果跑得慢，90% 的情况下问题出在三个地方——数据分布不均匀导致个别 task 拖后腿、shuffle 阶段产生了大量不必要的磁盘和网络 IO、以及内存配置和实际数据量不匹配导致频繁 GC 甚至 OOM。剩下的 10% 才是一些边角料问题，比如小文件太多、UDF 写得低效之类的。

所以接下来的内容会围绕这三个核心问题展开，每个问题我都会先讲清楚它是怎么产生的，再给出具体的排查方法和优化手段。中间会穿插一些我自己踩过的坑和实测有效的技巧，希望能帮你少走一些弯路。

2. 数据倾斜：那个让 99 个 task 等 1 个 task 的元凶

2.1 数据倾斜到底是怎么发生的

数据倾斜这个词听起来很学术，但用大白话讲就是：你有一百个 task 要处理数据，其中九十九个 task 各自处理一百万条记录，剩下那一个 task 要处理一亿条记录。结果就是九十九个 task 十分钟就跑完了，然后集体在那儿干等，等那个倒霉的 task 慢慢磨。整个作业的耗时完全取决于最慢的那个 task，这就是木桶效应在分布式计算里的完美体现。

那为什么会出现这种分布不均？最常见的原因是groupByKey、join、distinct这类操作会根据 key 进行数据重分布。如果某个 key 对应的数据量特别大——比如你按用户 ID 聚合，结果有个用户是爬虫账号，产生了几千万条记录——那这个 key 就会全部落到同一个 partition 里，对应的 task 自然就慢得离谱。

还有一种情况是 join 操作时，如果两张表都很大，Spark 会选择 SortMergeJoin，这时候如果 join key 分布不均，同样会导致倾斜。更隐蔽的是，有时候数据本身分布是均匀的，但你的分区策略有问题，比如用了repartition(1)把数据全塞到一个分区里，那不管数据怎么分布都会倾斜。

2.2 怎么快速判断作业是否存在数据倾斜

在动手优化之前，你得先确认问题确实出在数据倾斜上。我常用的判断方法有这么几个：

第一，看 Spark UI 的 Stage 页面。如果某个 Stage 里大部分 task 的耗时都在几秒到几十秒，但有几个 task 耗时是几分钟甚至几十分钟，那基本可以确定是倾斜了。注意看 task 的 Duration 分布，如果最大值和平均值差距在十倍以上，就要警惕。

第二，看 task 的输入数据量。在 Stage 详情页里，每个 task 的 Input Size 和 Shuffle Read Size 都能看到。如果某个 task 的 Shuffle Read Size 是其他 task 的几十倍，那倾斜就实锤了。

第三，看 executor 的日志。如果某个 executor 频繁 GC 或者报 OOM，而其他 executor 很闲，那大概率是它分到了倾斜的 key。

我一般会先用第一个方法快速扫一眼，确认有倾斜之后再去看具体是哪个 key 导致的。定位具体 key 的方法后面会讲。

2.3 两阶段聚合：用加盐的方式打散热点 key

解决数据倾斜最通用的思路就是“加盐”——给热点 key 加上随机前缀，让它先分散到多个 partition 里做局部聚合，然后再去掉前缀做全局聚合。这个方法对groupByKey和reduceByKey类型的操作特别有效。

具体怎么做呢？假设你原来是这样写的：

# 原始写法，如果 user_id 分布不均就会倾斜 result = df.groupBy("user_id").agg(sum("amount").alias("total"))

改成两阶段聚合的写法：

import pyspark.sql.functions as F # 第一阶段：加随机前缀，局部聚合 salted = df.withColumn("salt", (F.rand() * 10).cast("int")) \ .withColumn("salted_key", F.concat(F.col("user_id"), F.lit("_"), F.col("salt"))) partial = salted.groupBy("salted_key").agg( F.sum("amount").alias("partial_total"), F.first("user_id").alias("user_id") ) # 第二阶段：去掉前缀，全局聚合 result = partial.groupBy("user_id").agg(F.sum("partial_total").alias("total"))

这里的 salt 数量怎么定？我的经验是：如果倾斜的 key 只有几个，salt 设 10 到 100 就够了；如果倾斜的 key 比较多，可以适当加大。但 salt 也不是越大越好，因为第二阶段聚合的数据量会随着 salt 数量线性增长。一般建议 salt 数量不要超过集群核心数的两倍。

这个方法的代价是多了一次 shuffle，但因为第一阶段把数据打散了，每个 task 处理的数据量更均匀，整体耗时反而会大幅下降。我实测过一个倾斜严重的作业，用两阶段聚合之后，耗时从四十分钟降到了八分钟。

2.4 大表 join 小表：广播变量才是正解

如果倾斜发生在 join 操作上，而且其中一张表比较小（比如几 MB 到几十 MB），那最简单的办法就是用广播 join。Spark 会把小表广播到每个 executor 上，然后在大表的每个 partition 里直接做 map 端的 join，完全避免了 shuffle。

在 Spark SQL 里，你可以通过 hint 来强制广播：

SELECT /*+ BROADCAST(small_table) */ * FROM big_table JOIN small_table ON big_table.id = small_table.id

或者在 DataFrame API 里用broadcast函数：

from pyspark.sql.functions import broadcast result = big_df.join(broadcast(small_df), "id")

广播 join 的阈值默认是 10MB，可以通过spark.sql.autoBroadcastJoinThreshold调整。但注意不要设得太大，否则广播本身会消耗大量网络带宽和 executor 内存。我一般会把这个值设在 50MB 到 100MB 之间，具体看集群的内存情况。

如果两张表都很大怎么办？那就只能用加盐 join 了。思路和两阶段聚合类似：给大表的 key 加上随机前缀，把小表的 key 复制多份并加上对应的前缀，然后做 join。这个方法实现起来稍微复杂一些，但效果很好。

2.5 定位具体是哪个 key 导致了倾斜

前面说了怎么判断有没有倾斜，但很多时候你还想知道到底是哪个 key 在捣乱。我常用的方法是用sample加countByKey来快速定位：

# 对数据采样，统计每个 key 的出现次数 top_keys = df.select("user_id") \ .sample(withReplacement=False, fraction=0.1) \ .rdd.countByKey() \ .items() # 按出现次数排序，取前 20 个 sorted_keys = sorted(top_keys, key=lambda x: x[1], reverse=True)[:20] for key, count in sorted_keys: print(f"Key: {key}, Count: {count}")

采样比例根据数据量调整，数据量大的话 0.01 甚至 0.001 就够了。找到热点 key 之后，你可以选择单独处理这些 key——比如把它们过滤出来单独计算，最后再合并结果。这个方法虽然有点“土”，但在紧急情况下特别管用。

注意：采样统计本身也会触发一次作业，如果数据量特别大，建议在非高峰时段做这个排查。

3. 序列化与内存：那些让你频繁 GC 的隐形杀手

3.1 为什么序列化方式能决定作业快慢

Spark 在节点之间传输数据、在内存里缓存数据、以及把数据写到磁盘上时，都需要把对象序列化成字节流。默认情况下，Spark 用的是 Java 自带的序列化机制，这个机制的特点是兼容性好但速度慢、体积大。而 Kryo 序列化机制则快得多，序列化后的体积也能小很多——通常能减少到 Java 序列化的十分之一左右。

这个差距在实际作业里意味着什么？假设你的作业需要 shuffle 10GB 的数据，用 Java 序列化可能实际传输 100GB，而用 Kryo 可能只需要 10GB。网络传输量差了十倍，耗时差距可想而知。而且序列化后的数据更小，意味着内存里能缓存更多数据，磁盘 IO 也会减少。

启用 Kryo 序列化的配置很简单：

spark = SparkSession.builder \ .appName("MyApp") \ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ .config("spark.kryo.registrationRequired", "false") \ .getOrCreate()

但这里有个坑：Kryo 需要你注册自定义的类，否则它会把每个类的全限定名都写进序列化结果里，反而增加体积。对于自定义的 case class 或者 Java Bean，一定要注册：

spark.conf.set("spark.kryo.classesToRegister", "com.example.MyClass,com.example.AnotherClass")

如果类特别多，可以用spark.kryo.registrator指定一个自定义的注册器。我一般会在项目里建一个KryoRegistrator类，把所有需要序列化的类都注册进去。

3.2 内存管理：executor 内存到底该怎么分

Spark 的内存模型经历过几次大的变化，现在（Spark 2.x 和 3.x）的统一内存管理模型把 executor 内存分成了几个部分：预留内存（默认 300MB）、用户内存（默认 40%）、以及 Spark 内存（默认 60%）。Spark 内存又分为存储内存和执行内存，两者可以互相借用。

这个模型听起来有点绕，但实际调优时你只需要关注几个关键点：

第一，spark.executor.memory决定了每个 executor 的总内存。这个值不是越大越好，因为内存越大，GC 的停顿时间可能越长。我一般会根据数据量和集群配置来定，常见的是 8GB 到 16GB。

第二，spark.memory.fraction控制 Spark 内存占总内存的比例，默认 0.6。如果你的作业有大量缓存操作（比如cache()或persist()），可以适当调大这个值；如果主要是 shuffle 操作，保持默认就行。

第三，spark.memory.storageFraction控制存储内存占 Spark 内存的比例，默认 0.5。如果你的作业缓存了大量数据，可以调大这个值，避免缓存被执行内存挤掉。

我踩过的一个坑是：有一次作业频繁 OOM，我以为是内存不够，就把 executor 内存从 8GB 加到了 16GB，结果 OOM 反而更频繁了。后来才发现真正的问题是数据倾斜导致单个 task 处理的数据量太大，加内存只是让 GC 更痛苦而已。所以内存调优一定要先确认没有数据倾斜。

3.3 GC 调优：别让垃圾回收拖了后腿

GC（垃圾回收）是 Spark 作业性能的隐形杀手。当 JVM 堆内存里对象太多时，GC 会频繁触发，每次触发都会暂停所有工作线程（Stop-The-World），导致 task 执行变慢。如果你的 Spark UI 里看到 executor 的 GC 时间占比超过 10%，那就说明 GC 有问题了。

减少 GC 的手段主要有这么几个：

第一，用 Kryo 序列化减少对象体积，这个前面说过了。

第二，尽量用基本类型代替包装类型。比如用Int而不是Integer，用Array[Int]而不是List[Int]。在 Scala 里，基本类型的数组内存占用远小于对象列表。

第三，调整 GC 策略。对于大内存的 executor，推荐使用 G1 GC：

--conf spark.executor.extraJavaOptions="-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35"

G1 GC 的特点是停顿时间可预测，适合大堆内存的场景。如果是小内存 executor（小于 8GB），用默认的 Parallel GC 也可以。

第四，减少cache()的使用。很多人习惯性地把 DataFrame 缓存起来，但实际上如果数据只用一次，缓存反而增加了内存压力和 GC 负担。只在数据会被多次使用时才缓存。

3.4 堆外内存：一个容易被忽略的优化点

除了 JVM 堆内存，Spark 还可以使用堆外内存（Off-Heap Memory）来存储序列化后的数据和执行内存。堆外内存的好处是不受 JVM GC 管理，可以减少 GC 压力。

启用堆外内存的配置：

spark.conf.set("spark.memory.offHeap.enabled", "true") spark.conf.set("spark.memory.offHeap.size", "4g")

但堆外内存也不是银弹。它的缺点是调试困难，如果出现内存泄漏很难排查。而且堆外内存的分配和释放也有开销。我一般会在 GC 问题特别严重、且堆内存调优已经到极限的情况下才考虑用堆外内存。

提示：启用堆外内存后，要确保 executor 的物理内存足够，因为堆外内存是额外分配的，不包含在spark.executor.memory里。

4. Shuffle 优化：减少磁盘 IO 和网络传输的关键

4.1 Shuffle 为什么这么贵

Shuffle 是 Spark 里最昂贵的操作，没有之一。它涉及到磁盘 IO、网络传输、序列化/反序列化、以及数据排序等多个环节。一个 Stage 的 shuffle 数据量如果很大，那这个 Stage 的耗时基本就取决于 shuffle 的效率了。

Shuffle 的过程大致是这样的：map 端的 task 把数据按 key 分区后写到本地磁盘，reduce 端的 task 再从各个 map 端拉取属于自己的分区数据。这个过程中，磁盘读写和网络传输是主要的瓶颈。

所以 shuffle 优化的核心思路就两个：减少 shuffle 的数据量，以及提高 shuffle 的效率。下面分别来说。

4.2 调整 shuffle 分区数：不是越多越好，也不是越少越好

spark.sql.shuffle.partitions这个参数控制的是 shuffle 之后的分区数，默认值是 200。这个默认值对于小数据量来说太大了——200 个 task 每个只处理几 KB 数据，task 启动和调度的开销远大于实际计算的开销。对于大数据量来说又太小了——每个 task 要处理几 GB 数据，容易 OOM 而且并行度不够。

那怎么确定合适的值呢？我的经验公式是：shuffle 分区数 = shuffle 数据量 / 每个 task 的理想处理量。每个 task 的理想处理量一般在 100MB 到 200MB 之间。比如你的 shuffle 数据量是 50GB，那分区数设在 250 到 500 之间比较合适。

但手动设置有个问题：数据量是动态变化的，今天 50GB 明天可能就 500GB 了。所以更好的做法是启用自适应查询执行（AQE）里的自动分区调整：

spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")

AQE 会根据每个 stage 的实际数据量自动合并小分区，避免分区过多或过少的问题。这个功能在 Spark 3.0 之后默认开启，但有些参数需要手动配置。

4.3 减少不必要的 shuffle：从代码层面入手

很多 shuffle 其实是可以通过改写代码来避免的。举几个常见的例子：

第一个例子：用reduceByKey代替groupByKey。groupByKey会把所有相同 key 的 value 都拉到一起，不做任何预聚合；而reduceByKey会在 map 端先做一次局部聚合，大大减少 shuffle 的数据量。如果只是要做求和、计数这类聚合操作，永远优先用reduceByKey或aggregateByKey。

第二个例子：用filter提前过滤数据。如果你在 join 之后才做过滤，那 join 的时候就已经 shuffle 了大量无用数据。把过滤条件提前到 join 之前，能显著减少 shuffle 量。

第三个例子：用coalesce代替repartition。repartition会触发全量 shuffle，而coalesce只是合并分区，不涉及数据重分布。如果你只是想把分区数减少（比如写文件时不想产生太多小文件），用coalesce就够了。

第四个例子：避免多次 join 同一份数据。如果你需要对同一份数据做多次 join，可以考虑先把它缓存起来，或者用广播变量。

4.4 Shuffle 文件管理：小文件问题的解决思路

Shuffle 过程中会产生大量的中间文件，如果分区数很多，每个分区都会产生一个文件，最终可能导致磁盘上出现几十万个小文件。这不仅会拖慢 shuffle 的读写速度，还可能耗尽文件系统的 inode。

解决这个问题的办法有几个：

第一，合理设置分区数，避免分区过多。这个前面说过了。

第二，启用 shuffle 文件合并。Spark 默认会把同一个 executor 上同一个 reduce 分区的多个 map 输出合并成一个文件，这个机制叫spark.shuffle.consolidateFiles（在 Spark 2.x 之后默认开启）。

第三，定期清理 shuffle 数据。Spark 会在作业结束后保留 shuffle 数据一段时间（默认是spark.shuffle.service.removeShuffle控制的），如果磁盘空间紧张，可以调小这个时间。

第四，如果用的是外部 shuffle 服务，可以配置spark.shuffle.service.index.cache.size来缓存 shuffle 索引文件，减少磁盘 IO。

4.5 实测有效的 shuffle 参数组合

经过多次实测，我总结了一套比较通用的 shuffle 参数配置，适用于大多数中等规模的数据处理作业：

spark.conf.set("spark.sql.shuffle.partitions", "500") spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m") spark.conf.set("spark.shuffle.compress", "true") spark.conf.set("spark.shuffle.spill.compress", "true") spark.conf.set("spark.io.compression.codec", "lz4") spark.conf.set("spark.shuffle.file.buffer", "64k") spark.conf.set("spark.shuffle.io.maxRetries", "5") spark.conf.set("spark.shuffle.io.retryWait", "10s")

这里解释几个关键参数：spark.shuffle.compress和spark.shuffle.spill.compress开启 shuffle 数据的压缩，能显著减少磁盘和网络 IO；spark.io.compression.codec用 lz4 而不是默认的 snappy，因为 lz4 的压缩和解压速度更快；spark.shuffle.file.buffer调大到 64KB，减少写磁盘的次数；spark.shuffle.io.maxRetries和retryWait增加重试次数和等待时间，避免网络抖动导致 shuffle 失败。

注意：这些参数不是万能的，具体值需要根据你的集群配置和数据特征来调整。建议先在测试环境验证，再上生产。

5. SQL 执行计划：看懂它，你就能找到一半的性能问题

5.1 为什么要学会看执行计划

很多人写 Spark SQL 就是凭感觉，写完跑一下，慢了就加资源，再慢就换更贵的机器。这种做法在短期内可能有效，但长期来看既浪费资源又解决不了根本问题。而执行计划（Execution Plan）就是 Spark 给你的“诊断报告”，它会告诉你 Spark 打算怎么执行你的 SQL，包括用哪种 join 策略、怎么分区、有没有做谓词下推等等。

看执行计划的方法很简单，在 SQL 前面加EXPLAIN关键字：

df.explain(True) # 打印详细的执行计划

或者：

EXPLAIN EXTENDED SELECT * FROM table_a JOIN table_b ON table_a.id = table_b.id

执行计划里最关键的是这几个部分：Scan表示数据扫描方式，Filter表示过滤条件，Join表示 join 策略，Exchange表示 shuffle 操作。如果你看到Exchange出现在不该出现的地方，那说明有可以优化的 shuffle。

5.2 识别常见的执行计划问题

第一个常见问题是CartesianProduct。如果你在 join 的时候忘了写 join 条件，或者条件写错了，Spark 会执行笛卡尔积，把两张表的每一行都组合一次。这个操作的数据量是两张表行数的乘积，几乎必然导致 OOM。看到执行计划里有CartesianProduct，第一件事就是检查 join 条件。

第二个问题是BroadcastNestedLoopJoin。这个 join 策略是在没有等值 join 条件时使用的，效率很低。如果你看到这个，说明你的 join 条件可能不是等值条件，或者 Spark 无法识别为等值条件。尽量把 join 条件写成等值形式。

第三个问题是Exchange出现在Filter之前。理想情况下，过滤操作应该在 shuffle 之前执行，这样能减少 shuffle 的数据量。如果执行计划显示先 shuffle 再 filter，那说明谓词下推没有生效。检查一下过滤条件是不是用了 UDF 或者复杂的表达式，这些会阻止谓词下推。

第四个问题是SortMergeJoin用于小表 join 大表。如果小表足够小，应该用BroadcastHashJoin来避免 shuffle。检查spark.sql.autoBroadcastJoinThreshold的值，以及小表的实际大小。

5.3 用 AQE 自动优化执行计划

Spark 3.0 引入的自适应查询执行（AQE）是一个非常重要的优化特性。它能在运行时根据实际的统计数据动态调整执行计划，包括自动合并小分区、自动处理数据倾斜、自动选择 join 策略等。

启用 AQE 的配置：

spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") spark.conf.set("spark.sql.adaptive.localShuffleReader.enabled", "true")

其中spark.sql.adaptive.skewJoin.enabled是自动处理数据倾斜的开关，它会自动检测倾斜的分区并把它拆分成多个小分区。这个功能对于前面说的数据倾斜问题特别有效，很多时候不需要手动加盐就能解决。

但 AQE 也不是万能的。它需要收集运行时统计信息，这会带来一定的额外开销。而且对于特别复杂的查询，AQE 的优化效果可能有限。我一般会先开启 AQE，如果效果不理想再手动调优。

5.4 一个真实的执行计划优化案例

之前遇到过一个作业，SQL 逻辑很简单：两张表 join 之后做聚合。但跑起来特别慢，四十分钟才出结果。用EXPLAIN看了一下执行计划，发现两个问题：

第一，join 策略是SortMergeJoin，但其中一张表只有 5MB，完全可以广播。原因是spark.sql.autoBroadcastJoinThreshold被设成了 1MB，太小了。

第二，聚合操作之前有一次多余的Exchange，原因是spark.sql.shuffle.partitions设成了 2000，导致 shuffle 数据被分成了 2000 个分区，每个分区数据量很小，task 调度开销巨大。

调整了两个参数之后——把广播阈值调到 50MB，把 shuffle 分区数降到 200——作业耗时从四十分钟降到了六分钟。这个案例说明，很多时候性能问题不是资源不够，而是配置不合理。

6. 那些让我印象深刻的踩坑经历

6.1 一个 UDF 让作业慢了二十倍

有一次写了一个数据处理作业，逻辑是用 Python UDF 对某个字段做字符串处理。作业跑起来之后慢得离谱，一个简单的 ETL 跑了将近一个小时。排查了半天，最后发现瓶颈在那个 Python UDF 上。

原因很简单：Python UDF 需要在 JVM 和 Python 进程之间来回序列化数据，每行数据都要经历一次“JVM 序列化 -> 发送到 Python -> Python 处理 -> 序列化回 JVM”的过程。数据量大的时候，这个开销是灾难性的。

后来我把 Python UDF 改写成了 Spark 内置函数，耗时直接降到了三分钟。所以我的经验是：能用内置函数解决的，绝对不要用 UDF。如果非要用 UDF，优先用 Scala UDF 而不是 Python UDF，因为 Scala UDF 直接在 JVM 里执行，没有跨进程通信的开销。

如果确实需要用 Python 的逻辑，可以考虑用 Pandas UDF（向量化 UDF），它用 Arrow 格式批量传输数据，比逐行传输的 Python UDF 快很多：

from pyspark.sql.functions import pandas_udf import pandas as pd @pandas_udf("string") def process_text(s: pd.Series) -> pd.Series: return s.str.upper().str.strip()

6.2 小文件问题：一个被低估的性能杀手

有一次作业跑得特别慢，但看执行计划没什么问题，资源也够。后来发现是输入数据目录下有几十万个小文件，每个文件只有几 KB。Spark 在读取这些小文件时，每个文件都要单独打开、读取、关闭，光是文件系统的元数据操作就耗费了大量时间。

解决小文件问题的办法有几个：如果小文件是上游作业产生的，可以在上游用coalesce减少输出文件数；如果小文件已经存在，可以用 Spark 的spark.sql.files.maxPartitionBytes控制每个分区的读取量，或者用spark.sql.files.openCostInBytes让 Spark 把小文件合并到同一个分区里读取。

我一般会在写入数据时用coalesce控制文件数量：

df.coalesce(10).write.parquet("/path/to/output")

这里的 10 是根据输出数据量估算的，目标是让每个文件在 128MB 左右。如果数据量是 1GB，那 8 到 10 个文件比较合适。

6.3 缓存用错了地方，反而拖慢了作业

cache()和persist()是常用的优化手段，但用错了地方反而会拖慢作业。我见过有人把整个 DataFrame 都缓存起来，结果内存不够，Spark 只能把部分缓存数据溢写到磁盘，反而增加了 IO。

正确的做法是：只缓存会被多次使用的数据，而且缓存之前先做过滤和聚合，减少缓存的数据量。另外，缓存级别也要根据数据大小和内存情况来选择。如果内存充足，用MEMORY_ONLY；如果内存紧张，用MEMORY_AND_DISK；如果数据量特别大，可以考虑用DISK_ONLY或者带序列化的缓存级别。

还有一个容易忽略的点：缓存用完要及时释放。用unpersist()释放不再需要的缓存，避免占用宝贵的内存资源。

df.cache() # 使用 df 做多次操作 df.count() df.groupBy("category").count().show() # 用完之后释放 df.unpersist()

6.4 动态资源分配：让集群利用率翻倍

Spark 的动态资源分配（Dynamic Resource Allocation）是一个经常被忽略但效果很好的优化手段。它允许 Spark 根据作业的实际负载动态增减 executor 数量——需要更多计算资源时自动申请，空闲时自动释放。

启用动态资源分配的配置：

spark.conf.set("spark.dynamicAllocation.enabled", "true") spark.conf.set("spark.dynamicAllocation.minExecutors", "2") spark.conf.set("spark.dynamicAllocation.maxExecutors", "50") spark.conf.set("spark.dynamicAllocation.initialExecutors", "5") spark.conf.set("spark.dynamicAllocation.executorIdleTimeout", "60s") spark.conf.set("spark.dynamicAllocation.schedulerBacklogTimeout", "1s")

这个功能特别适合多个作业共享集群的场景。以前每个作业都要预留固定数量的 executor，导致集群利用率很低；启用动态分配之后，作业按需申请资源，集群利用率能提升一倍以上。

但动态资源分配也有代价：executor 的申请和释放需要时间，如果作业的负载波动特别频繁，可能会导致资源频繁申请释放，反而增加开销。所以executorIdleTimeout和schedulerBacklogTimeout这两个参数要根据作业的特征来调整。

7. 从代码到集群：一套完整的优化检查清单

7.1 代码层面的自查要点

在提交作业之前，我一般会过一遍这个清单：

• 有没有用groupByKey？如果有，能不能改成reduceByKey或aggregateByKey？
• 有没有用 Python UDF？如果有，能不能改成内置函数或 Pandas UDF？
• join 操作有没有等值条件？小表能不能广播？
• 有没有不必要的repartition？能不能用coalesce代替？
• 过滤操作是不是尽量提前了？
• 有没有重复计算同一个 DataFrame？需不需要缓存？
• 数据倾斜的风险点在哪里？有没有加盐或广播的方案？

这个清单看起来简单，但每次过一遍都能发现一两个可以优化的点。我自己的经验是，代码层面的优化往往比调参的效果更明显，而且成本更低。

7.2 配置层面的参数模板

下面是我常用的一套 Spark 配置模板，适用于大多数中等规模的数据处理作业。你可以根据自己的集群情况和数据特征调整：

spark = SparkSession.builder \ .appName("OptimizedJob") \ .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .config("spark.sql.adaptive.skewJoin.enabled", "true") \ .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m") \ .config("spark.sql.shuffle.partitions", "500") \ .config("spark.sql.autoBroadcastJoinThreshold", "50m") \ .config("spark.shuffle.compress", "true") \ .config("spark.shuffle.spill.compress", "true") \ .config("spark.io.compression.codec", "lz4") \ .config("spark.shuffle.file.buffer", "64k") \ .config("spark.executor.memory", "8g") \ .config("spark.executor.cores", "4") \ .config("spark.executor.memoryOverhead", "2g") \ .config("spark.dynamicAllocation.enabled", "true") \ .config("spark.dynamicAllocation.minExecutors", "2") \ .config("spark.dynamicAllocation.maxExecutors", "50") \ .config("spark.executor.extraJavaOptions", "-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35") \ .getOrCreate()

这里有几个参数需要特别说明：spark.executor.memoryOverhead是堆外内存，用于 JVM 本身的开销、字符串常量池、网络缓冲区等，一般设为 executor 内存的 10% 到 20%；spark.executor.cores设为 4 到 5 比较合适，太多会导致每个 core 分到的内存太少，容易 OOM。

7.3 监控与持续调优

性能优化不是一次性的工作，而是一个持续的过程。我一般会关注这几个监控指标：

第一，Spark UI 里的 Stage 耗时分布。如果某个 Stage 的耗时突然增加，说明数据特征可能发生了变化。

第二，executor 的 GC 时间占比。如果超过 10%，说明内存配置需要调整。

第三，shuffle 的读写数据量。如果 shuffle 数据量远大于输入数据量，说明有可以优化的 shuffle 操作。

第四，task 的耗时分布。如果最大值和平均值差距很大，说明有数据倾斜。

第五，磁盘和网络 IO。如果 IO 等待时间很长，说明 shuffle 或缓存策略需要优化。

我习惯在作业里加一些自定义的监控埋点，记录每个关键步骤的耗时和数据量。这样当作业变慢时，能快速定位到是哪个环节出了问题。

7.4 一个真实的优化前后对比

最后分享一个我最近做的优化案例。一个日志分析作业，原始版本跑一次要四十五分钟，经过以下优化之后降到了七分钟：

这个案例里，每一项优化的效果都不是孤立的，它们叠加在一起才产生了这么明显的提升。单独看每一项可能只快了几分钟，但组合起来就是几倍的差距。

提示：优化的时候建议一次只改一个变量，改完测一下效果，确认有效再改下一个。如果一次改太多，出了问题很难定位是哪个改动导致的。

8. 写在最后的一些个人体会

做 Spark 性能优化这些年，我最大的体会是：不要一上来就调参数，先看代码逻辑和执行计划。很多性能问题其实是代码写得不合理导致的，调参只能缓解症状，不能根治问题。就像前面说的那个 Python UDF 的案例，你把 executor 内存加到 64GB 也没用，因为瓶颈根本不在内存上。

另一个体会是：优化要有数据支撑，不能凭感觉。我见过太多人凭直觉调参数，结果越调越慢。正确的做法是先通过 Spark UI、执行计划、日志等手段定位瓶颈，然后有针对性地优化，最后用数据验证效果。这个过程可能需要反复几次，但每次都能学到新东西。

还有一个容易被忽略的点：不同的数据特征需要不同的优化策略。一个在 A 数据集上效果很好的配置，换到 B 数据集上可能完全不起作用。所以不要盲目照搬别人的配置，要理解每个参数背后的原理，然后根据自己的实际情况来调整。

最后想说，Spark 性能优化是一个需要不断积累经验的领域。每次遇到新的问题，都是一次学习的机会。我到现在也不敢说自己精通所有优化手段，但至少遇到问题时知道从哪里入手、怎么排查、怎么验证。希望这篇文章能帮你建立起自己的优化思路，而不是只记住几个参数值。