1. 数据倾斜的本质与典型场景
数据倾斜是分布式计算中的经典难题，当我在2016年第一次处理TB级用户行为日志时，就遭遇过某个城市用户量激增导致的shuffle崩溃。这种现象的本质在于数据分布的幂律特性——少数几个key承载了绝大部分数据量。在Spark中具体表现为：
Join操作倾斜：电商订单表与用户表关联时，某些"羊毛党"用户的订单量可能是普通用户的万倍。华为云文档中提到的案例正是此类场景，当a.id=b.id的关联键存在严重倾斜时，会导致hash分桶后单个task处理的数据量远超其他节点。
GroupBy聚合倾斜：统计短视频播放量时，头部1%的视频可能占据90%的播放记录。我曾遇到过一个热点视频导致200个executor中，有199个早早完成却在等待最后一个executor跑完的情况。
数据源倾斜：读取Hive分区时，某些分区的数据量异常庞大。比如按天分区的日志表，遇到促销日的日志量可能是平日的10倍。
关键诊断指标：在Spark UI中观察任务执行时间分布，如果发现stage内有少量task的运行时间明显长于其他task（通常差距在3倍以上），基本可以确认存在数据倾斜。
2. 基础优化策略与实战验证
2.1 预处理倾斜键
对于已知的倾斜key，最直接的方案是在业务逻辑允许的情况下进行预处理。去年我们处理广告点击日志时，就采用了以下方案：
// 示例：对异常user_id增加随机前缀 val skewedUserIds = Seq("u123456", "u789012") // 已知热点用户 val processedDF = rawDF.map(row => { val userId = row.getAs[String]("user_id") if(skewedUserIds.contains(userId)) { val salt = (new util.Random).nextInt(10) Row(userId + "_" + salt, row.getAs[String]("event_type")) } else { Row(userId, row.getAs[String]("event_type")) } })这种方案需要两次聚合：先对带盐值的数据做局部聚合，再去盐后做全局聚合。实测将一个原本需要4小时的任务降到了27分钟。
2.2 调整并行度与分区策略
通过spark.sql.shuffle.partitions（默认200）可以增加reduce端并行度。但要注意：
- 分区数不是越多越好，超过实际数据量时会产生大量小文件
- 对于Skewed Join，更推荐使用自适应查询执行(AQE)：
-- 在Spark 3.0+中启用AQE SET spark.sql.adaptive.enabled=true; SET spark.sql.adaptive.coalescePartitions.enabled=true; SET spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB;我在生产环境测试发现，当单个分区超过256MB且大于其他分区中值的5倍时（华为云文档中的默认阈值），AQE的倾斜处理效果最佳。
3. 高级优化技巧与避坑指南
3.1 广播Join的妙用
当参与Join的一个表较小时（<10MB），强制广播可以避免shuffle：
// 方式1：通过API强制广播 df1.join(broadcast(df2), Seq("key")) // 方式2：SQL提示 sql("SELECT /*+ BROADCAST(t2) */ * FROM t1 JOIN t2 ON t1.id=t2.id")但要注意：
- 实际数据量可能大于统计值，我曾遇到过因数据压缩导致误判的情况
- 广播表过大会导致driver内存溢出
3.2 倾斜Join优化方案对比
| 方案 | 适用场景 | 优缺点 |
|---|---|---|
| 随机前缀法 | 已知倾斜key且可修改业务逻辑 | 效果显著但需要两次聚合，代码侵入性强 |
| AQE自动倾斜处理 | Spark3.0+，中等规模倾斜 | 自动化程度高，但对FULL OUTER JOIN等复杂场景无效（如华为云文档所述） |
| 分离倾斜key | 极少数极端倾斜key | 需要精确识别倾斜key，可能影响业务完整性 |
| 增加shuffle分区 | 轻度倾斜场景 | 简单但效果有限，可能增加小文件问题 |
4. 生产环境综合解决方案
4.1 实时监控与动态调整
我们在YARN集群上部署的监控方案包含：
- Prometheus采集Spark executor的shuffle读写指标
- 当检测到单个task处理数据量超过阈值时，自动触发以下操作：
- 动态调整
spark.sql.adaptive.skewJoin.skewedPartitionFactor - 对极端倾斜任务进行熔断，转储到Hive表人工处理
- 动态调整
4.2 小文件治理组合拳
针对Spark Streaming产生的小文件问题（如热搜词所示），我们采用：
// 1. 控制每批次的输出量 spark.conf.set("spark.sql.sources.maxRecordsPerFile", 500000) // 2. 使用Delta Lake的OPTIMIZE命令定期合并 spark.sql("OPTIMIZE delta.`/path/to/table` ZORDER BY (timestamp)")这种方案使得HDFS文件数量从日均20万+降至5000左右，NameNode压力显著降低。
5. 参数调优黄金法则
根据百次调优经验，我总结出以下参数组合（Spark 3.3+）：
# 基础AQE配置 spark.sql.adaptive.enabled=true spark.sql.adaptive.coalescePartitions.enabled=true spark.sql.adaptive.advisoryPartitionSizeInBytes=64MB # 倾斜Join专项 spark.sql.adaptive.skewJoin.enabled=true spark.sql.adaptive.skewJoin.skewedPartitionFactor=8 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB # 小文件控制 spark.sql.shuffle.partitions=400 spark.sql.adaptive.rebalancePartitionsSmallPartitionFactor=0.2这些参数需要根据集群规模动态调整，比如在100个executor的集群中，建议将spark.sql.shuffle.partitions设为executor核数的2-3倍。
遇到JOIN倾斜时，不妨先用EXPLAIN EXTENDED查看执行计划，重点观察numPartitions的变化。曾经有个任务通过增加spark.sql.adaptive.nonEmptyPartitionRatioForBroadcastJoin=0.2，使得广播Join的触发更积极，性能提升了40%。
