数据倾斜这个话题只要是搞过分布式计算的人基本都绕不开。它不像那些报错信息一样直接摆在你面前而是悄无声息地把你的作业从5分钟拖到50分钟从跑得动变成跑不动甚至直接OOM。我最早接触数据倾斜的时候光看现象完全懵了明明集群资源看着没问题、数据量也没到瓶颈任务就是卡在某个Stage死活不动后来才慢慢搞明白这是“数据分布不均”惹的祸。这篇文章我想把我这些年踩过的坑、定位问题的方法、以及真正落地有效的解决方案整理出来。不管你是刚入门的新手还是已经被倾斜折磨过的老手按这个思路走一遍至少能解决90%的问题。1. 数据倾斜的底层原理到底为什么会出现这种问题很多人在实际开发中都遇到过这样一个场景明明一个Stage只有几百个任务整体进度卡在99%最后3个任务死活跑不完。这大概率就是数据倾斜了。1.1 数据分布的本质理解“木桶效应”数据倾斜的核心可以用一句话概括集群里各个节点的任务处理的数据量严重失衡某一个任务的负担远大于其他任务。就拿生活中的例子来说就好比往10个桶里倒100斤水正常情况每个桶10斤但如果其中有个桶被分配到了50斤另外几个桶才各分到5斤那么无论你等多久最后完成的一定是那个最重的桶。在分布式计算框架里每个任务处理的数据量取决于某个特定的key被分配到了哪个分区。比如我们在做Join的时候如果按照用户的ID来关联那么一个超级大V用户产生的所有数据都会被分到同一个任务里。这个任务要处理的数据量可能比别的任务高出几百倍于是整个作业的执行时间就被这个任务拖住了。1.2 常见产生数据倾斜的三种运营场景第一种Key分布极度不均匀。这是最常见也最直接的原因。比如日志数据里某个IP可能产生了海量访问记录电商订单里某个头部商家的订单量占了全量的30%。当这些数据作为聚合或关联的key时天然就会形成倾斜。第二种Join操作引发的数据膨胀。两个表做内连接时如果一个表里有大量重复的key那么在Shuffle阶段相同key的数据就会被打到同一个Reducer上。尤其是小表关联大表时如果小表的数据在关联字段上没有去重极容易把一个Reducer压垮。第三种过滤条件导致的数据分布失真。这个很多人容易忽略。数据源本身虽然分布均匀但经过一层过滤后部分key的数据被大量过滤掉而另一部分key的数据几乎没被过滤。最终落到Shuffle阶段的数据就变成了偏科生。1.3 从执行计划层面看倾斜的本质从引擎的视角来看数据倾斜的发生通常伴随着Shuffle操作。不管是MapReduce、Spark还是Flink只要涉及group by、join这类需要重新分配数据的操作就会经过一个Shuffle阶段。框架按照key的哈希值把数据路由到不同的下游任务里如果hash结果过度集中在某个区间那这个区间对应的任务就成了瓶颈。注意不是所有的数据量大都叫数据倾斜。如果集群所有任务都处理着差不多的数据量只是整体数据量太大了那是资源不足只有当某一个或某几个任务明显比其他任务处理的数据量多出一个量级以上才算真正的倾斜。2. 数据倾斜的故障现象与技术判断方法数据倾斜的故障现象其实非常好辨认难的是“是否真的是倾斜”以及“倾斜发生在哪个环节”。这一节我把典型的症状和对应的排查方法一起写清楚。2.1 现象盘点从时间特征到资源特征数据倾斜的主要现象通常在运行日志和监控面板上都会有明显特征现象一任务长时间挂起进度条卡死在99%。大多数分布式调度平台如Spark的Web UI都会显示每个Task的进度状态你会发现绝大部分Task已经Finished唯独两三个或一两个Task的状态始终处于Running而且Duration时间在持续拉长。这是倾斜最直观的表现。现象二部分节点的CPU和内存严重超负荷。虽然集群整体看起来利用率不高但你可以从监控面板上看到某台或少数几台机器的CPU打满、内存占用居高不下混得差的节点甚至直接把磁盘写爆。现象三频繁发生OOM内存溢出。单个Reducer因为要处理的数据量过大尽管堆内存已经调得很大仍然扛不住。日志中会出现类似“FetchFailed”、“Container killed on request”这类异常重试多次依然失败。现象四整个作业输出结果出现明显的Partition大小差异。比如任务最终写出了100个结果文件其中几个文件大小高达2GB而其他文件才几十KB。这种不平衡极其直观基本不用看日志就能判断。2.2 用执行计划数据确认倾斜位置光看现象只能确定“发生了倾斜”但我们要找到“倾斜在哪里”才能对症下药。我的建议是在开发阶段就把执行计划打印出来做分析。以Spark为例执行计划里的每个Stage都包含了Task数量和输入数据量等信息。如果某个Stage的输入数据量特别大而它的前一个Stage输出数据量很小那么这个Stage里大概率潜伏着倾斜。更通用一些的做法是在Web UI里查看每个Task的Shuffle Read Size如果出现几个Task的读取量以GB计、其余Task以MB计的情况那基本可以断定倾斜就发生在这个位置。2.3 一条亲身实践的定位路径我不建议新手一上来就封闭式地怀疑“是不是数据倾斜了”而是建议灵活运用以下三步定位路径打开运行日志或者监控面板找到耗时最长的Stage。卡点的Stage一定是最可疑的阶段。进入该Stage的任务列表按耗时排序观察头部的几个任务与尾部任务的耗时差异。如果时间差异在3倍以上需要继续看数据量的差异。进入任务详情看这个任务的输入数据量、Shuffle读写数据、GC时间等指标。如果数据量是其他任务的上百倍那没跑了就是倾斜。通过这套路径不仅能定位到是不是倾斜还能基本锁死是哪个Stage、哪个维度出的问题后面做优化就有方向了。3. 全场景解决方案针对不同原因对症下药数据倾斜的解决方案市面上聊了很多真正常用的、效果稳定的其实也就那么几类。最关键的是要根据具体场景选择合适的手段否则是隔靴搔痒。3.1 对两表Join的数据倾斜主流方案是加盐打散再聚合这类场景最常见于事实表和维表关联比如订单表和商家表做关联统计。倾斜的根源是大数据量的一方事实表在某几个key上集中了过多数据。这时如果直接到Reduce端做join必然会压垮某个reducer。方案加盐打散。先给数据量大的那方在key后面拼接一个随机前缀比如0到n之间的随机整数然后数据量小的那方对每条记录也拼接同样的前缀种子。因为加了前缀盐原本相同的key就被打散到了不同的reduce分区里Reduce端的压力被均匀分摊最后再把前缀去掉做一次二次聚合或join。需要注意这个方案有个关键细节加盐打散只适用于参与Join的key在语义上允许被拆开处理的情况。例如“按用户统计订单金额”把大用户的订单拆到多个reduce分别统计后再合并结果是完全正确的但如果是“根据用户表join订单表取用户余额”同一条用户记录拆散可能导致Join结果重复就需要配合小表广播或者避免加盐。3.2 对GroupBy聚合的数据倾斜两阶段聚合先局部后全局GroupBy聚合的倾斜和Join有所不同因为很多业务场景下聚合是可拆解的。最常规的做法就是两阶段聚合第一阶段先在key后面拼接随机数然后做一次局部聚合partial aggregate把大量相同前缀的数据先在本地合并第二阶段去随机前缀再做一次全量聚合。这里有个容易忽略的点第一阶段聚合并不能真正意义上减少需要Shuffle的数据总量因为加了随机数之后原始重复的key会被打散到不同的Task但每个Task都可能收到同一个原始key的数据。它的核心价值是把原本落在一个Reduce端的大数据块拆成了多份让每个Reduce都能分摊压力。如果业务场景不允许这种拆解比如聚合的结果需要精确到某条明细数据那可以考虑预聚合在数据源头或者Map端提前做一层过滤把能合并的先合并掉再从源头减少倾斜key的数据量。3.3 大表Join小表时用广播变量彻底消除Shuffle如果小表的数据量在GB级别以下具体量级取决于框架配置通常几百MB内那根本没必要让两个表走Shuffle Join。把维表做成广播变量在Map端完成关联压根就不会产生Shuffle自然也就不存在倾斜问题。以Spark为例参数spark.sql.autoBroadcastJoinThreshold的默认值是10MB超过这个阈值但又不是特别大的表可以手动调大阈值让优化器选择广播。当小表的数据量能塞进广播变量后整个Job甚至不需要Stage划分执行速度会有一个质的飞跃。3.4 单个倾斜Key的极端处理先剔除再单独跑最后合并有些场景下倾斜是由极个别超级热点key造成的比如某个热门商品在促销期间占据了99%的浏览量。对这种极端倾斜常规的加盐打散在先去重后二次聚合时第二阶段依然会倾斜。我个人常用的策略是热点key分离通过采样或者日志分析找出这些超级热门key然后在Map端把它们单独提取出来不走常规的Shuffle路径——要么单独建立一个任务去处理要么用独立的资源去跑最后再把结果Union回到主链路里。这样做的好处是倾斜的这部分数据有独立的计算资源不再占用全链路的时间整体效率非常明显。代价是需要多一点编码量并且热点key的判定标准比如出现次数超过多少算热点要结合业务确定。3.5 对动态分区写入的倾斜通过分区拆分或提高并行度处理还有一个经常被人忽略的场景数据写入Hive动态分区时发生倾斜。比如按日期分区写入某些日期的数据量比其他日期大很多导致写分区的任务长期运行。这类场景的解决方案相对直接适当增加分区写入的并行度调大分桶数或者设置任务并发尽量避免一个Task处理某个超量分区。另外可以考虑先落临时表再通过静态分区二次重分布让每个分区都按数据量人工控制写入路径。4. 高阶调优与参数配置从引擎层面压制倾斜除了业务侧的改写引擎参数层面的调优也同样重要。参数调优不是万能的但配合合理的业务逻辑改写可以起到“112”的效果。4.1 调整并行度与Reducer数量以Spark为例如果某个Stage的Task数量过少比如只有十几二十个那么即使数据量分布并不极端单个Task需要处理的数据也会变得巨大无形中放大了倾斜的影响。将spark.sql.shuffle.partitions适当调大比如从200调到500甚至1000可以让Shuffle后的数据粒度更细降低单个Reducer承受的压力。不过这里有个非常现实的问题并行度越大元数据开销也会变大。如果数据量本身并不大强行把并行度调到几千只会增加Shuffle过程中的文件数量和调度开销甚至导致作业变慢。我的建议是结合数据的实际条数和大小通常保证单个Reduce数据量在100MB以内是一个比较稳妥的参考值。4.2 开启AQEAdaptive Query Execution在使用较新的引擎时可以把自适应查询执行特性打开。这个机制会在运行时动态调整Reduce端的并行度自动合并小分区、拆解倾斜分区从而把倾斜问题放在引擎层去缓解。以Spark 3.x为例参数配置大致如下spark.sql.adaptive.enabledtrue spark.sql.adaptive.coalescePartitions.enabledtrue spark.sql.adaptive.skewJoin.enabledtrue spark.sql.adaptive.skewJoin.skewedPartitionFactor5 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes256MB开启后当某个分区的数据量比其他分区大出skewedPartitionFactor倍并且超过阈值时引擎会自动把该分区拆分为多个小分区来进行处理。这个方案对于Join场景下的倾斜非常有效但对GroupBy场景的作用相对有限。4.3 从资源角度兜底给“头部任务”留够空间即使前面所有方案都做了仍然可能存在个别任务的数据量无法进一步压缩的情况。这时候需要做的是保障资源兜底适当增大单个Executor的内存spark.executor.memory、调整Shuffle服务的重试次数spark.shuffle.io.maxRetries、以及增大文件打开上限等参数让头部任务不至于轻易被打挂。但这里我要说句真心话资源兜底是最后的保险不是首选方案。你不可能靠无限加内存来根治倾斜这就像一个人发高烧你退烧药得吃但不能只吃退烧药而不管感染源。5. 常见问题与排查技巧实录数据倾斜的优化是一个不断测试、观察、调整的过程。我把自己平时遇到过的、以及身边同事踩过的典型问题整理成了一个速查表遇到相似情况可以直接对上号。5.1 问题速查对照表现象特征推断原因优先尝试的解决方案99%卡住少量Task运行时间远超平均Shuffle后某几个key数据集中先看Shuffle Read Size尝试两阶段聚合或加盐某个Task频繁GC或OOM单个分区内存过大调大并行度、设置AQE倾斜自动拆分、增加Executor内存兜底结果文件大小严重不均写分区/写文件阶段倾斜检查是否动态分区写入人工控制输出并行度或静态分区改写Join场景下倾斜且小表够小未开启广播Join先尝试广播变量再看是否仍需加盐开启AQE后无明显改善AQE阈值配置不匹配调低skewedPartitionThresholdInBytes或检查是否属于GroupBy聚合加盐后结果正确性存疑拆开处理不合理改用支持局部合并的算法或考虑热点Key分离方案5.2 独家避坑实操心得关于加盐随机前缀的范围我强烈建议不要用太大。有些文章推荐随机到100但我自己实测下来随机数范围太大容易导致小表也拆得太散产生大量的小文件下游读取反而变慢。用10以内的随机数通常就能起到比较好的打散效果。这个数字可以根据数据量做一次实验调整但没必要一上来就疯狂加码。关于小表广播有个细节很多人会忽略广播变量不是只对大表关联小表才有意义两个中等表做Join时如果一方可以在Map端按需裁剪先过滤、再广播其实也能起到很好的效果。关键在于不要让广播变量携带大量无效字段尽量只保留Join Key和最终需要用到的列。关于Hive动态分区倾斜我发现它跟上面所有问题的表现很不一样它往往不是卡在某个Task运行很久而是整个作业都跑得动但各个Reducer的输出大小差异巨大最终导致后续任务全部堆积。遇到这种情况优先检查分区字段本身的分布如果某个分区下的数据量大到超出其他分区的几倍不妨在写入前先把这份超大分区的数据单独路由到另一个目录最后再合并。5.3 一套可复制的排查路径最后把我经常使用的排查路径直接兜出来你遇到类似问题完全可以照做打开任务列表把耗时Top5的任务记录下来。看这些任务的输入数据量、Shuffle读写量、GC时间三项指标对比平均值。如果读写量差异超过一个数量级确认倾斜记录下对应Stage编号和参与计算的字段。查看执行计划中该Stage之前的上游操作确认是Join引发的还是GroupBy引发的或者是否由读取源本身分布不均造成。根据原因选择对应方案Join场景优先试广播小表或者加盐大表GroupBy场景优先做两阶段聚合极端key用热点分离。这套流程我大概跑了上百次基本可以覆盖绝大多数倾斜场景。只要你对自己业务里的数据分布有一个合理预期用这套组合拳去处理问题基本都能在半天内解决。6. 写在最后倾斜问题的长期治理思路这一节不聊具体技术细节了说说我在实际项目中沉淀下来的一些总体体会。数据倾斜严格来说并不是一个“一次性修复”的问题而是伴随数据量增长和业务模式变化不断出现的。我以前负责过一个统计报表的链路一开始数据量很小怎么跑怎么稳后来业务扩张某个城市的数据量暴增原本跑得很正常的任务一夜之间开始倾斜。所以建议各位在项目设计初期就对高基数、高重复率、高热点的字段做一次数据画像分析提前预判哪些字段可能出现倾斜。另外养成做数据探查的习惯。每个任务上线前抽几分钟用简单的group by看看核心字段的分布对后续排查问题能省非常多时间。很多人都忽略这一步等线上出了问题才反过来查数据非常被动。还有一点是监控和告警。不要等任务挂在那里几个小时才去人工看尽量给关键作业配上运行时长、失败重试、分区倾斜率相关的告警规则让系统在倾斜刚刚冒头时就通知到你。这部分的收益是长期的尤其是你同时维护多条链路的时候。我自己的经验是数据倾斜优化的本质是对数据分布规律的深刻理解加上对引擎机制的灵活运用。如果你能把这两个维度都吃透遇到这类问题基本不会慌。以后再有同学问我怎么处理倾斜我经常说的一句话是先别看参数先看你的数据到底长什么样。