1. 为什么 Spark 性能优化是个绕不开的坎干过大数据的人都有一个共识：Spark 作业能跑起来不难，难的是跑得又快又稳。我见过太多团队，集群规模不小，机器配置也不差，但一个 ETL 任务动辄跑两三个小时，资源利用率却低得可怜。更让人头疼的是，有些作业平时看着还行，一到数据量翻倍就直接 OOM 或者长尾拖死，整个调度链路跟着遭殃。Spark 性能优化这件事，本质上是在三个维度上找平衡：CPU 利用率、内存利用率和 IO 吞吐。任何一个维度成为瓶颈，整个作业的效率就被它锁死。而大部分性能问题，根源不在 Spark 本身，而在于数据分布、分区策略、算子选择和资源配置这几个环节没有对齐业务特征。这篇文章适合谁看？如果你已经写过 Spark 作业，能看懂 DAG 图，知道 RDD 和 DataFrame 的区别，但在面对“作业跑得慢”“内存老是不够”“任务倾斜严重”这些问题时还靠试参数来碰运气，那这篇内容就是给你准备的。我会从设计思路、核心细节、实操过程到问题排查，把 Spark 性能优化这件事拆开讲透，每个结论都尽量给出背后的计算逻辑和判断依据，让你不只是抄配置，而是真正理解为什么这么配。2. 整体优化思路与方案选型拆解2.1 先定位瓶颈，再谈优化很多人一上来就问“Spark 怎么调优”，这个问题本身就没法回答。因为优化的前提是知道瓶颈在哪。Spark 作业的瓶颈大致分四类：CPU 密集、内存密集、IO 密集和网络密集。不同类型的瓶颈，优化手段完全不同，甚至互相矛盾。举个例子，如果你的作业卡在 shuffle 阶段，那加大 executor 内存可能有用；但如果瓶颈是 CPU 算力不足，加内存就是浪费资源。所以我的习惯是，拿到一个慢作业，先做三件事：看 Spark UI 的 Stage 详情：重点关注每个 Stage 的 duration、task 数量、shuffle read/write 大小、GC 时间占比。看 task 的时间分布：如果 max task 时间和 median task 时间差距超过 3 倍，基本可以判定存在数据倾斜。看资源等待时间：如果 task 的 scheduler delay 很高，说明资源调度跟不上，需要调整资源配置或动态分配策略。这三步做完，瓶颈基本就浮出水面了。下面这张表是我常用的瓶颈判断速查表：现象大概率瓶颈优先排查方向GC 时间占比超过 10%内存不足或对象过多executor 内存、序列化方式、数据结构某几个 task 特别慢数据倾斜key 分布、分区策略、join 方式shuffle read 巨大shuffle 分区不合理分区数、聚合方式、broadcast 可行性CPU 利用率低但作业慢IO 或调度瓶颈数据源读取方式、并行度、资源分配频繁 spill内存不够或分区过大分区粒度、聚合内存比例2.2 优化的优先级排序Spark 优化的手段很多，但收益和成本差异巨大。我一般按这个优先级来排：第一优先级：数据层面优化。包括分区裁剪、谓词下推、列裁剪、数据压缩。这些优化几乎零成本，收益却最直接。比如一个 Parquet 表有 200 个字段，你只 select 其中 10 个，列裁剪就能把 IO 降低 95%。很多作业慢，就是因为读了一堆根本用不上的数据。第二优先级：算子与逻辑优化。比如用 reduceByKey 替代 groupByKey，用 broadcast join 替代 shuffle join，用 mapPartitions 替代 map。这些改动需要理解业务逻辑，但收益通常很显著。第三优先级：资源配置优化。调整 executor 数量、内存大小、core 数量、并行度。这一步需要结合集群实际情况，参数之间还有耦合关系，不能孤立地调。第四优先级：底层调优。比如序列化方式、GC 策略、堆外内存。这些属于精细调优，收益有限但风险较高，一般在前三步做完之后再考虑。注意：不要一上来就调 GC 参数或者换序列化器。我见过太多人花了两天调 JVM 参数，结果发现真正的瓶颈是数据倾斜，一个 salting 就解决了。2.3 不同场景下的方案取舍Spark 作业大致分三类场景，每类的优化重点不同：批处理 ETL 场景：这类作业通常数据量大、逻辑复杂、对延迟不敏感。优化重点是减少 shuffle、提高并行度、合理分区。broadcast join 和分区裁剪是主力手段。交互式查询场景：这类作业对响应时间敏感，通常数据量中等。优化重点是缓存策略、AQE（自适应查询执行）、以及合理的文件布局（小文件合并）。流处理场景：这类作业对延迟和稳定性要求高。优化重点是背压处理、状态管理、以及 checkpoint 频率。微批大小和 trigger 间隔需要仔细权衡。理解自己作业属于哪一类，才能把有限的优化精力花在刀刃上。3. 核心细节解析与实操要点3.1 分区策略：性能的地基分区是 Spark 性能优化的地基，分区没搞好，后面怎么调都是事倍功半。分区的核心目标是：让每个 task 处理的数据量尽量均匀，且大小适中。分区太大，单个 task 内存压力大，容易 spill 甚至 OOM；分区太小，task 数量过多，调度开销和 GC 开销上来了。那多少算合适？我的经验值是每个分区 128MB 到 256MB。这个数字不是拍脑袋来的，它和 HDFS block 大小、Spark 默认的 shuffle 分区处理能力都有关系。计算分区数的公式大致是：理想分区数 = 数据总量 / 每分区目标大小比如你有 100GB 数据要处理，目标每分区 200MB，那理想分区数就是 500 左右。但实际中还要考虑 core 的数量，分区数最好是 core 总数的 2 到 4 倍，这样能保证每个 core 都有活干，又不至于调度太频繁。对于 shuffle 分区，spark.sql.shuffle.partitions这个参数默认是 200，这个默认值在数据量小的时候还行，数据量一大就完全不够用。我一般会把它设成executor 数量 × 每 executor 的 core 数 × 2到× 4。但更好的做法是开启 AQE，让 Spark 自己根据 shuffle 数据量动态合并分区。# 开启自适应查询执行 spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "200MB") spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")AQE 的好处是它能在运行时根据实际数据量调整分区，避免了你手动估算的误差。实测下来，开启 AQE 后，很多作业的 shuffle 分区数能自动从 200 调整到几千，性能提升非常明显。3.2 数据倾斜：最顽固的性能杀手数据倾斜是 Spark 性能问题里最顽固的一个。它的表现是：绝大多数 task 几秒钟就跑完了，但少数几个 task 跑了几十分钟甚至几个小时。原因通常是某个 key 的数据量远超其他 key。判断数据倾斜的方法很简单，在 Spark UI 里看 Stage 的 task 时间分布，如果 max 和 median 差距超过 3 倍，基本就是倾斜了。更精确的做法是采样统计 key 的分布：# 采样统计 key 分布 sample_df = df.sample(fraction=0.1) key_counts = sample_df.groupBy("join_key").count().orderBy("count", ascending=False) key_counts.show(20)如果发现某个 key 的数量级远超其他，那就是倾斜的源头。解决数据倾斜有几种常用手段：方案一：加盐打散。给倾斜的 key 加上随机前缀，把它打散到多个分区，聚合完再去掉前缀。这个方法适合聚合场景。import pyspark.sql.functions as F # 加盐：给 key 加上 0 到 N 的随机前缀 salt_range = 10 salted_df = df.withColumn("salted_key", F.concat(F.col("join_key"), F.lit("_"), (F.rand() * salt_range).cast("int"))) # 聚合后再去掉盐 result = salted_df.groupBy("salted_key").agg(...) \ .withColumn("join_key", F.split("salted_key", "_")[0]) \ .groupBy("join_key").agg(...)方案二：broadcast join。如果倾斜发生在 join 阶段，且其中一张表足够小（默认 10MB 以内，可调），直接广播小表，避免 shuffle。from pyspark.sql.functions import broadcast result = large_df.join(broadcast(small_df), "join_key")方案三：拆分倾斜 key 单独处理。把倾斜的 key 和非倾斜的 key 分开处理，倾斜的部分单独优化，最后 union 回去。这个方法最灵活，但代码复杂度也最高。实操心得：加盐的盐值范围不是越大越好。盐值太大，会导致分区数暴增，调度开销反而拖慢作业。我一般从 10 开始试，观察 task 时间分布，如果还不够均匀再往上加。另外，加盐只适合聚合场景，如果是 join 场景，需要两边都加盐，逻辑会复杂很多。3.3 算子选择：小改动大收益Spark 的算子选择对性能影响极大，但很多人写代码时并不在意。我挑几个最常见的对比讲一下。reduceByKey vs groupByKey：这是最经典的对比。groupByKey 会把所有相同 key 的 value 拉到一起，不做预聚合，网络传输量大；reduceByKey 会在 map 端先做本地聚合，再 shuffle，网络传输量小得多。同样逻辑，reduceByKey 通常比 groupByKey 快几倍。join vs broadcast join：普通 join 需要 shuffle 两张表，broadcast join 把小表广播到每个 executor，避免了大表的 shuffle。如果小表能放进内存，broadcast join 是首选。map vs mapPartitions：map 对每一行调用一次函数，mapPartitions 对每个分区调用一次。如果函数里有创建连接、初始化资源这类操作，mapPartitions 能大幅减少开销。# 不推荐：每行都创建一次连接 def process_row(row): conn = create_connection() # 每行都创建，开销巨大 return conn.query(row) df.rdd.map(process_row) # 推荐：每个分区创建一次连接 def process_partition(rows): conn = create_connection() # 每个分区只创建一次 for row in rows: yield conn.query(row) df.rdd.mapPartitions(process_partition)collect vs take：collect 把所有数据拉到 driver，数据量大直接 OOM；take 只取前 N 条，安全得多。调试时用 take，别用 collect。3.4 内存管理与序列化Spark 的内存模型分三块：execution memory、storage memory 和 user memory。execution memory 用于 shuffle、join、sort 等操作，storage memory 用于缓存 RDD 和 broadcast 数据，user memory 用于用户自定义的数据结构和函数。默认情况下，execution 和 storage 共享一块统一内存区域，比例是 50% 对 50%，可以通过spark.memory.fraction和spark.memory.storageFraction调整。如果作业 shuffle 很重，可以适当提高 execution 的比例；如果缓存了很多数据，就提高 storage 的比例。序列化方式对内存和网络传输影响很大。Spark 默认用 Java 序列化，慢且占空间。换成 Kryo 序列化，通常能减少 2 到 5 倍的内存占用，速度也快不少。spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") spark.conf.set("spark.kryoserializer.buffer.max", "256m")但 Kryo 需要注册自定义类才能发挥最佳效果，否则每个对象都要写入类名，反而浪费空间：spark.conf.set("spark.kryo.registrationRequired", "true") # 通过 SparkConf 注册自定义类 conf.registerKryoClasses([MyClass1, MyClass2])注意：开启 registrationRequired 后，未注册的类会直接报错。这个开关适合在测试环境先跑一遍，把用到的类都注册好再上生产。我一般会先用非严格模式跑，从日志里收集需要注册的类，再开启严格模式。4. 实操过程与核心环节实现4.1 一个真实慢作业的优化全过程我拿一个典型的 ETL 作业来演示完整的优化过程。这个作业的逻辑是：从两张 Hive 表读取数据，做 join 和聚合，然后写入结果表。原始配置是默认的，跑一次要 90 分钟。第一步：定位瓶颈。打开 Spark UI，发现作业分三个阶段，其中 Stage 2 的 shuffle read 高达 800GB，且 task 时间分布极不均匀，max task 跑了 45 分钟，median 只有 2 分钟。这明显是数据倾斜加 shuffle 分区不合理。第二步：检查数据分布。对 join key 做采样统计，发现有一个 key 占了总数据量的 40%，这就是倾斜的根源。第三步：优化 join 方式。检查两张表的大小，发现其中一张表只有 50MB，完全可以 broadcast。改成 broadcast join 后，shuffle 直接消失了。# 优化前：普通 join，触发 shuffle result = large_df.join(small_df, "user_id") # 优化后：broadcast join，无 shuffle result = large_df.join(broadcast(small_df), "user_id")第四步：优化聚合。聚合阶段用的是 groupByKey，改成 reduceByKey 后，网络传输量减少了 70%。第五步：调整资源配置。原来每个 executor 配了 4 个 core 和 8GB 内存，core 太多导致每个 core 分到的内存少，频繁 GC。改成每个 executor 2 个 core 和 16GB 内存，GC 时间从 15% 降到 3%。第六步：开启 AQE。让 Spark 自动合并小分区，shuffle 分区数从固定的 200 调整到动态的 1500 左右。优化后，作业时间从 90 分钟降到 18 分钟，资源消耗还降低了 30%。这个案例说明，大部分性能问题不需要复杂的底层调优，把数据层面和算子层面的事情做好，收益就已经非常可观了。4.2 资源配置的计算方法资源配置是很多人头疼的问题，因为参数太多，而且互相影响。我分享一套自己的计算方法，虽然不能保证最优，但至少不会犯大错。第一步：确定 executor 的 core 数。经验值是每个 executor 配 4 到 5 个 core。core 太少，executor 数量多，调度开销大；core 太多，每个 core 分到的内存少，GC 压力大。我一般从 4 开始试。第二步：确定 executor 的内存。每个 core 配 4 到 8GB 内存。比如 4 个 core 的 executor，配 16GB 到 32GB 内存。但要注意，executor 内存不能超过节点物理内存的 80%，要给操作系统和其他进程留空间。第三步：确定 executor 数量。总 core 数 = executor 数量 × 每 executor 的 core 数。总 core 数一般不超过集群总 core 数的 80%，留一些给 driver 和其他作业。第四步：确定 driver 内存。如果作业里有 collect、broadcast 大表、或者 driver 端做复杂计算，driver 内存要相应加大。一般 4GB 到 8GB 够用，特殊情况可以到 16GB。第五步：确定并行度。并行度 = 总 core 数 × 2 到 4。这个并行度决定了同时运行的 task 数量，太少资源利用不足，太多调度开销大。举个例子，一个 10 节点的集群，每节点 32 core、128GB 内存。我的配置方案是：参数值计算依据executor cores4经验值，平衡调度和 GCexecutor memory20GB每 core 5GB，留出 overheadexecutor 数量1610 节点 × 每节点 1.6 个，取整总 core 数6416 × 4，占集群 80% 的 80%driver memory8GB有 broadcast 操作，适当加大并行度19264 × 3这套配置不是最优解，但能保证作业稳定运行，不会因为资源问题频繁失败。实际中还需要根据作业特点微调。4.3 文件布局与小文件治理数据源的存储格式和文件布局对 Spark 性能影响巨大。我见过一个作业，逻辑很简单，但跑得特别慢，最后发现是数据源有几十万个小文件，每个文件只有几 KB。Spark 读取时，每个小文件都要单独打开、读取、关闭，IO 开销完全压过了计算开销。解决小文件问题有两个方向：写入时控制文件大小和读取时合并小文件。写入时，可以通过repartition或coalesce控制输出文件数量：# 控制输出文件数量，每个文件约 128MB target_file_size = 128 * 1024 * 1024 # 128MB total_size = df.rdd.map(lambda x: len(str(x))).sum() # 估算总大小 num_files = max(1, int(total_size / target_file_size)) df.repartition(num_files).write.parquet("output_path")读取时，可以开启小文件合并：spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728") # 128MB spark.conf.set("spark.sql.files.openCostInBytes", "4194304") # 4MBmaxPartitionBytes控制单个分区读取的最大字节数，openCostInBytes告诉 Spark 打开一个文件的成本，Spark 会根据这两个参数把多个小文件合并到一个分区里读取。实操心得：小文件治理最好在数据写入阶段就做好，不要等到读取时再补救。我一般会在 ETL 的最后一步加一个 repartition，确保输出文件大小在 128MB 到 256MB 之间。另外，如果数据源是 Hive 表，可以用ALTER TABLE ... CONCATENATE或者INSERT OVERWRITE重写来合并小文件。4.4 缓存与持久化策略缓存是 Spark 里一把双刃剑。用得好，能避免重复计算，大幅提速；用不好，占满内存，导致 spill 甚至 OOM。缓存的适用场景是：同一个 DataFrame 被多次使用，且计算成本高。比如一个 DataFrame 经过复杂的 join 和聚合后，被用于多个下游计算，这时候缓存就很有价值。# 缓存中间结果 intermediate_df = complex_transformation(input_df) intermediate_df.cache() # 或者 persist(StorageLevel.MEMORY_AND_DISK) # 多次使用 result1 = intermediate_df.filter(...).count() result2 = intermediate_df.groupBy(...).agg(...)存储级别有几种选择：存储级别说明适用场景MEMORY_ONLY只存内存，存不下就重算数据量小，内存充足MEMORY_AND_DISK内存优先，存不下写磁盘通用场景，推荐MEMORY_ONLY_SER序列化后存内存内存紧张，CPU 充足DISK_ONLY只存磁盘数据量极大，内存严重不足我一般用 MEMORY_AND_DISK，兼顾性能和稳定性。如果内存特别紧张，用 MEMORY_ONLY_SER，序列化后能省不少内存，但读取时要反序列化，CPU 开销会上升。注意：缓存用完记得 unpersist，否则会一直占着内存。我见过一个作业，缓存了五六个 DataFrame 都没释放，最后 executor 内存被占满，新任务全部 spill 到磁盘，性能反而比不缓存还差。5. 常见问题与排查技巧实录5.1 常见问题速查表问题现象可能原因排查方法解决方案作业 OOM分区太大、缓存过多、数据倾斜看 Spark UI 的 Storage 和 Executor 页减小分区、释放缓存、处理倾斜GC 时间过长内存不足、对象过多、序列化不当看 Executor 页的 GC 时间占比加大内存、换 Kryo、减少对象创建任务长尾数据倾斜、分区不均看 task 时间分布加盐、broadcast、拆分 keyshuffle 数据量巨大分区不合理、算子选择不当看 Stage 的 shuffle read/write调整分区、用 reduceByKey作业卡在某个 Stage资源不足、调度延迟看 scheduler delay调整资源配置、开启动态分配频繁 spill内存不够、聚合数据量大看 spill 指标加大内存、提高并行度、预聚合读取数据慢小文件多、格式不合适看 input size 和 task 数合并小文件、换列式存储5.2 排查工具与技巧Spark UI 是首选工具。重点看几个地方：Stage 页的 task 时间分布、Executor 页的 GC 时间和内存使用、SQL 页的物理计划。物理计划里能看到 join 方式、分区数、过滤条件下推情况，是定位问题的关键。日志分析。executor 日志里会记录 spill、GC、shuffle 的详细信息。我一般会 grep 几个关键词：spill、GC、OutOfMemory、FetchFailed。这些关键词出现的地方，往往就是问题所在。采样调试。对于大数据量的作业，直接跑全量太慢。我习惯先用sample取 1% 到 10% 的数据跑一遍，观察逻辑是否正确、性能瓶颈在哪，确认没问题再跑全量。# 采样调试 sample_df = df.sample(fraction=0.01, seed=42) sample_df.write.mode("overwrite").parquet("debug_path")分阶段验证。把复杂作业拆成几个阶段，每个阶段单独跑，观察每个阶段的耗时和资源消耗。这样能精确定位到哪个阶段是瓶颈。5.3 独家避坑经验坑一：不要盲目加大并行度。并行度不是越大越好。并行度太大，task 数量暴增，调度开销和 GC 开销会拖慢作业。我见过有人把并行度设成几万，结果作业比原来还慢。并行度控制在总 core 数的 2 到 4 倍就够了。坑二：broadcast join 不是万能的。broadcast 的表必须足够小，默认阈值是 10MB，超过这个值会报错或者退化成普通 join。如果小表有几百 MB，强行 broadcast 会导致 driver OOM。这时候应该考虑其他方案，比如拆分 join 或者用 bucket join。坑三：动态资源分配要慎用。动态资源分配（dynamic allocation）能让 Spark 根据负载自动增减 executor，听起来很美，但在某些场景下会导致资源震荡。比如作业前期需要大量资源，后期需求骤降，executor 被回收后，如果又有新任务，重新申请 executor 需要时间，反而拖慢作业。我一般只在多租户共享集群上开启动态分配，独占集群上关闭。坑四：数据本地性不能忽视。如果 task 和数据的本地性级别是ANY，说明数据需要跨节点传输，网络开销会很大。理想情况下应该是NODE_LOCAL或PROCESS_LOCAL。如果本地性差，可以适当增加spark.locality.wait的时间，让 Spark 多等一会儿，尽量在数据所在节点启动 task。坑五：不要忽略 driver 端的内存。很多人只关注 executor 内存，忽略了 driver。如果作业里有 collect、broadcast、或者 driver 端做复杂计算，driver 内存不足会直接导致作业失败。我一般把 driver 内存设成 4GB 到 8GB，特殊情况到 16GB。5.4 性能优化的检查清单每次优化完，我会对照这个清单检查一遍，确保没有遗漏：数据源是否做了列裁剪和分区裁剪是否用了合适的文件格式（Parquet/ORC 优于 Text/CSV）小文件是否合并到合理大小（128MB 到 256MB）join 是否优先用了 broadcast聚合是否用了 reduceByKey 而非 groupByKey是否存在数据倾斜，是否处理了shuffle 分区数是否合理，是否开启了 AQE序列化是否用了 Kryo资源配置是否匹配作业特点缓存是否合理使用并及时释放是否有不必要的 collect 或 driver 端计算GC 时间占比是否在可接受范围（低于 10%）这个清单不是每次都要全部过一遍，但至少前六项是必查的。大部分性能问题，前六项就能解决。6. 一些实战中的个人体会Spark 性能优化这件事，说到底是一个“理解数据、理解业务、理解框架”的过程。工具和参数是表象，真正决定优化效果的，是你对数据分布的理解和对作业逻辑的把握。我个人的习惯是，每接手一个新作业，先不急着调参数，而是花时间把数据摸清楚：数据量多大、key 分布如何、有没有倾斜、增长趋势怎样。这些信息比任何调优技巧都重要。很多时候，一个简单的数据预处理或者逻辑调整，比调一百个参数都管用。另外，优化是一个持续的过程，不是一次性的任务。数据在变、业务在变、集群在变，今天的最优配置明天可能就不是了。所以我会定期回顾作业的性能指标，发现异常及时处理，而不是等到作业跑不动了才去救火。最后分享一个小技巧：如果你不确定某个优化有没有效果，就做 A/B 测试。保持其他条件不变，只改一个变量，跑两遍对比结果。这样虽然麻烦一点，但能让你清楚地知道每个优化的实际收益，避免盲目调参。我自己的经验是，数据层面和算子层面的优化，收益通常最大；资源配置和底层调优，收益有限但能提升稳定性。把精力花在刀刃上，比什么都调一遍要高效得多。
1. 为什么 Spark 性能优化是个绕不开的坎干过大数据的人都有一个共识：Spark 作业能跑起来不难，难的是跑得又快又稳。我见过太多团队，集群规模不小，机器配置也不差，但一个 ETL 任务动辄跑两三个小时，资源利用率却低得可怜。更让人头疼的是，有些作业平时看着还行，一到数据量翻倍就直接 OOM 或者长尾拖死，整个调度链路跟着遭殃。Spark 性能优化这件事，本质上是在三个维度上找平衡：CPU 利用率、内存利用率和 IO 吞吐。任何一个维度成为瓶颈，整个作业的效率就被它锁死。而大部分性能问题，根源不在 Spark 本身，而在于数据分布、分区策略、算子选择和资源配置这几个环节没有对齐业务特征。这篇文章适合谁看？如果你已经写过 Spark 作业，能看懂 DAG 图，知道 RDD 和 DataFrame 的区别，但在面对“作业跑得慢”“内存老是不够”“任务倾斜严重”这些问题时还靠试参数来碰运气，那这篇内容就是给你准备的。我会从设计思路、核心细节、实操过程到问题排查，把 Spark 性能优化这件事拆开讲透，每个结论都尽量给出背后的计算逻辑和判断依据，让你不只是抄配置，而是真正理解为什么这么配。2. 整体优化思路与方案选型拆解2.1 先定位瓶颈，再谈优化很多人一上来就问“Spark 怎么调优”，这个问题本身就没法回答。因为优化的前提是知道瓶颈在哪。Spark 作业的瓶颈大致分四类：CPU 密集、内存密集、IO 密集和网络密集。不同类型的瓶颈，优化手段完全不同，甚至互相矛盾。举个例子，如果你的作业卡在 shuffle 阶段，那加大 executor 内存可能有用；但如果瓶颈是 CPU 算力不足，加内存就是浪费资源。所以我的习惯是，拿到一个慢作业，先做三件事：看 Spark UI 的 Stage 详情：重点关注每个 Stage 的 duration、task 数量、shuffle read/write 大小、GC 时间占比。看 task 的时间分布：如果 max task 时间和 median task 时间差距超过 3 倍，基本可以判定存在数据倾斜。看资源等待时间：如果 task 的 scheduler delay 很高，说明资源调度跟不上，需要调整资源配置或动态分配策略。这三步做完，瓶颈基本就浮出水面了。下面这张表是我常用的瓶颈判断速查表：现象大概率瓶颈优先排查方向GC 时间占比超过 10%内存不足或对象过多executor 内存、序列化方式、数据结构某几个 task 特别慢数据倾斜key 分布、分区策略、join 方式shuffle read 巨大shuffle 分区不合理分区数、聚合方式、broadcast 可行性CPU 利用率低但作业慢IO 或调度瓶颈数据源读取方式、并行度、资源分配频繁 spill内存不够或分区过大分区粒度、聚合内存比例2.2 优化的优先级排序Spark 优化的手段很多，但收益和成本差异巨大。我一般按这个优先级来排：第一优先级：数据层面优化。包括分区裁剪、谓词下推、列裁剪、数据压缩。这些优化几乎零成本，收益却最直接。比如一个 Parquet 表有 200 个字段，你只 select 其中 10 个，列裁剪就能把 IO 降低 95%。很多作业慢，就是因为读了一堆根本用不上的数据。第二优先级：算子与逻辑优化。比如用 reduceByKey 替代 groupByKey，用 broadcast join 替代 shuffle join，用 mapPartitions 替代 map。这些改动需要理解业务逻辑，但收益通常很显著。第三优先级：资源配置优化。调整 executor 数量、内存大小、core 数量、并行度。这一步需要结合集群实际情况，参数之间还有耦合关系，不能孤立地调。第四优先级：底层调优。比如序列化方式、GC 策略、堆外内存。这些属于精细调优，收益有限但风险较高，一般在前三步做完之后再考虑。注意：不要一上来就调 GC 参数或者换序列化器。我见过太多人花了两天调 JVM 参数，结果发现真正的瓶颈是数据倾斜，一个 salting 就解决了。2.3 不同场景下的方案取舍Spark 作业大致分三类场景，每类的优化重点不同：批处理 ETL 场景：这类作业通常数据量大、逻辑复杂、对延迟不敏感。优化重点是减少 shuffle、提高并行度、合理分区。broadcast join 和分区裁剪是主力手段。交互式查询场景：这类作业对响应时间敏感，通常数据量中等。优化重点是缓存策略、AQE（自适应查询执行）、以及合理的文件布局（小文件合并）。流处理场景：这类作业对延迟和稳定性要求高。优化重点是背压处理、状态管理、以及 checkpoint 频率。微批大小和 trigger 间隔需要仔细权衡。理解自己作业属于哪一类，才能把有限的优化精力花在刀刃上。3. 核心细节解析与实操要点3.1 分区策略：性能的地基分区是 Spark 性能优化的地基，分区没搞好，后面怎么调都是事倍功半。分区的核心目标是：让每个 task 处理的数据量尽量均匀，且大小适中。分区太大，单个 task 内存压力大，容易 spill 甚至 OOM；分区太小，task 数量过多，调度开销和 GC 开销上来了。那多少算合适？我的经验值是每个分区 128MB 到 256MB。这个数字不是拍脑袋来的，它和 HDFS block 大小、Spark 默认的 shuffle 分区处理能力都有关系。计算分区数的公式大致是：理想分区数 = 数据总量 / 每分区目标大小比如你有 100GB 数据要处理，目标每分区 200MB，那理想分区数就是 500 左右。但实际中还要考虑 core 的数量，分区数最好是 core 总数的 2 到 4 倍，这样能保证每个 core 都有活干，又不至于调度太频繁。对于 shuffle 分区，spark.sql.shuffle.partitions这个参数默认是 200，这个默认值在数据量小的时候还行，数据量一大就完全不够用。我一般会把它设成executor 数量 × 每 executor 的 core 数 × 2到× 4。但更好的做法是开启 AQE，让 Spark 自己根据 shuffle 数据量动态合并分区。# 开启自适应查询执行 spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "200MB") spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")AQE 的好处是它能在运行时根据实际数据量调整分区，避免了你手动估算的误差。实测下来，开启 AQE 后，很多作业的 shuffle 分区数能自动从 200 调整到几千，性能提升非常明显。3.2 数据倾斜：最顽固的性能杀手数据倾斜是 Spark 性能问题里最顽固的一个。它的表现是：绝大多数 task 几秒钟就跑完了，但少数几个 task 跑了几十分钟甚至几个小时。原因通常是某个 key 的数据量远超其他 key。判断数据倾斜的方法很简单，在 Spark UI 里看 Stage 的 task 时间分布，如果 max 和 median 差距超过 3 倍，基本就是倾斜了。更精确的做法是采样统计 key 的分布：# 采样统计 key 分布 sample_df = df.sample(fraction=0.1) key_counts = sample_df.groupBy("join_key").count().orderBy("count", ascending=False) key_counts.show(20)如果发现某个 key 的数量级远超其他，那就是倾斜的源头。解决数据倾斜有几种常用手段：方案一：加盐打散。给倾斜的 key 加上随机前缀，把它打散到多个分区，聚合完再去掉前缀。这个方法适合聚合场景。import pyspark.sql.functions as F # 加盐：给 key 加上 0 到 N 的随机前缀 salt_range = 10 salted_df = df.withColumn("salted_key", F.concat(F.col("join_key"), F.lit("_"), (F.rand() * salt_range).cast("int"))) # 聚合后再去掉盐 result = salted_df.groupBy("salted_key").agg(...) \ .withColumn("join_key", F.split("salted_key", "_")[0]) \ .groupBy("join_key").agg(...)方案二：broadcast join。如果倾斜发生在 join 阶段，且其中一张表足够小（默认 10MB 以内，可调），直接广播小表，避免 shuffle。from pyspark.sql.functions import broadcast result = large_df.join(broadcast(small_df), "join_key")方案三：拆分倾斜 key 单独处理。把倾斜的 key 和非倾斜的 key 分开处理，倾斜的部分单独优化，最后 union 回去。这个方法最灵活，但代码复杂度也最高。实操心得：加盐的盐值范围不是越大越好。盐值太大，会导致分区数暴增，调度开销反而拖慢作业。我一般从 10 开始试，观察 task 时间分布，如果还不够均匀再往上加。另外，加盐只适合聚合场景，如果是 join 场景，需要两边都加盐，逻辑会复杂很多。3.3 算子选择：小改动大收益Spark 的算子选择对性能影响极大，但很多人写代码时并不在意。我挑几个最常见的对比讲一下。reduceByKey vs groupByKey：这是最经典的对比。groupByKey 会把所有相同 key 的 value 拉到一起，不做预聚合，网络传输量大；reduceByKey 会在 map 端先做本地聚合，再 shuffle，网络传输量小得多。同样逻辑，reduceByKey 通常比 groupByKey 快几倍。join vs broadcast join：普通 join 需要 shuffle 两张表，broadcast join 把小表广播到每个 executor，避免了大表的 shuffle。如果小表能放进内存，broadcast join 是首选。map vs mapPartitions：map 对每一行调用一次函数，mapPartitions 对每个分区调用一次。如果函数里有创建连接、初始化资源这类操作，mapPartitions 能大幅减少开销。# 不推荐：每行都创建一次连接 def process_row(row): conn = create_connection() # 每行都创建，开销巨大 return conn.query(row) df.rdd.map(process_row) # 推荐：每个分区创建一次连接 def process_partition(rows): conn = create_connection() # 每个分区只创建一次 for row in rows: yield conn.query(row) df.rdd.mapPartitions(process_partition)collect vs take：collect 把所有数据拉到 driver，数据量大直接 OOM；take 只取前 N 条，安全得多。调试时用 take，别用 collect。3.4 内存管理与序列化Spark 的内存模型分三块：execution memory、storage memory 和 user memory。execution memory 用于 shuffle、join、sort 等操作，storage memory 用于缓存 RDD 和 broadcast 数据，user memory 用于用户自定义的数据结构和函数。默认情况下，execution 和 storage 共享一块统一内存区域，比例是 50% 对 50%，可以通过spark.memory.fraction和spark.memory.storageFraction调整。如果作业 shuffle 很重，可以适当提高 execution 的比例；如果缓存了很多数据，就提高 storage 的比例。序列化方式对内存和网络传输影响很大。Spark 默认用 Java 序列化，慢且占空间。换成 Kryo 序列化，通常能减少 2 到 5 倍的内存占用，速度也快不少。spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") spark.conf.set("spark.kryoserializer.buffer.max", "256m")但 Kryo 需要注册自定义类才能发挥最佳效果，否则每个对象都要写入类名，反而浪费空间：spark.conf.set("spark.kryo.registrationRequired", "true") # 通过 SparkConf 注册自定义类 conf.registerKryoClasses([MyClass1, MyClass2])注意：开启 registrationRequired 后，未注册的类会直接报错。这个开关适合在测试环境先跑一遍，把用到的类都注册好再上生产。我一般会先用非严格模式跑，从日志里收集需要注册的类，再开启严格模式。4. 实操过程与核心环节实现4.1 一个真实慢作业的优化全过程我拿一个典型的 ETL 作业来演示完整的优化过程。这个作业的逻辑是：从两张 Hive 表读取数据，做 join 和聚合，然后写入结果表。原始配置是默认的，跑一次要 90 分钟。第一步：定位瓶颈。打开 Spark UI，发现作业分三个阶段，其中 Stage 2 的 shuffle read 高达 800GB，且 task 时间分布极不均匀，max task 跑了 45 分钟，median 只有 2 分钟。这明显是数据倾斜加 shuffle 分区不合理。第二步：检查数据分布。对 join key 做采样统计，发现有一个 key 占了总数据量的 40%，这就是倾斜的根源。第三步：优化 join 方式。检查两张表的大小，发现其中一张表只有 50MB，完全可以 broadcast。改成 broadcast join 后，shuffle 直接消失了。# 优化前：普通 join，触发 shuffle result = large_df.join(small_df, "user_id") # 优化后：broadcast join，无 shuffle result = large_df.join(broadcast(small_df), "user_id")第四步：优化聚合。聚合阶段用的是 groupByKey，改成 reduceByKey 后，网络传输量减少了 70%。第五步：调整资源配置。原来每个 executor 配了 4 个 core 和 8GB 内存，core 太多导致每个 core 分到的内存少，频繁 GC。改成每个 executor 2 个 core 和 16GB 内存，GC 时间从 15% 降到 3%。第六步：开启 AQE。让 Spark 自动合并小分区，shuffle 分区数从固定的 200 调整到动态的 1500 左右。优化后，作业时间从 90 分钟降到 18 分钟，资源消耗还降低了 30%。这个案例说明，大部分性能问题不需要复杂的底层调优，把数据层面和算子层面的事情做好，收益就已经非常可观了。4.2 资源配置的计算方法资源配置是很多人头疼的问题，因为参数太多，而且互相影响。我分享一套自己的计算方法，虽然不能保证最优，但至少不会犯大错。第一步：确定 executor 的 core 数。经验值是每个 executor 配 4 到 5 个 core。core 太少，executor 数量多，调度开销大；core 太多，每个 core 分到的内存少，GC 压力大。我一般从 4 开始试。第二步：确定 executor 的内存。每个 core 配 4 到 8GB 内存。比如 4 个 core 的 executor，配 16GB 到 32GB 内存。但要注意，executor 内存不能超过节点物理内存的 80%，要给操作系统和其他进程留空间。第三步：确定 executor 数量。总 core 数 = executor 数量 × 每 executor 的 core 数。总 core 数一般不超过集群总 core 数的 80%，留一些给 driver 和其他作业。第四步：确定 driver 内存。如果作业里有 collect、broadcast 大表、或者 driver 端做复杂计算，driver 内存要相应加大。一般 4GB 到 8GB 够用，特殊情况可以到 16GB。第五步：确定并行度。并行度 = 总 core 数 × 2 到 4。这个并行度决定了同时运行的 task 数量，太少资源利用不足，太多调度开销大。举个例子，一个 10 节点的集群，每节点 32 core、128GB 内存。我的配置方案是：参数值计算依据executor cores4经验值，平衡调度和 GCexecutor memory20GB每 core 5GB，留出 overheadexecutor 数量1610 节点 × 每节点 1.6 个，取整总 core 数6416 × 4，占集群 80% 的 80%driver memory8GB有 broadcast 操作，适当加大并行度19264 × 3这套配置不是最优解，但能保证作业稳定运行，不会因为资源问题频繁失败。实际中还需要根据作业特点微调。4.3 文件布局与小文件治理数据源的存储格式和文件布局对 Spark 性能影响巨大。我见过一个作业，逻辑很简单，但跑得特别慢，最后发现是数据源有几十万个小文件，每个文件只有几 KB。Spark 读取时，每个小文件都要单独打开、读取、关闭，IO 开销完全压过了计算开销。解决小文件问题有两个方向：写入时控制文件大小和读取时合并小文件。写入时，可以通过repartition或coalesce控制输出文件数量：# 控制输出文件数量，每个文件约 128MB target_file_size = 128 * 1024 * 1024 # 128MB total_size = df.rdd.map(lambda x: len(str(x))).sum() # 估算总大小 num_files = max(1, int(total_size / target_file_size)) df.repartition(num_files).write.parquet("output_path")读取时，可以开启小文件合并：spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728") # 128MB spark.conf.set("spark.sql.files.openCostInBytes", "4194304") # 4MBmaxPartitionBytes控制单个分区读取的最大字节数，openCostInBytes告诉 Spark 打开一个文件的成本，Spark 会根据这两个参数把多个小文件合并到一个分区里读取。实操心得：小文件治理最好在数据写入阶段就做好，不要等到读取时再补救。我一般会在 ETL 的最后一步加一个 repartition，确保输出文件大小在 128MB 到 256MB 之间。另外，如果数据源是 Hive 表，可以用ALTER TABLE ... CONCATENATE或者INSERT OVERWRITE重写来合并小文件。4.4 缓存与持久化策略缓存是 Spark 里一把双刃剑。用得好，能避免重复计算，大幅提速；用不好，占满内存，导致 spill 甚至 OOM。缓存的适用场景是：同一个 DataFrame 被多次使用，且计算成本高。比如一个 DataFrame 经过复杂的 join 和聚合后，被用于多个下游计算，这时候缓存就很有价值。# 缓存中间结果 intermediate_df = complex_transformation(input_df) intermediate_df.cache() # 或者 persist(StorageLevel.MEMORY_AND_DISK) # 多次使用 result1 = intermediate_df.filter(...).count() result2 = intermediate_df.groupBy(...).agg(...)存储级别有几种选择：存储级别说明适用场景MEMORY_ONLY只存内存，存不下就重算数据量小，内存充足MEMORY_AND_DISK内存优先，存不下写磁盘通用场景，推荐MEMORY_ONLY_SER序列化后存内存内存紧张，CPU 充足DISK_ONLY只存磁盘数据量极大，内存严重不足我一般用 MEMORY_AND_DISK，兼顾性能和稳定性。如果内存特别紧张，用 MEMORY_ONLY_SER，序列化后能省不少内存，但读取时要反序列化，CPU 开销会上升。注意：缓存用完记得 unpersist，否则会一直占着内存。我见过一个作业，缓存了五六个 DataFrame 都没释放，最后 executor 内存被占满，新任务全部 spill 到磁盘，性能反而比不缓存还差。5. 常见问题与排查技巧实录5.1 常见问题速查表问题现象可能原因排查方法解决方案作业 OOM分区太大、缓存过多、数据倾斜看 Spark UI 的 Storage 和 Executor 页减小分区、释放缓存、处理倾斜GC 时间过长内存不足、对象过多、序列化不当看 Executor 页的 GC 时间占比加大内存、换 Kryo、减少对象创建任务长尾数据倾斜、分区不均看 task 时间分布加盐、broadcast、拆分 keyshuffle 数据量巨大分区不合理、算子选择不当看 Stage 的 shuffle read/write调整分区、用 reduceByKey作业卡在某个 Stage资源不足、调度延迟看 scheduler delay调整资源配置、开启动态分配频繁 spill内存不够、聚合数据量大看 spill 指标加大内存、提高并行度、预聚合读取数据慢小文件多、格式不合适看 input size 和 task 数合并小文件、换列式存储5.2 排查工具与技巧Spark UI 是首选工具。重点看几个地方：Stage 页的 task 时间分布、Executor 页的 GC 时间和内存使用、SQL 页的物理计划。物理计划里能看到 join 方式、分区数、过滤条件下推情况，是定位问题的关键。日志分析。executor 日志里会记录 spill、GC、shuffle 的详细信息。我一般会 grep 几个关键词：spill、GC、OutOfMemory、FetchFailed。这些关键词出现的地方，往往就是问题所在。采样调试。对于大数据量的作业，直接跑全量太慢。我习惯先用sample取 1% 到 10% 的数据跑一遍，观察逻辑是否正确、性能瓶颈在哪，确认没问题再跑全量。# 采样调试 sample_df = df.sample(fraction=0.01, seed=42) sample_df.write.mode("overwrite").parquet("debug_path")分阶段验证。把复杂作业拆成几个阶段，每个阶段单独跑，观察每个阶段的耗时和资源消耗。这样能精确定位到哪个阶段是瓶颈。5.3 独家避坑经验坑一：不要盲目加大并行度。并行度不是越大越好。并行度太大，task 数量暴增，调度开销和 GC 开销会拖慢作业。我见过有人把并行度设成几万，结果作业比原来还慢。并行度控制在总 core 数的 2 到 4 倍就够了。坑二：broadcast join 不是万能的。broadcast 的表必须足够小，默认阈值是 10MB，超过这个值会报错或者退化成普通 join。如果小表有几百 MB，强行 broadcast 会导致 driver OOM。这时候应该考虑其他方案，比如拆分 join 或者用 bucket join。坑三：动态资源分配要慎用。动态资源分配（dynamic allocation）能让 Spark 根据负载自动增减 executor，听起来很美，但在某些场景下会导致资源震荡。比如作业前期需要大量资源，后期需求骤降，executor 被回收后，如果又有新任务，重新申请 executor 需要时间，反而拖慢作业。我一般只在多租户共享集群上开启动态分配，独占集群上关闭。坑四：数据本地性不能忽视。如果 task 和数据的本地性级别是ANY，说明数据需要跨节点传输，网络开销会很大。理想情况下应该是NODE_LOCAL或PROCESS_LOCAL。如果本地性差，可以适当增加spark.locality.wait的时间，让 Spark 多等一会儿，尽量在数据所在节点启动 task。坑五：不要忽略 driver 端的内存。很多人只关注 executor 内存，忽略了 driver。如果作业里有 collect、broadcast、或者 driver 端做复杂计算，driver 内存不足会直接导致作业失败。我一般把 driver 内存设成 4GB 到 8GB，特殊情况到 16GB。5.4 性能优化的检查清单每次优化完，我会对照这个清单检查一遍，确保没有遗漏：数据源是否做了列裁剪和分区裁剪是否用了合适的文件格式（Parquet/ORC 优于 Text/CSV）小文件是否合并到合理大小（128MB 到 256MB）join 是否优先用了 broadcast聚合是否用了 reduceByKey 而非 groupByKey是否存在数据倾斜，是否处理了shuffle 分区数是否合理，是否开启了 AQE序列化是否用了 Kryo资源配置是否匹配作业特点缓存是否合理使用并及时释放是否有不必要的 collect 或 driver 端计算GC 时间占比是否在可接受范围（低于 10%）这个清单不是每次都要全部过一遍，但至少前六项是必查的。大部分性能问题，前六项就能解决。6. 一些实战中的个人体会Spark 性能优化这件事，说到底是一个“理解数据、理解业务、理解框架”的过程。工具和参数是表象，真正决定优化效果的，是你对数据分布的理解和对作业逻辑的把握。我个人的习惯是，每接手一个新作业，先不急着调参数，而是花时间把数据摸清楚：数据量多大、key 分布如何、有没有倾斜、增长趋势怎样。这些信息比任何调优技巧都重要。很多时候，一个简单的数据预处理或者逻辑调整，比调一百个参数都管用。另外，优化是一个持续的过程，不是一次性的任务。数据在变、业务在变、集群在变，今天的最优配置明天可能就不是了。所以我会定期回顾作业的性能指标，发现异常及时处理，而不是等到作业跑不动了才去救火。最后分享一个小技巧：如果你不确定某个优化有没有效果，就做 A/B 测试。保持其他条件不变，只改一个变量，跑两遍对比结果。这样虽然麻烦一点，但能让你清楚地知道每个优化的实际收益，避免盲目调参。我自己的经验是，数据层面和算子层面的优化，收益通常最大；资源配置和底层调优，收益有限但能提升稳定性。把精力花在刀刃上，比什么都调一遍要高效得多。