大数据架构性能优化：从数据倾斜到查询加速全攻略

大数据架构性能优化：从数据倾斜到查询加速全攻略

做大数据的人，几乎都经历过这种场景：凌晨跑的离线报表突然慢了十倍，打开Spark UI一看，一个大Stage卡在99%，一个Task在那里磨磨蹭蹭，其他几百个Task早就跑完了。你盯着屏幕，知道"数据倾斜"这个老大难又来找你了。这套流程我走了很多年，从Hive on MR到Spark再到Flink，从数据倾斜治理到查询加速，前前后后踩过的坑加起来比我掉的头发还多。这篇文章我不讲空泛的架构理论，就把这些年实测管用的经验、排障思路和可直接抄作业的配置整理出来，覆盖数据倾斜根治、存储布局优化、执行引擎调优、资源调度治理，以及几个生产环境真实故障的完整复盘。适合正在被大数据任务跑不动、集群利用率上不去、SLA天天告警折磨的开发和数据工程师。

1. 先搞懂数据倾斜：一场“一个人干活，一群人在等”的分布式困境

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

分布式计算框架的核心思想是“分而治之”：

先把数据切分成多个分片（partition），再分发到不同的节点上并行处理，最后汇总结果。在这个模型里，系统有一个默认假设：数据分布是均匀的。只要每个分片的数据量和计算量差不多，整个任务的执行时间就取决于最少节点的处理能力。

但现实世界的数据，很少是均匀的。订单表里，一个“默认渠道”可能占了80%的数据量；用户表里，一线城市的用户ID在活跃度上碾压其他城市；日志表里，空字符串、Null值、特殊占位符往往会集中在一个分组里。当这些“超级Key”被分发时，系统的“哈希分区”机制会把相同Key的所有记录全部路由到同一个分区，于是这个分区上的Task要处理的数据量是其他Task的几十倍甚至上百倍。

我习惯用仓库分拣来打比方：一个仓库有一百个分拣员，本来每个人分一车包裹，结果有一辆超大卡车拉来了全仓80%的货，而且司机只允许一个人分拣。于是整个仓库的效率就被这一个分拣员拖死了——这就叫数据倾斜。

具体到技术层面，三个最常见的触发点：

• Shuffle阶段：不管是group by还是join，框架需要确保相同Key的记录落到同一个Reducer/Executor，这是按Key的哈希值取模来决定分区号的。某个Key的记录数畸高，对应分区就畸重。
• Join操作：大表与小表Join，如果大表中某一个Key关联的记录特别多，这个Key所在分区的Join操作就会被无限放大。
• count(distinct)：去重操作如果只用一个Reducer，所有记录全部集中到一个节点做去重，这几乎是“人为制造”的最严重倾斜。

1.2 数据倾斜的连锁反应和判断标准

数据倾斜不是只让单个任务变慢，它是会传导的。一个Stage卡住，后续Stage全部等待，整个Job的执行时间被拉长。资源利用率也难看：明明集群有几百个核，大部分节点闲在那里，只有一两个节点读写磁盘打满，CPU和内存都被极少数Task消耗殆尽，其他Task完成后还得等着，白白占用资源。

更麻烦的是，倾斜Task往往会伴随**磁盘Spill（溢写）和内存GC（垃圾回收）**问题。数据量超过Executor内存上限后，框架会把数据临时溢写到磁盘，后续还要再读回来，频繁的序列化和反序列化带来大量的CPU开销。如果溢写还压不住，就会出现OOM（内存溢出），任务直接被Kill，甚至会拖垮整个节点上的其他Executor。

业务侧的反应更直接：报表出不来、SLA告警、下游依赖的任务堵成一串，数据仓库的产出时间被拉到不可接受的程度。

如何判断你的任务是不是真的发生了数据倾斜？我一般看三个信号：

1. 长尾分布：同一个Stage里，绝大多数Task在几十秒内完成，只有个别Task运行时间超过十分钟，甚至差别在一个数量级以上。
2. Task输入数据量差异巨大：UI里能直接看到每个Task的Shuffle Read/Write Size，倾斜的Task数据量往往是其他Task的5倍、10倍甚至50倍。
3. 进度百分比停留在99%：Spark UI中显示Stage完成度已经99%，但最后一个Task迟迟不结束，这个画面是数据倾斜的“标志性截图”。

2. 快速定位数据倾斜：三步找出“罪魁祸首Key”

2.1 从UI指标看“谁在拖后腿”

数据倾斜的定位，第一步不是看代码，而是看运行时的UI指标。不管是Spark的HistoryServer、Flink的Web UI，还是Hive on Tez的DAG视图，你都要训练自己在几分钟内判断出具体是哪个Stage、哪个Task歪得离谱。

看UI的时候，我建议按这个顺序来：

• 先看Stage列表：找到耗时最长的Stage，点进去。一个Job有多个Stage，倾斜一般出现在有Shuffle的Stage，也就是宽依赖（wide dependency）所在的阶段。
• 再看Task列表，按Duration倒序：把运行时间最长的几个Task列出来，对比它们的Shuffle Read Size。如果最长Task的读取量比其他Task大了一个数量级，倾斜基本坐实。
• 看GC Time和Spill：如果某个Task的GC时间占了运行时间的大半，或者Shuffle Spill磁盘溢写量明显偏大，说明这个Task手上的数据量已经超出了Executor的内存承受能力。
• 看“最大Task”的数据分布关键信息：UI不会直接告诉你倾斜的Key是什么，但它会告诉你哪个Task最重、读取的数据来自哪些上游分区。这个过程能帮我们把排查范围缩小到一个或几个Task。

如果是Flink任务，就看TaskManager的BackPressure（背压）状态：倾斜的分区会让对应Subtask持续高水位，其他Subtask被队头阻塞拖住，整个链路的吞吐随之下降。

2.2 用SQL快速统计Key分布，锁定超级Key

UI告诉我们“某个Task很重”，但要真正解决问题，必须找到那个“超级Key”。最快的方式是跑一个聚合查询，把数据按疑似倾斜的Key统计一遍，按数量倒序排列：

SELECT channel_id, COUNT(1) AS cnt FROM dwd_order_detail WHERE dt = '2024-06-15' GROUP BY channel_id ORDER BY cnt DESC LIMIT 50;

运行这个SQL的时间可能不短，因为它在全量数据上做group by，但它的价值在于一次性把分布情况暴露出来。如果看到第一个Key的cnt是几亿，而第二个只有几百万，那个“几亿”的Key就是罪魁祸首。

在超大表上，为了更快出结果，可以先做一个采样预估，而不是直接跑全表聚合：

SELECT channel_id, COUNT(1) * 100 AS approximate_cnt FROM ( SELECT * FROM dwd_order_detail WHERE dt = '2024-06-15' TABLESAMPLE(1 PERCENT) ) t GROUP BY channel_id ORDER BY approximate_cnt DESC LIMIT 50;

这里我做了1%的采样，把count结果乘以100得到一个近似值。数据均匀时这个预估很准，但要注意：如果倾斜Key本身占比很小，采样可能漏掉它。实操中我通常先全量跑一次group by，因为它本来就是排查任务，耗时可控——只要确保别跟正常生产任务抢太多资源。

2.3 区分“真倾斜”和“伪倾斜”

这一步新手特别容易忽略。有些任务慢，UI看起来也像倾斜，但本质上数据分配是均匀的，问题是出在单条记录的处理开销特别大上。我把它叫做“伪倾斜”。

典型场景：

• 单条超大字段：某条记录里存了一个几十MB的JSON或一个超长的文本字段。分区数据量看着均匀，但某个Task恰好处理了包含超大字段的记录，每条都要反复序列化、传输，处理时间自然被拉长。
• UDF计算不均匀：某个UDF（用户自定义函数）对特定数据的计算复杂度特别高，比如字符串正则匹配一个超长文本，或者调用外部API且刚好遇到超时重试。
• Join字段的类型不一致，触发隐式转换：一张表join字段是String，另一张是Long，引擎为了匹配类型会做转换，老版本引擎甚至会因为无法下推而产生大量额外的计算。

“伪倾斜”怎么区分？我常用的方法是对照实验：把疑似倾斜的Key过滤掉再跑一次同一个任务。如果任务时间恢复到了正常水平，说明确实是那个Key的数据量大导致的问题；如果时间还是慢，并且UI上Task的数据量看起来一样，那大概率是伪倾斜，得去检查单条数据的处理开销。

3. 数据倾斜治理：从临时救火到根治疗法

3.1 加盐+两阶段聚合：最经典的“均匀化”手段

定位到超级Key之后，最常用的根治方案是对Key做“加盐”处理，然后再分两阶段聚合。

原理并不复杂：既然某一个Key对应的数据量太大，那就人为把它拆散。给原始Key拼接一个随机数前缀（比如0到N之间的随机整数），原来一个超级Key就变成了一组大小差不多的子Key，每个子Key的数据量变小了，能均匀分布到不同分区做局部聚合。局部聚合完成后，去掉前缀，再对同Key结果做一次总的合并。

用Spark SQL写个例子，计算各渠道订单金额：

-- 先加盐，再局部聚合 SELECT SUBSTR(salted_key, 2) AS channel_id, SUM(sub_total) AS total_amount FROM ( SELECT CONCAT(CAST(FLOOR(RAND() * 10) AS INT), '_', channel_id) AS salted_key, amount AS sub_total FROM dwd_order_detail WHERE dt = '2024-06-15' ) t GROUP BY SUBSTR(salted_key, 2);

这里我做的是两阶段：

• 第一阶段：给channel_id加上0到9的随机前缀，然后按salted_key分组，因为RAND()会把记录均匀打散到10个子组，原本一个超大分组被拆成10个相对均匀的分组。
• 第二阶段：将子Key前缀去掉，再做一次sum合并。

用PySpark实现同样的逻辑，写法也一样直观：

from pyspark.sql.functions import concat, lit, round, rand, col salt_num = 10 df_salted = df.withColumn( "salted_key", concat(round(rand() * (salt_num - 1)).cast("int"), lit("_"), col("channel_id")) ) # 第一阶段：按盐化key局部聚合 df_agg1 = df_salted.groupBy("salted_key").agg({"amount": "sum"}) # 第二阶段：去掉盐前缀，最终聚合 df_result = ( df_agg1.withColumn( "channel_id", expr("SUBSTR(salted_key, LOCATE('_', salted_key) + 1)") ) .groupBy("channel_id") .agg({"sum(amount)": "sum"}) )

加盐数N怎么选？这是个经验活。我一般参考倾斜Key的数据量占比：如果它占总数据量的40%，N取8到16就很合适；如果占80%以上，N可能要取到32甚至64。N越大，拆分后的子任务越均匀，但第二阶段合并的分组也越多，引入的额外Shuffle开销也越大。实操中N取8到16是性价比比较高的区间，除非倾斜极端严重，否则不需要更大。

要注意：count(distinct)这类精确去重不能盲目用上面的加盐方案，因为你一旦加盐，同一个原始Key的重复值就会散到不同分组里，每个分组去重的前提被破坏了。去重场景一般用先加盐分组，再去重合并的方式，或者调整成group by + collect_set + 求size的组合，这个我在后面章节专门展开。

3.2 广播小表：Join倾斜时最先试的优化

如果倾斜发生在Join场景，而其中一张表足够小（几MB到几十MB），优先考虑使用Broadcast Join（广播表）。

常规的Join执行，会把两张表按Join Key做Shuffle：相同Key的记录被分发到同一个节点，再进行匹配。这意味着大表的巨大Key会把成堆记录压向单节点。而广播模式完全不同：小表在每个Executor节点上保存一份完整的副本，大表分片不用跟小表做全局Shuffle，直接在本地和内存里的小表副本做匹配。倾斜问题瞬间消解。

Spark 3.0之后，开启AQE（Adaptive Query Execution）还能在某些场景下自动把大表Join小表切换成广播模式。手动指定时，可以在SQL里用Hint：

SELECT /*+ BROADCAST(dim_channel) */ a.channel_id, SUM(a.amount) AS total_amount FROM dwd_order_detail a JOIN dim_channel c ON a.channel_id = c.channel_id WHERE a.dt = '2024-06-15' GROUP BY a.channel_id;

Spark技术中，广播一个大小超过spark.sql.autoBroadcastJoinThreshold的表可以通过显式hint强制实现。但不要无脑广播一切表：广播表会复制到每个Executor，如果表太大反而会把Driver和Executor的内存吃爆。我的经验是单表不要超过100MB，否则就要评估节点内存能不能承受了。

3.3 大表Join大表：加盐+拆分重组方案

广播只适用于小表，两边都是数据量很大的表时，就得用加盐+拆分重组的组合拳。

这个方案的核心思路是：只针对倾斜的超级Key做文章，其他Key保持原有处理逻辑。

具体分三步：

1. 确定倾斜Key集合：最直接的办法是查这个维度表，统计每个Key在大表中的关联记录量，列出Top N。也可以先用第一条SQL把大表自身按Key group by，筛出数据量超过阈值的那几个Key。
2. 大表侧（大Key）加盐：对命中了倾斜Key的记录，给Join Key拼上随机前缀0到N。
3. 小表/维表侧扩容：在另一张比较小的表里，把与倾斜Key匹配的那几条记录做N份复制，并把它们的Join Key也分别拼上0到N的前缀。这样，原本一个超级Key被均匀拆成了N个组合Key，大表和小表就能正常匹配上了。

用SQL写出来大致是这样：

-- 假设确定 channel_id='0' 是倾斜Key，盐数 N=16 WITH salted_big AS ( SELECT CASE WHEN channel_id = '0' THEN CONCAT(CAST(FLOOR(RAND() * 16) AS INT), '_', channel_id) ELSE channel_id END AS join_key, amount FROM dwd_order_detail WHERE dt = '2024-06-15' ), expanded_small AS ( SELECT CASE WHEN channel_id = '0' THEN CONCAT(CAST(explode(sequence(0, 15)) AS STRING), '_', channel_id) ELSE channel_id END AS join_key, channel_name FROM dim_channel ) SELECT s.amount, c.channel_name FROM salted_big s LEFT JOIN expanded_small c ON s.join_key = c.join_key;

这里的explode(sequence(0, 15))就是把维表中的'0'这一个Key扩展成16行，分别配上0到15的前缀。这是最原始的写法，在实际数据中你一般不会直接在SQL里写死'0'，而是把倾斜Key列表做成一张临时表，再和维表做交叉关联，代码更通用。

这个方案有两个缺点：

• 需要提前确定倾斜Key，有运维成本。
• 大Key的Join从一次变成了跟N次匹配做，对个别Key来说计算量反而放大了。

所以在没有AQE的旧版本Spark里，这个方案是唯一选择；如果用的是Spark 3.2+，可以把这项工作交给AQE的自动倾斜Join优化（见3.5），它在运行时动态把倾斜分区拆成多个子分区，人工方案更适合那些AQE无法覆盖或者你想完全掌控的执行计划。

3.4 自定义Partitioner：把数据“重新排班”

有些场景不适合加盐，比如Streaming处理链路中要保序，或者业务上对Key的语义有严格约束，这时候可以通过自定义分区器（Custom Partitioner）把数据重排。

框架默认的分区策略是哈希：hash(key) % numPartitions。如果Key分布本来就歪，哈希结果也大概率歪。自定义Partitioner可以让你自己定义“哪个Key去哪个分区”——比如按照数据量的分位数做设计，把数据量大的Key单独放到一个专属分区，数据量小的合并到一个分区，让每个分区承载的数据量尽量接近。

在Spark中，repartition和partitionBy允许传入分区器：

from pyspark.sql.functions import col # 用哈希重分区 df = df.repartition(64, col("channel_id")) # 更细粒度控制，可以使用RDD API + 自定义Partitioner from pyspark.rdd import RDD class ChannelPartitioner: def __init__(self, num_partitions): self.num_partitions = num_partitions def get_partition(self, key): # 超级Key单独放一个分区，其余Key按哈希均匀分布 if key == "0": return 0 return hash(key) % (self.num_partitions - 1) + 1 def num_partitions(self): return self.num_partitions

自定义分区器适合那种数据分布规律非常明确、并且你知道“哪些Key大、哪些Key小”的情况。缺点是它跟业务强绑定，代码维护成本较高，一旦数据分布发生了变化，你可能要重新调参。所以在实际项目中，能用加盐解决的，我不会轻易上自定义分区器。

3.5 Spark 3.0+ AQE：能不能让框架自己搞定

如果你的Spark版本已经是3.0以上，那spark.sql.adaptive.enabled必须打开。AQE（自适应查询执行）在运行时能基于已经完成的Stage统计信息动态调整执行计划，对数据倾斜有直接的缓解能力。

AQE的倾斜处理做两件事：

• 动态调整Shuffle分区数：一开始分区数过高或过低，AQE会在Shuffle之后统计每个分区的数据量，自动合并小的分区（spark.sql.adaptive.coalescePartitions.enabled），或者拆分为数较多的分区，让整体负载更均衡。
• 动态拆分倾斜分区（Skew Join）：当AQE检测到某个Shuffle分区的数据量远超中位数时，会自动把这个分区按比例拆成多个子分区，并且把另一侧的关联数据复制多份，效果等同于自动版的“加盐+扩容”。

把配置打开后，通常还要微调几个参数：

spark.sql.adaptive.enabled=true spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.skewJoin.enabled=true spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB

这些参数的含义：

• skewedPartitionFactor=5：分区数据量超过所有分区中位数5倍以上，才被认为是倾斜分区。
• skewedPartitionThresholdInBytes=256MB：即便倍数达到阈值，分区数据量至少要超过256MB才能被触发拆分，避免频繁拆微小分区。
• advisoryPartitionSizeInBytes=128MB：这是“理想分区大小”，AQE合并/拆分分区时会尽量向这个大小靠拢。

实践中的体感：AQE对倾斜的自动处理能覆盖70%左右的常规场景，但别把宝完全押在它身上。比如超级Key那样达到“一Key顶一片”的程度，自动拆分后可能依然生成一个巨无霸分区，再加上扩容另一侧表的成本，瓶颈只是被转移。所以我的做法是AQE必开，但“先数据治理，再框架兜底”——能通过业务规则消除的超级Key，永远优先在数据源头解决。

4. 查询加速：从存储布局到执行引擎的全面提速

数据倾斜解决好之后，查询性能的另一个大头就是存储层和执行层。大数据架构跑得再快，如果底层的存储布局一塌糊涂，计算引擎再强也白搭。

4.1 分区、分桶与文件裁剪：让引擎少读数据

查询加速的第一原则，是让引擎只读它需要的数据。分布式架构中天然的数据裁剪手段是分区（Partition）和分桶（Bucket）。

分区是按业务维度对表进行物理切分，最常见的维度是日期和地域。每次查询带上了分区过滤条件，引擎就能直接把不符合条件的分区目录跳过，完全不用读数据。

分区设计有两个要点：

• 分区粒度要合理：按天分区是最常见的，数据量大可以按时钟、小时分区。但不要为了“看起来粒度高”就把维度设得很细，比如把分区键做成“日期+城市”联合分区，可能会造成目录数量和文件数量暴涨，元数据开销反噬性能。
• 避免过度分区：我见过一个表按天+小时+渠道三分层分区，结果一个小时的数据量才几百KB，产生了大量碎分区。查询时要扫描的目录数量上百个，NameNode压力大，列裁剪也没法很好工作。

分桶是比分区更精细的数据组织方式，它按指定Key的哈希值把记录均匀分布到一个固定数量的文件中。分桶配合排序，可以实现更好的聚合性能和Bucket Join（当两张桶表桶数相同且在join key上分桶时，可以跳过全量Shuffle）。

实践建议：地图数据按日期分区，用户数据按user_id分桶到32或64个桶。分桶键尽量选择高基数、分布均匀的字段，字段基数过低（比如只有0和1）就是分桶灾难。

4.2 列式存储+压缩：让宽表查询快一个数量级

从Hive时代的纯TextFile切到列式存储，查询性能经常能提升5到10倍。原因很简单：

• 行存储（TextFile、Avro SequenceFile）在查询某几列时，必须整行读入，再丢掉不需要的列。对于上千字段的宽表，I/O浪费非常夸张。
• 列式存储（Parquet、ORC）把同一列的数据连续存放，引擎只读取SQL里引用的那几列。列数越宽、查询列数越少，优势越明显。

Parquet和ORC除了列裁剪，还有几个重要的加速特性：

• 谓词下推：存储格式会记录每个RowGroup的min/max统计信息，数据源层就能把不满足过滤条件的RowGroup直接跳过，减少I/O量。
• 压缩比高：列式数据相关性高，压缩后体积远小于行格式。常见的搭配是Parquet + Snappy，也可以使用Parquet + ZSTD。ZSTD的压缩比高于Snappy，CPU开销略大但现代集群完全扛得住。

建表时示例：

CREATE TABLE dwd_order_detail_parquet ( channel_id STRING, user_id STRING, amount DECIMAL(10,2), ... ) PARTITIONED BY (dt STRING) STORED AS PARQUET TBLPROPERTIES ('parquet.compression'='zstd');

小技巧：对经常做聚合查询的大表，可以在建表时加上ORDER BY，让数据在文件内部有序排列。Parquet的min/max索引在一段有序数据上可以越过大量RowGroup，过滤效果接近“分区裁剪”的威力。比如订单表按dt和city_id排序之后，查询某个城市的订单，引擎只要少量扫描几次就能定位数据块。

4.3 硬件无关的SQL调优：写法和执行策略的取舍

存储布局搭好了，接下来看SQL本身的写法。同样一个查询，写法不同，执行计划可能天差地别。

先过滤，再聚合：把过滤条件尽量下推到离数据源最近的地方。引擎在这方面已经做了大量自动优化，但SQL的书写顺序和逻辑结构仍然影响执行计划。比如在子查询里先做WHERE再JOIN，比先JOIN再WHERE要理智得多。

避免不必要的UDF和笛卡尔积：查询中如果出现UDF，可能会阻断下推优化，要对这批数据做单独的评估。

count(distinct)值得单独优化：在超大数据集上，count(distinct)往往是一个Reduce的灾难。如果业务上允许一定误差，可以用HyperLogLog近似去重（Spark里有approx_count_distinct函数），能省至少一个数量级的时间。如果必须精确，可以把一次全局去重改成两步：

-- 第一步：按照明细维度先做去重计数 SELECT dt, channel_id, COUNT(DISTINCT user_id) AS uv FROM dwd_order_detail GROUP BY dt, channel_id; -- 第二步：把多个维度的结果累加 SELECT dt, SUM(uv) AS total_uv FROM tmp_uv_by_channel GROUP BY dt;

这样做的好处是：每个维度组合的uv先在各分区单独算，分散了去重的压力，最后只对很小规模的结果做汇总。本质上，这跟我前面讲的“两阶段聚合”一脉相承。

避免ORDER BY全量全局排序：除非业务必须，否则ORDER BY会触发全局单分区排序，代价极高。很多情况下只需要SORT BY或DISTRIBUTE BY + SORT BY，用分区级排序代替全局排序。

选择正确的JOIN策略：三张表以上Join时，Join顺序由优化器决定。老版本的Spark优化器可能选错顺序，需要主动用Hint控制。Shuffle Hash Join（小表批量的哈希连接）适合小表，Sort Merge Join（排序合并连接）适合两张大表等值连接，Broadcast Join适合小维表。

4.4 小文件治理：看不见的查询性能杀手

这是我在生产环境踩过最多坑的地方。所谓小文件，是指单个文件大小远小于数据块默认大小（比如128MB），通常只有几KB到几MB。

小文件对查询性能的杀伤力，比数据倾斜更隐蔽：

• 任务数爆炸：每个文件至少被一个Task读取，几千个文件意味着几千个Task，任务调度的开销把真正干活的时间全挤掉了。
• 元数据压力：文件数量会上亿，NameNode和Metastore的内存压力直接爆表。
• 动态分区写、Streaming作业频繁落地、INSERT OVERWRITE等操作是生产上小文件来源的三个主力。

治理方案主要有三种：

1. 合并已有小文件：跑一个INSERT OVERWRITE任务，读取原表的同时按照合理分区和repartition策略重新写入。

INSERT OVERWRITE TABLE dwd_order_detail_parquet PARTITION (dt='2024-06-15') SELECT /*+ REPARTITION(1) */ channel_id, user_id, amount FROM dwd_order_detail_raw WHERE dt = '2024-06-15';

如果分区文件数还不是太少，可以用REPARTITION指定一个合理的分区数，让每个文件接近128MB。以上示例按每个分区一个文件来处理。

2. Spark AQE的coalescePartitions参数：开启后会让Shuffle输出阶段的文件数向64MB到128MB靠拢，对Shuffle后的文件数量有很好的收敛效果。

3. 控制写入源头：Flink写Hive时设置sink.partition-commit.trigger和分区提交策略，让文件在分区提交前预先合并；Spark Structured Streaming写文件时，用coalesce和maxRecordsPerFile控制单个文件大小。

小文件治理的验算思路：一个数据块大小是128MB，期望每个分区文件数量 = 分区数据总量 / 该大小。如果某天分区数据量是1GB，那么理想文件数量应该在8个左右。对应开发的提示是，在设计的合并任务里，按这个思路算出来的分区数来指定repartition的并行度。

5. 资源与执行引擎调优：让每一度电都花在刀刃上

5.1 Shuffle分区数（并行度）的设置原则

很多人的任务跑不快，不是代码写得不对，而是Shuffle分区数设得离谱。

Spark的spark.sql.shuffle.partitions默认是200。这个值在数据量几百MB的时代是合理的，但放到现在动辄几十GB、上百GB的离线任务里，200个分区会导致每个分区背负几GB数据，Task处理时间以小时论也就不奇怪了。

分区数怎么才算合理？我的经验公式是：

期望分区数 ≈ 单次Shuffle数据总量 / 期望单个Task处理的数据量

单个Task处理的数据量，我在日调度任务里一般取200MB到500MB之间。为什么是这个范围？因为：

• 如果分区太大，单Task的处理时长就上去了，某个Task失败后重试成本太高。
• 如果分区太小，任务调度开销、分区元数据开销、以及最后的输出小文件问题都冒出来了。

举个例子：某个Job在Shuffle阶段要处理50GB数据，希望每个Task处理约300MB，那分区数应该在：

50GB * 1024 / 300MB ≈ 170 个分区

但170这个值还要结合Executor的并行能力来微调。如果Executors总数是10个，每个4核，那同时能跑40个Task。170个分区意味着每个Executor要串行处理4-5批Task，这个排队比例是合理的。如果Executor总数只有4个，每核并行度才16，那170个分区会导致严重的排队，这时要么增加Executor，要么调大分区数让每个Task处理更少的数据，加快单批流转。

5.2 内存模型、GC与Spill：为什么“加大内存”不是万能药

大数据任务出现了OOM或异常GC，我见过太多人第一反应就是“把Executor内存调大”。但内存加大跟性能提升往往不是线性关系，而且Spark的内存模型不是“给多少用多少”。

Spark内存分两大块：

• Execution Memory（执行内存）：用于Shuffle、Join、聚合等操作，这块内存在任务执行时可以动态借用。
• Storage Memory（存储内存）：用于缓存RDD和数据块，是Spark的Cache、广播变量等功能的存储区域。

如果任务出现频繁Spill，真正的解法往往不是“加大内存总量”，而是：

1. 提升spark.memory.fraction比例：默认0.6，意思是统一内存区域占总内存的60%，剩余40%给RPC和用户代码。如果用户代码没有特别大开销，这个值可以调到0.75以上。
2. 降低spark.memory.storageFraction：默认0.5，是Storage与Execution的硬边界。如果任务没有大量Cache动作，这个值可以调低到0.3，让更多内存给Shuffle和聚合使用。
3. 启用Kryo序列化：默认的Java序列化对象体积大、序列化慢。换成Kryo能显著降低对象大小和GC压力。
4. 减少大对象在Java堆里的占用：比如把超大的String换成更高效的数值类型，或者使用堆外内存（offHeap）为某些对象腾出空间。

这里顺便提一句，很多其他领域的性能优化也遵循同一套逻辑：Julia里做性能优化时会强调“减少内存分配、避免大对象”；移动端性能优化时会关注“避免GC抖动、减少布局层级”。根源都是“内存的碎片化和大对象的频繁创建”会拖垮整体性能。大数据的调优思路跟它们是一致的——先减少不必要的对象，再想怎么加大容量。

5.3 CBO与执行计划：让优化器“睁眼做决定”

Spark SQL里有一个隐藏性能开关：CBO（Cost-Based Optimizer，基于代价的优化器）。CBO的意义在于，优化器在决定Join顺序、Join策略时，需要知道每张表的数据量、字段基数等统计信息。统计信息缺失，优化器就只能靠猜。

开启CBO前要做的准备工作是收集统计信息：

ANALYZE TABLE dwd_order_detail COMPUTE STATISTICS; ANALYZE TABLE dwd_order_detail COMPUTE STATISTICS FOR COLUMNS channel_id, user_id;

然后开启相关配置：

spark.sql.cbo.enabled=true spark.sql.cbo.joinReorder.enabled=true spark.sql.statistics.size.autoUpdate.enabled=true

开启后，用EXPLAIN看执行计划，会比关闭时多出Cost信息。我经常在优化完一批SQL后，用EXPLAIN COST反复看Join的顺序和策略是否合理。如果发现优化器选了奇怪的Join顺序，再通过Hint手动调整。

读取执行计划是查询调优的基本功。我常用的命令：

EXPLAIN SELECT ... ; EXPLAIN EXTENDED SELECT ...; EXPLAIN COST SELECT ...;

EXPLAIN执行计划里重点看三样东西：

1. Join策略：是不是出现了BroadcastExchange，还是ShuffleHashJoin或SortMergeJoin。
2. Exchange节点：Shuffle发生在哪，分区数是多少，数据量预估。这个预估数值跟实际数据量的偏差能帮你判断统计信息是否需要更新。
3. Filter下推位置：过滤条件是不是出现在读取文件的物理计划里，还是在更靠近输出的地方。如果Filter在很晚才出现，说明这个过滤条件在某些阶段没能下推，要检查字段类型是否一致、是否有函数包裹导致谓词下推失效。

6. 生产环境真实排障实录：三个案例和三张速查表

6.1 案例一：默认渠道号导致Join倾斜，报表两小时跑不完

现象：某业务线日报任务，每天凌晨两点开始跑，经常跑到早上六点还出不来。UI里看到一个Join Stage卡在99%，某个Executor的GC时间占比超过70%。

排查：按2.2的SQL对dwd_order_detail按channel_id做group by，发现channel_id = '0'的订单量占了将近90%。这个'0'是历史遗留下来的“默认渠道”，几乎来自老版本客户端的默认值。

治理：由于另一张维度表dim_channel只有几千行，直接广播；同时在大表侧把'0'这个超级Key做了加盐处理。因为广播表足够小，加盐后的结果可以正常Join。

效果：

这个案例说明一件事：倾斜通常不是框架问题，而是数据质量问题的投影。'0'这种默认值如果在业务上不可消除，至少可以在数仓建模时通过过滤、替换或二级映射来减轻它的“重量”。

6.2 案例二：count(distinct)在超亿级UV上被“算爆”

现象：一个渠道流量分析作业，要对亿级流量数据做COUNT(DISTINCT user_id)，该任务永远在三小时左右徘徊，而且Spark UI上只看到一个Reduce，所有的计算压力都压在一个Task上。

排查：EXPLAIN看执行计划，发现Aggregate节点只有一个分区，也就是SinglePartition，这是经典的“全局去重”场景。业务上对UV虽然要求精确，但明细维度很多，完全可以分维度预聚合。

治理：改成先GROUP BY dt, channel_id做局部去重聚合，再对结果SUM(uv)。因为中间结果小，第二次聚合几乎不花时间。如果需要周度/月度等长周期指标，把中间结果落到一张汇总表，用增量更新的方式不断叠加。

效果：任务从3小时12分钟降到28分钟。

6.3 案例三：小文件让查询“卡在扫描层”

现象：某张Hive表进行按日分区的数据写入，一天的数据只有约500MB，但文件数却有两千多个。查询一个聚合指标，光扫描阶段就花了7分钟，实际聚合计算只用了3秒。

排查：通过Metastore查看分区内文件数和文件大小，发现平均每个文件只有200KB左右，远低于128MB的理想值。这些文件来自某个Flink任务频繁写小数量的数据块。

治理：通过对表进行重写合并，INSERT OVERWRITE按分区重新写入，分区内文件数从2000+降到4个。同时把后续写表改成至少半小时凑一个文件再commit的策略。

效果：扫描时间从7分钟降到20秒，整体查询时间从10分钟以上压到1分钟以内。

6.4 排障工具与命令速查表

在这张表的基础上，我每次排障会做一个简单的“时间线记录”：什么时候现象出现、阶段UI的哪个指标异常、执行计划里哪个节点最重、调整后重新跑的耗时对比。大数据性能优化长期靠的是这种有条理的积累，而不是碰运气式的瞎试。

最后再分享一个我个人的体会：性能优化的起点永远是“先看数据，再看代码，再看配置，最后看硬件”。很多团队一遇到性能问题就急着加机器、调参数，结果数据倾斜依然存在，SQL写得再烂再好的引擎配置也救不回来。数据分布式情况的治理，优先级高于一切技术手段，因为框架只是参数的放大器和执行器，它解决不了数据本身质量的问题。把这一条刻在脑子里，你的大数据架构性能稳定天花板会比大多数人高一大截。