上个月我们团队接到一个棘手的任务：核心报表链路每天跑一次，耗时从年初的半小时慢慢涨到了两个多小时，业务方每天早上都在群里催数。一开始大家习惯性想加资源，但加了executor、扩了队列，效果也就那样，成本倒是实打实上去了。后来我从头到尾把这条链路翻了一遍，从数据分布、执行计划、框架参数到存储格式逐一排查，最后把整条作业链压缩到25分钟左右，跑完还能把资源让回给别的团队。这篇文章就把这套分布式计算框架优化的思路和操作细节整理出来，希望能帮到正在被慢作业折磨的工程师。
这篇文章不聊虚的，重点讲清楚几件事：怎么快速定位瓶颈、数据倾斜怎么治、资源参数怎么配才合理、SQL怎么写才不浪费计算资源、小文件怎么合并，以及一整条链路优化的先后顺序。适合用Spark、Hive、Flink做离线或实时计算的同学参考，尤其是那种"作业能跑但越跑越慢"的团队。
1. 先定位瓶颈：优化动作前必须搞清的问题
很多人一上来就调参数、改SQL，结果改了一通不知道到底改对了没有。我自己的经验是：先花半小时把执行链路看清楚，尤其把瓶颈定位准确，后面改起来才有方向，改完才知道是不是真的生效。
1.1 Spark UI里最该看的三块信息
无论你用Spark、Flink还是跑在Yarn上的MapReduce，第一件事都是打开作业的UI页面。以Spark为例，重点看三块：
Stage列表：每个Stage的耗时和Shuffle读写量。如果Stages很多、依赖链很长，先用DAG图看整体结构，找出哪几个Stage占据了绝大部分耗时。
Task耗时分布：点进某个耗时的Stage后，观察所有Task的耗时。如果大量Task在秒级完成，极少数Task要跑几十分钟，基本可以断定是数据倾斜；如果所有Task耗时都比较均匀地长，那可能是数据量大、资源不足或者计算逻辑本身太重。
Executor的GC时间与内存曲线：如果Event Timeline上能看到频繁的Full GC，并且GC时间占比超过10%，说明内存堆配置可能有问题，或者某个算子把大量数据堆积在内存里。
举个例子，我接手的那条链路里有一个Join Stage，UI上显示6000个Task，其中5900个在1到3秒内完成，剩下100个都在10分钟以上。这时候看Executor的CPU利用率，整体其实不高，问题明显不是资源不够，而是少数Task拖慢了整个Stage。
1.2 不同瓶颈的判读标准
定位瓶颈时，最怕的就是把"数据倾斜"和"资源不足"搞混。下面的表格是我平时做快速判读用的，分享出来供参考：
| 现象 | 优先怀疑方向 | 验证方法 |
|---|---|---|
| 少量Task耗时远超中位数，且这些Task处理的数据量（如Shuffle Read Bytes）明显偏大 | 数据倾斜 | 检查Stage输入数据的Key分布 |
| 所有Task耗时同步变长，且CPU/内存利用率高 | 资源不足或数据量整体增长 | 看集群当前是否有其他作业抢占资源 |
| Task耗时不高，但Stage总数很多，依赖链长 | 执行计划不合理 | 用explain看逻辑计划和物理计划 |
| GC时间占比高，任务反复出现内存溢出 | 内存参数或算子内存占用问题 | 调整Executor内存与spark.memory参数 |
| 磁盘IO或网络IO接近饱和 | 基础设施瓶颈或Shuffle量过大 | 查看节点监控，确认数据本地性 |
这一步骤的核心是"用数据说话"。不要凭感觉觉得某个SQL慢就是Join慢，先看Stage和Task分布，再看Shuffle量，顺藤摸瓜找到真正的问题点。
2. 数据倾斜治理：分布式计算里性价比最高的优化点
数据倾斜是分布式计算框架优化中最常见也最典型的性能杀手。我见过的生产事故里，十个慢作业里有六七个都跟数据倾斜有关。
2.1 倾斜是怎么发生的
分布式计算的核心思想是把数据分到多个Task并行处理。Shuffle阶段会按照Key的哈希值对数据进行分区，落到不同Task上。问题在于，如果某个Key在数据集中占据的份额过大，比如某个热门商品ID的订单量是其他商品的上百倍，那个处理该Key的Task就要比别人多算几十倍的数据，于是整个Stage的耗时被它所拖累。
常见的倾斜Key一般有这么几类：空字符串、默认值（比如“default”、“unknown”）、特定业务ID（如测试账号、超大店铺），这些Key在业务上往往没什么聚合价值，却在计算时造成了严重倾斜。
2.2 三个立竿见影的治理方案
方案一：过滤或剔除无效Key
如果倾斜来自空值或默认值，且这些数据本身对结果无意义，最简单的方式就是提前过滤。比如统计订单量时，把user_id为空或为默认值的数据直接去掉：
SELECT user_id, COUNT(*) AS cnt FROM orders WHERE user_id IS NOT NULL AND user_id != '' GROUP BY user_id;方案二：大表Join小表时用广播
业务里最常见的是大表关联维度表。维度表往往只有几千到几万行，但是无论表多小，如果不做任何优化，Spark默认都会走Shuffle。正确的做法是让Executor把小表加载到内存里，直接在每个Task内部完成关联，彻底避免Shuffle。
用Spark SQL可以这样：
SELECT /*+ BROADCAST(dim) */ a.id, b.name, a.amount FROM fact_table a JOIN dim_table b ON a.dim_id = b.id;也可以动态判断表的大小，Spark 3.x中如果一张表小于spark.sql.autoBroadcastJoinThreshold（默认10MB），会自动走BroadcastJoin，不需要额外加Hint。但要注意，当两张参与Join的表都很大时，这条路走不通。
方案三：加盐拆分+两阶段聚合
如果倾斜的Key既有业务价值又无法过滤，最常用的手段是加盐（Salting）。原理是把一个倾斜的Key加上随机前缀拆成多个子Key，让数据分散到不同Task上，先做一轮局部聚合，再去掉前缀做全局聚合。
# 伪代码：两阶段聚合 from pyspark.sql import functions as F, Window # 第一轮：倾斜key加随机后缀，打散 df_expanded = skewed_df.withColumn( "salt_key", F.concat(F.col("key"), F.lit("_"), F.rand() * 10) ) partial_agg = df_expanded.groupBy("salt_key").agg(F.sum("amount").alias("partial_amount")) # 第二轮：去盐后做全局聚合 final_result = ( partial_agg.withColumn("raw_key", F.split("salt_key", "_")[0]) .groupBy("raw_key") .agg(F.sum("partial_amount").alias("total_amount")) )加盐粒度决定了效果：盐分越多，倾斜Key被打散得越均匀，但第二轮聚合时需要重读一轮数据。实际调优时我一般先尝试10到100之间的盐值，观察Task耗时分布，再逐步调整。
2.3 治理效果怎么验证
改完之后不要只看作业总耗时，还是回到Stage的Task耗时分布图去看。如果之前有Task跑30分钟，现在所有Task都在3分钟内完成，说明倾斜确实被解决了。另外要留意Shuffle Read Bytes的变化，确认是不是真的把某一批超大Key的数据拆开了，而不是把问题从某个Task转移到了另一个Task。
3. 资源参数与调度策略：从“加机器”到“用对参数”
在没定位瓶颈之前盲目加资源，是分布式计算框架优化中最容易踩的坑。但反过来，当确认了作业确实存在资源不足时，怎么配置参数又成了另一门学问。
3.1 Executor数量、内核与内存的配比逻辑
Executor的数量和规格，决定了作业有多少并行度、多少内存可用。这里容易犯的毛病是“越多越好”。实际上，Executor内存过大或过小都会出问题：内存过大，GC时停顿时间也长；内存过小，数据在内存里放不下，频繁落盘。
我常用的配比参考如下（假设单节点可用资源比较充裕）：
| 场景 | 每个Executor核数 | 每个Executor内存 | 说明 |
|---|---|---|---|
| 中小型ETL作业 | 2 | 4GB | 避免浪费资源，任务调度更细 |
| 数据量大但逻辑简单 | 4 | 8GB | 平衡吞吐与GC |
| Join/聚合密集 | 4 | 12GB以上 | 预留Shuffle读缓存 |
同时要关注Spark内存模型。Executor内存被分为执行内存和存储内存，通过spark.memory.fraction（默认0.6）控制。如果作业Shuffle量很大，可以适当调高执行内存的比例；如果Cache的表很多，则维持默认或调低。具体数值要结合GC表现来验证，不要照抄网上的配置。
3.2 Shuffle并行度与数据量的匹配
Shuffle并行度是个经常被忽视但影响巨大的参数。Spark里有两个关键参数：
spark.sql.shuffle.partitions：Spark SQL执行Shuffle时默认分区数，默认值是200spark.default.parallelism：RDD操作默认的并行度
如果一份数据量很大，但并行度没提上去，相当于每个Task处理的数据量过大；反过来并行度太高，又会增加调度和元数据开销。比较通用的经验是：让每个分区处理的数据量在128MB到256MB之间。
举例来说，如果Shuffle读的总数据量是200GB，那么把spark.sql.shuffle.partitions设为1000到1500之间比较合理，每个分区大概处理130到200MB。
-- 通过spark-submit参数设置 spark-submit --conf spark.sql.shuffle.partitions=1200 ...3.3 队列与动态资源分配的实际取舍
在生产集群里，多团队共用一个资源池很常见。如果一个大作业把队列打满，其他团队的作业全被卡死，这种“优化”得不偿失。
我建议开启动态资源分配，让作业根据实际任务量伸缩Executor：
spark.dynamicAllocation.enabled=truespark.dynamicAllocation.minExecutors=2spark.dynamicAllocation.maxExecutors=50spark.dynamicAllocation.executorIdleTimeout=60s
开启后，作业两头都有余量：数据量大时自动扩展，数据量小的时候自动回收。但要注意，动态分配和某些离线调度系统配合时，可能因为初始化过慢而导致延迟，需要在调度器层面做兼容。如果你们平台对资源有严格配额，也可以关闭动态分配，改用固定值，但一定要在业务低峰期运行大作业。
4. SQL代码层面的执行计划优化
参数调整和数据倾斜解决了大部分问题，但还有一种情况：框架运行“健康”，数据分布也均匀，可作业还是很慢。这时候要回到SQL本身，看执行计划。
4.1 用Explain看执行计划差异
无论Spark SQL还是Hive，跑之前都可以先看执行计划。Spark 3.x里执行计划分为逻辑计划和物理计划，重点看物理计划中的Join算法：
- BroadcastHashJoin：小表广播，无Shuffle，最快
- SortMergeJoin：两侧都需要Shuffle和排序，数据量大时必然出现
EXPLAIN SELECT /*+ BROADCAST(dim) */ a.id, b.name FROM fact a JOIN dim b ON a.dim_id = b.id;我在实际优化中看过不少案例，大表和小表Join因为小表稍微超过了广播阈值，结果走了SortMergeJoin，Shuffle量从2GB变成30GB，耗时翻了好几倍。遇到这类情况，要么给SQL加BroadcastHint，要么调大spark.sql.autoBroadcastJoinThreshold，但阈值不要设太大，否则小表广播时GC压力会增大。
4.2 谓词下推与列裁剪
执行计划优化的另一重点是“过滤器下推”。SQL里写WHERE条件时，框架不一定能保证第一时间过滤数据。下面是两个常见的处理原则：
- 先过滤再Join：把数据量大的表先做WHERE过滤，再做关联，让Shuffle的数据量尽量小
- 只取需要的列：避免
SELECT *，减少列裁剪的数据量
虽然Spark SQL的优化器会自动做谓词下推，但有些场景下还是需要手动创造条件。比如子查询里关联条件写的时机不对，或使用了UDF，都可能阻碍下推。
4.3 常见SQL写法坑
整理几个自己在生产环境遇到过高频问题：
Avoid GroupByKey类操作
在RDD API里，groupByKey会直接把同一个Key的所有Value集中到一个Task上再处理。哪怕数据不倾斜，GroupByKey的Shuffle量通常也比ReduceByKey大一截。能先聚合再传上游的，尽量先做一轮合并。
# 不好：groupByKey后逐条求和 rdd.groupByKey().mapValues(lambda vals: sum(vals)) # 好：reduceByKey直接在shuffle前合并 rdd.reduceByKey(lambda a, b: a + b)Count Distinct的陷阱
COUNT(DISTINCT col)在数据量巨大时，往往需要精确去重，Shuffle和内存开销非常高。如果业务上能接受近似值，用approx_count_distinct，会走HyperLogLog算法，性能和准确性之间的平衡很不错。
不等值连接
Join条件中如果包含a.x >= b.x这类不等值条件，无法走Broadcast Join，只能做笛卡尔积，慢是必然的。尽量改造成等值连接，或者分区间处理再合并。
执行计划优化没有银弹，关键是每改一个点就跑一遍Explain看差异，把SQL的执行路径和Shuffle数据量对比一下，才会越来越顺手。
5. 小文件治理与存储格式选择：被低估的稳定性问题
分布式计算框架优化的最后一环，往往不在计算本身，而在于计算之后写出来的数据。小文件问题初期不那么明显，但会慢慢拖垮整个链路。
5.1 小文件为什么可怕
一个100GB的数据集，如果以1MB的小文件存储，会有10万个文件。每个文件在读取时都要经过NameNode（HDFS）或对象存储的元数据查询，在高并发下元数据服务会成为瓶颈；同时，任务在调度时也会因为分片过多，产生大量空转Task。最直接的表现就是：作业启动时间变长了，处理同样的数据，效率却逐年下降。
造成小文件的常见场景包括：
- 动态分区插入时，每个分区写入的并发Task数过多
- 流式任务频繁写入同一个表
- Hive表经过多次INSERT OVERWRITE，但Spark/Hive没做合并
5.2 几种合并小文件的实操方案
方案一：写数据前控制分区数
如果业务上对写入的分区数量没有严格要求，可以在写入前先做一次重分区：
// 例如适合32个左右分区 df.repartition(32).write.mode("overwrite").partitionBy("dt").parquet(outputPath)方案二：定时合并任务
对于经常动态写入的表，写一个定时的合并作业，读取数据后重新写入。
spark.sql("SET spark.sql.shuffle.partitions=200") val df = spark.read.parquet(inputPath) df.coalesce(50).write.mode("overwrite").parquet(outputPath)这里coalesce适合在数据量不大时减少分区，但如果数据量很大，用repartition配合sortBy来保证合并后文件大小均匀。
方案三：借助小文件合并工具
现在很多数据平台都有专门的小文件合并工具，原理可以简单理解成：扫描小文件，按目标大小合并成更大的文件，然后更新表的元数据。有工具直接用工具，没有的话，参照上面的方案写个定期任务也能解决问题。
5.3 存储格式与压缩的选型建议
小文件合并完，还要选对存储格式。对于分布式计算框架来说，我强烈建议列式存储。列式存储（Parquet、ORC）在查询时只需读取需要的列，对分析型作业收益非常大；同时这类格式自带统计信息，可以做谓词下推、更高效的压缩，读同一份数据的IO开销能小很多。
压缩格式方面，生产环境我用得比较多的组合是：
| 存储格式 | 压缩 | 适用场景 |
|---|---|---|
| Parquet | Snappy | 通用分析、读取较多 |
| ORC | ZSTD | Hive作业多、追求更高压缩率 |
| Parquet | ZSTD | 数据量非常大且读取频繁 |
ZSTD的压缩率和解压速度都不错，但要注意不同版本的Spark对它的支持程度，老版本可能需要额外引入Jar包。选压缩格式时，不能一味追求“压缩率最高”，还要考虑压缩和解压的CPU开销。实际测试下来，Snappy在绝大多数场景下的综合表现最稳。
6. 优化链路实测：一套组合拳把作业从2小时压到25分钟
最后用一个完整的案例把前面的方法论串起来。这条链路当时是我们最核心的报表前置任务，主要做的是大表关联维表、按维度聚合、再写回结果表。
6.1 优化前的情况
优化前作业总耗时123分钟，整体状况如下：
| 检查项 | 实际表现 |
|---|---|
| Stage耗时Top1 | 一个Join Stage占87分钟 |
| Task耗时分布 | 100个Task超过30分钟，其余都在秒级完成 |
| Shuffle总量 | 36GB |
| 小文件数量 | 结果表写入了5000多个小文件 |
| 队列情况 | 固定Executor 100个，资源利用率很低 |
6.2 一步步做了什么
第一步：看执行计划
发现一个事实表和维度表的Join走了SortMergeJoin，而且Seen表过滤条件写得比较靠后，导致Shuffle数据量过大。
第二步：调整广播Join
维度表只有两万行左右，加上BROADCASTHint后，Join Stage的Shuffle量从36GB降到4GB。这一步直接把两个多小时的作业压缩到70分钟。
第三步：处理倾斜Key
聚合阶段发现部分维度出现倾斜，采用加盐两阶段聚合，将100个长尾Task控制在2分钟以内。这一步之后作业总耗时降到38分钟。
第四步：调整并行度和Executor规格
把不合理的Executor数量从100降到40，每个Executor从4核8GB调整为6核12GB，同时将spark.sql.shuffle.partitions从默认的200改为400，GC时间明显下降。
第五步：小文件合并与写回优化
重跑作业前检查了目标表目录，发现大量小文件。先做了历史数据合并，并在写入时采用repartition(64)控制写文件数量。
6.3 整个过程的耗时变化
| 阶段 | 作业总耗时 | 说明 |
|---|---|---|
| 优化前 | 123分钟 | 排序连接+倾斜+小文件 |
| 广播Join后 | 70分钟 | 消除大量Shuffle |
| 加盐聚合后 | 38分钟 | 消除了Task长尾 |
| 参数调整后 | 25分钟 | 并行度和内存更匹配 |
| 小文件合并 | 25分钟（稳定） | 写结果表更快、后续读取也更快 |
6.4 后续维护的经验
优化不是一次性的。业务数据分布会变化，新功能上线后SQL逻辑也常调整，最关键的是要建立“基线+告警”机制：给核心作业设定基线耗时，超过120%或者出现新增长尾Task就触发告警提醒。另外，我倾向于每季度对核心表的数据量、文件数、分区数做一次体检，把问题消灭在出现“明显变慢”之前。
如果团队条件允许，还可以把关键作业的Spark UI指标自动采集下来，比如每阶段的Shuffle量、Task耗时中位数、GC时间，存成历史报表。这样以后作业再变慢，可以直接对比历史数据，快速判断是新逻辑引入的问题，还是数据量自然增长导致的。
最后再分享一点自己的体会
做分布式计算框架优化这几个月，我最大的感受是：顺序很重要。先看数据分布，再看执行计划，然后才是参数调整和存储治理。如果一上来就动参数，很容易把问题掩盖在一堆变了的配置里，后面再排查反而更难。还有一个容易被忽略的点是，优化之前一定要记录基准数据，改完每一步都回看指标，否则你会分不清到底是因为哪一步变快了。希望这篇文章里的定位方法和组合拳能帮你在自己的集群上少走一些弯路。
