作为一个常年跟Spark作业打交道的人我几乎可以断定你只要在生产环境跑过稍微有点规模的数据任务就一定见过这样的场景——某个Stage进度条卡在99%不动或者明明3000个Task里有2999个几秒钟就跑完了偏偏有一个跑了一个小时还在磨蹭最后整个Application被拖到OOM或者超时。用行话说这大概率就是数据倾斜。数据倾斜在Spark任务里出现的频率远比你想象的高。它本质上是某种“贫富不均”被计算引擎放大后的结果集群资源是固定的但数据处理却严重偏向某个Task。这不是什么深奥的玄学也不一定是你代码写得“烂”很多时候它就是由数据本身的分布特性决定的。今天我不打算讲教科书上的概念而是从我自己踩过的坑出发把这个问题的识别方法、产生原因、解决套路和实际操作细节一次性讲透。1. 先搞清楚“任务倾斜”到底是什么1.1 表象背后的本质是节点负载失衡很多刚接触的人会把数据倾斜和OOM内存溢出搞混。确实是倾斜严重的时候会伴随OOM但两者不是一回事。OOM是内存真不够了而倾斜是“某个Executor干了超额的活”。我们可以用生活场景来类比假设有个搬砖任务100个人每人分一堆砖正常情况下大家都差不多重。结果有个人分到了一整栋楼的砖其他人早干完喝茶了他还在那吭哧吭哧搬。Spark里面的数据倾斜就是——某个Partition里的数据量远大于其他Partition导致处理它的Task执行时间远超同类。这个本质可以用一句话概括任务执行时间 木桶上最长的那块板。整个Application能不能按时跑完不取决于那99%的快速Task而取决于那1%的慢Task。更难受的是那个慢Task还会因数据量过大引发GC频繁、序列化保真、网络IO暴增连带拖垮它所在节点的其他任务。1.2 从DAG和Executor日志里识别信号要判断是不是倾斜我会按三步走。不是靠猜是靠证据。第一步看Spark UI打开Stage页面盯着“Summary Metrics”那一栏。如果某个Task的Duration是中位数的几十倍Input Size / Shuffle Read Size也明显异常那基本就锁定倾斜了。注意看某个Executor的Task数量如果一个Executor上有几十个Task且都很快另一个Executor上只有一个极慢的Task这也是强信号。第二步是去看Shuffle Read Size。倾斜通常发生在Shuffle之后因为Shuffle是按Key重新分区的过程把相同Key的数据拉到同一个Reducer。若某个Reducer拿到的数据是平均值的百倍以上那它对应的Task怎么都快不起来。所以数字不会骗人看到Shuffle Read异常偏大的Task就说明倾斜点在Shuffle阶段。第三步别急着改代码先看Shuffle Write端的记录。很多倾斜的源头是Map端某个分区的输出量就很大而不是Reducer真的读太多。去看到底是“写出来就多”还是“读取时合并得多”这决定了后面用哪一种解决策略。1.3 为何默认分区机制容易放大倾斜Spark的默认并行度/分区逻辑很难做到“按数据量均匀切分”特别是涉及到HashPartitioner的时候它是基于Key的哈希值取模跟数据分布本身无关。如果Key分布天然就是长尾的比如“北京”这个城市独占了几亿条数据其他城市只有几千条那么按城市分组后Partition大小天然就畸轻畸重。如果只是数据分布不均匀也就算了问题是Spark默认的Reducer数量往往是固定的由spark.sql.shuffle.partitions决定默认200它不会自动感知你的数据到底是怎么分布的。固定数量的桶去接不均匀的水有的桶满到溢有的桶还是空的这还会引发另一个连锁反应同一Stage的Task执行完毕时间差距拉大整个Stage因为等待那个慢任务而白白浪费大量集群资源和时间。2. 常见倾斜场景的定位与根因分析说实话大多数倾斜都可以归为四类。我这么多年排查下来基本没有逃出过这个范围的。把根因找准了方案就呼之欲出了。2.1 数据分布本身的长尾特征这类倾斜最“不可控”因为根源在上游数据。典型的例子是统计某电商平台各城市的GMV结果“北京”自成一档其他省份加起来数据量都不到它的五分之一。再比如在日志分析中某个IP是爬虫或网关入口它产生的日志量是普通IP的万倍。判断方法很简单直接用Spark SQL对关联字段做一次COUNT GroupBy或者对明细数据做一次approx_count_distinct估算看下最大的几个Key占据的比例。如果单个Key的值超过了整体数据的10%甚至更高你就得意识到后面所有基于这个Key的操作都会被打成“一超多弱”的局面。这类倾斜不能靠调参解决你需要靠算法层面把热点Key“打散”。因为一个Key的量太大任何固定分桶都撑不住得让它不要再绑死在同一个人/同一个Executor身上。2.2 Join操作导致的关联键倾斜如果说数据分布的长尾是“天灾”那Join倾斜就有一半是“人祸”了。用两张大表做Join时如果关联键是城市ID、用户ID、商家ID而拥有头部流量的ID特别集中那么Shuffle Join阶段无论怎么Hash所有属于“北京”的数据都会涌入同一个Reducer去跟另一张表匹配。我见过一个很典型的案例两张比较大的表做用户行为关联一个是用户维表几千万条一个是行为事实表几亿条。关联键是user_id按理说user_id很分散。结果某头部渠道的少数几个测试账号行为数据占了整体四分之一。那次作业跑了快两小时没跑完问题就集中在三个Task上。这就是典型的Join键倾斜导致Hash分桶不变形但负载变形。这种情况半血斩杀的方式是先搞清楚是“大表 Join 大表”还是“大表 Join 小表”两者处理起来差异很大。如果是一大一小可以考虑广播如果是两张大表就需要配合加盐、布隆过滤器或先过滤后Join。2.3 过滤条件引发的计算侧倾斜数据本身分布正常但因为一个过滤条件导致某些Reducer的输入被“筛选”得特别多。举个例子你有全国用户数据要统计近一个月的活跃用户过滤条件是“最近一次登录时间 2023-01-01”。结果某几个大城市的数据密度极高在过滤后依然有几百万行其他三四线城市可能只剩几百行。此时即使上游Map端Shuffle Write是均等的经过表达式过滤后Reducer端Read的数据还是会出现巨大差异。这类情况在逻辑上是“合法”的——数据确实是处理后自然形成的分布但处理结果对系统不友好。排查时特别容易漏很多人在UI上看到Shuffle Read不均衡就想当然认为是上游数据倾斜实际是某个过滤条件把热点值集中了。2.4 分区键选择不当引起的存储与计算双重问题这类问题经常藏得很深表面看是“任务越跑越慢”“某天突然OOM”排查到底发现根因根本不是今天的SQL而是很久以前建表时分区字段选的就有问题。假设你要读取一份Hive表做分析表的分区键是“统计日期(yyyyMMdd)”但实际业务上某天是全量快照某天又是增量数据导致某几个分区文件数量特别大。当你的查询条件刚好命中这个重分区时Spark动辄scan几百GB数据Shuffle Read自然狂飙。再有就是分区字段是“小时”但某一次补数任务把一年的数据都写到了同一个小时分区。这类情况本质上是“存储层布局不均匀”但最终反馈到计算层时就表现为Task倾斜。所以在排查倾斜问题时不要只看计算层的Key记得回头看看读取的表源存储是否均匀。有时候在SQL前加一个过滤条件把无用的重分区数据排除掉速度提升比什么加盐都来得明显。3. 实战中验证可行的三类解决方案现在说重点怎么解决。基于上面的根因分析我按优先级顺序讲三个核心方案。这三个方案我都在生产环境验证过且有具体的落地方案和代码片段。3.1 两阶段聚合给热点Key“加盐”打散这个方案只针对GROUP BYCOUNT/SUM/AVG这类聚合操作性价比最高也最容易落地。核心思路是人为改变聚合时的Key粒度先把每个Key随机拆成N份“盐粒”做一次局部聚合然后用去掉盐粒的真实Key再做一次全局聚合。这样数据就不再挤在少数几个固定的Partition上而是被均匀分流。代价是多了一个Stage多一次Shuffle但换来的是整个作业从崩溃变成稳定运行。我用Scala写过比较通用的版本核心片段如下// 给非空key加盐空key单独处理 val saltedDf df.withColumn( salt, when(col(key).isNotNull, (rand() * 100).cast(int)).otherwise(0) ).withColumn(salted_key, concat(col(key), lit(_), col(salt))) // 第一轮带盐聚合 val firstAgg saltedDf.groupBy(salted_key).agg(sum(amount).as(part_sum)) // 第二轮去掉盐后全局聚合 val result firstAgg.withColumn( real_key, regexp_replace(col(salted_key), _\\d$, ) ).groupBy(real_key).agg(sum(part_sum).as(total_sum))关于“盐”的数量不是越多越好需要结合你预期的Reduce端并行度来压。通常取100左右就够如果热点特别严重可以取200。但不要取10000因为第一阶段的Shuffle文件数会暴涨对磁盘和网络本身就是压力。注意如果是COUNT DISTINCT两阶段聚合不能直接照搬上面的方法得先对“key 去重字段 salt”做整体去重然后再按key去重汇总。3.2 Broadcast Join用广播消灭Shuffle如果你的场景是“一个大表联合一个相对较小的表”且小表数据集的量在可控范围内我一般以单Executor能放下为准比如几百MB以内那优先选择Broadcast Join这根本不算一种“解决方案”而是最佳实践。Broadcast Join的原理是把小表收集到Driver端序列化后分发到每个Executor的内存中然后在Map端直接完成关联整个过程不走Shuffle。不产生Shuffle自然也就不存在Shuffle阶段的倾斜问题。在Spark SQL中开启方式极简只需要设置-- 设置小表阈值为20MB默认10MB可按资源调大 set spark.sql.autoBroadcastJoinThreshold20971520; -- 或者直接在SQL里给提示强制走广播 SELECT /* BROADCAST(t2) */ t1.*, t2.name FROM big_table t1 JOIN small_table t2 ON t1.id t2.id但有几个坑我得提前说。首先autoBroadcastJoinThreshold设置太大会导致Executor内存激增反而引发新的OOM所以阈值要冷静。其次Broadcast对小表的大小计算是依据“原始表扫描之后的物理大小”包含解压后的数据别天真地以为“源文件5MB就一定能广播”。最后如果你用到动态分区裁剪或某些复杂Join重排场景优化器不一定走广播所以最好加Hint强制指定。我碰到过一个有意思的案例某张维表在Hive层面是分区表并且每个分区只有几MB但整张表几个TBSpark不会载入全表去广播但也不会自动选择那个“过滤后的小分区”。这种情况下你要在Join前把需要广播的子集先过滤出来Cache一下再Join效果最好。3.3 动态重分区配合AQE让优化器替你分担压力从Spark 3.0开始AQEAdaptive Query Execution自适应查询执行是我强烈建议开启的因为它能在运行时动态调整Shuffle分区数算是优化器层面的一个“后置兜底”。开启方式set spark.sql.adaptive.enabledtrue; -- 开启后建议保持默认可以调整的参数如下 set spark.sql.adaptive.coalescePartitions.enabledtrue; set spark.sql.adaptive.coalescePartitions.minPartitionNum20; set spark.sql.adaptive.skewJoin.enabledtrue; set spark.sql.adaptive.skewJoin.skewedPartitionFactor5; set spark.sql.adaptive.skewJoin.skewedPartitionThreshold256MB;AQE的倾斜Join优化原理是在Shuffle结束后直接检测某个分区的数据量是否超过中位数的一定倍数skewedPartitionFactor默认5倍和绝对阈值skewedPartitionThreshold默认256MB。如果命中它会把这个分区切分成多个子分区分别去跟另一张表的对应Key做Join而不是要求一个Reducer去扛所有数据。这段逻辑我们不用自己写Spark在运行时就把活干了。不过它有适用前提必须是在运行时统计到的Shuffle数据量做的判断必须开启Shuffle服务的精确跟踪。如果你跑的是旧版本Spark比如2.x那就没法享受这块福利只能靠手动加盐。一个经常被忽略的细节AQE虽然能缓解Reduce端读倾斜但它对Map端Shuffle Write不均衡是无能为力的。因为写倾斜在Shuffle开始前就已经固定了AQE只能等到Shuffle Read看到数据之后才动作。所以对于“写出来就多”的场景还是要靠上游打散Key。4. 精细化调优与参数配置实操方案给出来了但真正的生产落地还差“参数配置”这一层。我见过太多同事开了AQE就以为万事大吉结果该慢还是慢。这里我把几个关键点的调优逻辑和实操顺序讲清楚。4.1 分区数、内存与并行度的合理设定先聊一个基础但重要的问题并发度Parallelism到底设置多大合适很多人直接抄网上配置写spark.sql.shuffle.partitions200但实际上更好的是锚定集群规模来算。我个人的计算公式是Shuffle分区数 ≈ 集群核心数 × 3 ~ 4倍。假设Executor数量是20每个Executor分配5个核心那总共100核。Shuffle分区数设计为300~400比较合理。太小则每个Task处理的数据量太大GC时间变长太大则每个Task处理数据少但调度开销、文件碎片会变多。另外还有一个非常需要盯的参数是spark.sql.adaptive.coalescePartitions.minPartitionSize或minPartitionNum。如果你发现开启AQE之后Task数被合并得太少导致某个超大Task又出现了那就调低这个阈值让它别合得那么狠。4.2 内存参数的调配不是越大越好很多人遇到慢任务第一反应就是“加内存”。但数据倾斜的根源不是内存不够而是某个进程分摊了太多数据。你把spark.executor.memory从4G调到32G也只是把爆炸时间延后了而已并不能消除倾斜。我倾向于把内存调整分成两件事来看第一调整spark.executor.memoryOverhead。倾斜Task往往伴随大对象序列化和网络缓冲默认的10%或最低384MB在极端场景下可能不够。调高到2~3GB有助于避免Executor突然被进程Kill掉。第二打开spark.shuffle.service.enabledtrue并且保证Executor的Shuffle数据可以持久化到独立的Shuffle Service进程。这样可以显著减少Executor因GC导致的Shuffle文件丢失同时提升AQE在动态合并时的稳定性。4.3 文件数量与文件大小的取舍如果你读的是HDFS或对象存储上的大量小文件这也是一个隐藏倾斜源。每个文件都会被一个Partition读取小文件过多会导致Driver端的Task数极其庞大而某个大文件又会让某个Task特别慢。这时候我建议在读取阶段做一次文件过滤或Coalesce优化# 读取时重分区Python示例 df spark.read.parquet(hdfs://path/to/table).repartition(200) # 缓存后复用 df.cache().count()或者如果你用的是Spark SQL可以通过设置spark.sql.files.maxPartitionBytes默认128MB和spark.sql.files.openCostInBytes默认4MB来控制分区读取粒度。小文件越多把这个maxPartitionBytes适当调低能让Driver把多个小文件合并读取减少Task数量。但调这些参数之前先看一眼Spark UI的Input Size表格。如果发现大量Task的Input在几MB级别数量却有上万那一定是小文件问题这时候优先去源头合并小文件比在计算层硬扛要有效得多。5. 案例复盘一个生产环境的倾斜解决全流程前面讲了一堆理论最后用一个我实际处理过的案例来完整串一遍。这样大家能更直观地看到从看到现象到定位、定方案、改代码、验证收益一个完整的思路链是怎么走的。5.1 现象与初始排查某公司的离线报表作业日常跑半小时可以完成某天开始突然要跑3小时以上。打开Spark UI发现整个Job有4个Stage前面3个都很正常第4个Stage最后聚合只有200个Task。中位数Duration是8秒但第4个Stage中有8个Task的Duration超过1800秒并且Shuffle Read Size惊人地达到了2GB以上其他大多数Task只有几十MB。我当时第一反应就是看第4个Stage的SQL逻辑。查询大概长这样SELECT province, COUNT(DISTINCT user_id) AS uv FROM user_visit_log GROUP BY province粗看就是一个按省份统计UV的常规需求但“某个特殊省份”的数据量确实比其他省份高几个数量级日志里99%的数据来源都指向了头部渠道的自动化测试访问。所以GROUP BY province之后该省份的数据几乎全压在一个Partition上那个Task自然卡死了。5.2 根因确认与两种方案的对比这一步很关键。我先跑了一个探查SQLSELECT province, COUNT(1) AS cnt FROM user_visit_log GROUP BY province ORDER BY cnt DESC LIMIT 10结果果不其然头部那个省份的数据量占了全量的60%。于是确认这是典型的数据分布长尾导致的GROUP BY聚合倾斜。接下来我做了方案对比。虽然AQE的skewJoin只针对Join场景但Spark 3.0之后的group-by倾斜没有专门的优化机制所以这里还是得靠手动加盐两阶段聚合。起初我担心SQL改写太复杂但后来发现其实用临时表加两个子查询就能搞定根本不伤筋动骨。5.3 加盐重构后的最终SQL与收益我重写后的SQL大致如下简化版WITH salted AS ( SELECT CASE WHEN province 热点省份 THEN CONCAT(province, _, FLOOR(RAND() * 100)) ELSE province END AS salted_province, user_id FROM user_visit_log ), partial_agg AS ( SELECT salted_province, COUNT(DISTINCT user_id) AS partial_uv FROM salted GROUP BY salted_province ), final_agg AS ( SELECT SPLIT(salted_province, _)[0] AS province, SUM(partial_uv) AS uv FROM partial_agg GROUP BY SPLIT(salted_province, _)[0] ) SELECT province, uv FROM final_agg这里有个细节先对热点省份加盐再把加盐后的Key做COUNT(DISTINCT)去重。因为加盐后的Key保证同一user_id会被打散到最多一个盐桶不会出现DISTINCT跨桶重复计算的问题。而对非热点省份保持原key不变避免无谓的Shuffle。这次改造后该作业从3小时缩短到40分钟左右。后面几次运行即便数据量再涨也没有出现任务卡死的现象。为什么选100这个盐数因为当时该热点省份的数据量约30GB加盐100后热点省份被分到100个Task分散承担每个Task约300MB其他省份本身量小计算压力可以接受整体负载就均衡了。如果热点数据再大我会把盐加到200或300但没必要无限加。6. 避坑指南与排错速查方案都在上面了但真正做好还得把“坑”提前踩平。我挑几个强调一下这几个坑都是我自己或身边同事在线上真实翻过车的。6.1 加盐之后别忘了处理NULL值和枚举值加盐过程中最常见的坑就是“NULL值被意外打散”。很多业务表里关联键是NULL如果你直接CONCAT(key, _, rand())那么NULL会被拼成null_123之类的东西然后所有NULL被随机分到100个桶最后聚合结果可能把原本应该在同一组的数据拆散或导致回填错误。我的习惯是NULL值单独保留一组不参与加盐直接用一个固定后缀或常量Key标记最后统一聚合。同样对“未知”、“其他”、“0”这类已经是收敛值的Key加盐不会帮助提升性能反而会把语义错误地放大。实际编码时可以用WHEN col IS NULL THEN NULL_GROUP做分支。6.2 动态分区裁剪与广播叠加时的顺序问题如果你的SQL中同时有多个大表并且带有分区过滤条件想用广播时一定要先确认优化器的执行计划。很多时候你以为你加了Broadcast Hint但实际Spark还是把它当成SortMergeJoin来跑因为AQE在运行时发现小表数据大于设定的阈值Hint被忽略。可以在执行前用EXPLAIN或EXPLAIN COST看下执行计划。如果发现广播节点没生效就直接在SQL外层再用子查询缓存一下先把“小表的过滤结果”物化到内存再Join。这也是为什么我说“完全没有必要执着于某个语法”执行计划才是指导原则。6.3 不要迷信AQE理解它的边界AQE是个好工具但它不是银弹。我上面说过它只优化Reduce端读取倾斜不解决写倾斜并且它依赖Shuffle后的统计信息如果你的Shuffle数据是加密传输或做了压缩格式特别扭曲它的判断可能失真。另外AQE如果你开了coalescePartitions在读取倾斜数据时可能因为合并策略把多个中等分区合并成大分区反而引入新的不均衡。这个时候你要打开spark.sql.adaptive.coalescePartitions.parallelismFirst或调低minPartitionSize别让它过度合并。对于Spark 2.4及以下版本AQE不可用。这时候如果遇到Join倾斜只能老老实实手动加盐或者用广播。我在旧集群上也维护过一套“动态盐值”方案运行时先采样算出热点Key再生成盐值然后注册临时函数。效果与AQE相当但代码复杂度高一截。6.4 常见问题速查表现象可能原因优先排查思路解决方案某个Task极慢Shuffle Read大GroupBy或Join键分布不均衡查看Spark UI的Shuffle Read Size排名加盐两阶段聚合 / AQE SkewJoin大量小Task堆积磁盘IO高HDFS/表源小文件过多检查Input总大小与Task个数合并小文件 / 调大maxPartitionBytesExecutor反复OOM广播表太大或单Task数据量过高看GC日志与Executor内存配置缩小广播阈值 / 加盐任务卡在最后合并阶段数据倾斜叠加SortMergeJoin查看Stage依赖和Join方式改为Broadcast或加盐任务迁移到新集群后变慢AQE参数阈值不当对比新老集群默认配置调大/调小coalesce与skew阈值偶尔倾斜时好时坏数据分布随上游调度变化多采集几天的样本数据预设热点Key动态加盐这个表格是我自己电脑里常备的每次接到排障任务先套一遍至少能省出半小时的盲目摸索时间。7. 后续还可以做的优化方向倾斜问题解决完之后不要觉得就结束了。我在实际运行中还会接几个后置动作让系统更稳定。第一个是把热点Key列表缓存到Redis或配置中心每天业务跑批前动态更新一次盐值。这样即使热点省份换了也能自动适配不用每次手动改代码。第二个是在SQL解析层做一层安全网比如设置任务最长运行预警如果某个Stage的预计执行时间超过阈值就触发熔断避免坏任务一直占着资源。第三个是定期做Shuffle数据的采样分析用DataFrame的sample方法抽取一定比例算一下每个Key的比重找出Top N热点对下一天的优化策略做参考。最后再分享个小技巧排查倾斜时优先去看“同一个Stage内Task耗时的方差”这比直接看平均值准确得多。均值被大量快速任务拉低了只有方差能暴露那几条“长尾”的真实存在。甚至可以写个小脚本定时拉取Spark Rest API里每个Task的Duration输出标准差超过阈值自动告警。这样不等用户来反馈“作业卡死”你就能提前处理。这个内容后续你还可以在数据同步链路中去验证比如基于Kafka实时流里的热点用户做动态分桶逻辑相通但实现细节更多那是另一个话题了。