企业官网建设流程全解析

作为一个常年跟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 BY+COUNT/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.autoBroadcastJoinThreshold=20971520; -- 或者直接在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，但整张表几个TB，Spark不会载入全表去广播，但也不会自动选择那个“过滤后的小分区”。这种情况下，你要在Join前把需要广播的子集先过滤出来，Cache一下再Join，效果最好。

3.3 动态重分区配合AQE：让优化器替你分担压力

从Spark 3.0开始，AQE（Adaptive Query Execution，自适应查询执行）是我强烈建议开启的，因为它能在运行时动态调整Shuffle分区数，算是优化器层面的一个“后置兜底”。

开启方式：

set spark.sql.adaptive.enabled=true; -- 开启后建议保持默认，可以调整的参数如下 set spark.sql.adaptive.coalescePartitions.enabled=true; set spark.sql.adaptive.coalescePartitions.minPartitionNum=20; set spark.sql.adaptive.skewJoin.enabled=true; set spark.sql.adaptive.skewJoin.skewedPartitionFactor=5; set spark.sql.adaptive.skewJoin.skewedPartitionThreshold=256MB;

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.partitions=200，但实际上更好的是锚定集群规模来算。

我个人的计算公式是：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.enabled=true，并且保证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 根因确认与两种方案的对比

这一步很关键。我先跑了一个探查SQL：

SELECT 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 常见问题速查表

这个表格是我自己电脑里常备的，每次接到排障任务先套一遍，至少能省出半小时的盲目摸索时间。

7. 后续还可以做的优化方向

倾斜问题解决完之后，不要觉得就结束了。我在实际运行中还会接几个后置动作，让系统更稳定。第一个是把热点Key列表缓存到Redis或配置中心，每天业务跑批前动态更新一次盐值。这样即使热点省份换了，也能自动适配，不用每次手动改代码。第二个是在SQL解析层做一层安全网，比如设置任务最长运行预警：如果某个Stage的预计执行时间超过阈值，就触发熔断，避免坏任务一直占着资源。第三个是定期做Shuffle数据的采样分析，用DataFrame的sample方法抽取一定比例，算一下每个Key的比重，找出Top N热点，对下一天的优化策略做参考。

最后再分享个小技巧：排查倾斜时，优先去看“同一个Stage内Task耗时的方差”，这比直接看平均值准确得多。均值被大量快速任务拉低了，只有方差能暴露那几条“长尾”的真实存在。甚至可以写个小脚本，定时拉取Spark Rest API里每个Task的Duration，输出标准差，超过阈值自动告警。这样不等用户来反馈“作业卡死”你就能提前处理。

这个内容后续你还可以在数据同步链路中去验证，比如基于Kafka实时流里的热点用户做动态分桶，逻辑相通但实现细节更多，那是另一个话题了。