Spark-Scala实战：数据倾斜与Shuffle调优，让任务跑得稳

1. 从“能跑”到“跑得稳”：这一节到底在解决什么问题

Spark-Scala 实战走到第 11 节，通常意味着前面已经把 RDD、DataFrame、基础算子、简单的读写流程都摸过一遍了。很多人到这一步会有一个错觉：代码能跑通、结果能出来，就算学会了。但真正上过生产环境的人都知道，能跑和跑得稳之间隔着一整条鸿沟。这一节我想聊的核心，就是把这个鸿沟填上——围绕Spark-Scala 语言实战里最容易被忽略、却最影响实际交付的那部分内容：数据倾斜的处理、Shuffle 的调优、缓存策略的取舍，以及一套能落地的排查方法论。

先说清楚这个内容适合谁。如果你已经能写出groupByKey、reduceByKey、join这些操作，但一遇到大数据量就卡住、一遇到某个 task 跑几个小时就束手无策，那这篇就是写给你的。如果你还在纠结map和flatMap的区别，建议先回去补基础，这里的内容会让你有点吃力。我下面讲的所有东西，都基于一个前提：你手上有一套能跑的 Spark-Scala 代码，现在要让它在大数据量下依然稳。

为什么单独把这一节拎出来讲？因为在我接触过的项目里，百分之七八十的性能问题最后都指向同一个根源：数据分布不均导致的 Shuffle 代价失控。而 Scala 作为 Spark 的原生语言，它的函数式特性、隐式转换、集合操作，恰恰给了我们很多优雅的调优手段，但同时也埋了不少坑。比如groupByKey和reduceByKey在语义上看着差不多，实际在 Shuffle 阶段的网络传输量能差出一个数量级；再比如cache()和persist()用错了存储级别，内存直接爆掉。这些细节，官方文档不会掰开揉碎讲，只有踩过才知道疼。

所以这一节的整体思路是：先讲清楚 Shuffle 到底在干什么、为什么它是性能杀手，然后落到具体的算子选型和参数配置上，再给出一套数据倾斜的实战排查流程，最后补充缓存和序列化的取舍经验。每一块我都会说清楚“为什么这么做”，而不是甩给你一堆配置让你抄。抄配置是最没用的，换个数据规模、换个集群规格，抄来的参数可能反而拖慢速度。理解了原理，你才能自己算、自己调。

2. Shuffle 到底在干什么：把性能账算明白

2.1 一次 Shuffle 的完整生命周期

很多人对 Shuffle 的理解停留在“会走磁盘、会走网络，所以慢”。这个认知没错，但太粗了，粗到你没法据此做任何优化决策。我把一次 Shuffle 拆成几个阶段来讲，你跟着走一遍，就知道钱（或者说时间）花在哪了。

Shuffle 的触发点，是那些需要跨分区重新组织数据的操作，典型的就是groupByKey、reduceByKey、join、repartition、distinct。以reduceByKey为例，它的流程大致是：Map 端每个 task 先对自己分区内的数据做一次本地聚合（这一步叫 map-side combine），把相同 key 的 value 先合并，然后按照分区器（默认是 HashPartitioner）把结果写到磁盘上的 Shuffle 文件里，每个下游分区对应一个文件块。Reduce 端再通过网络把属于自己的那些文件块拉过来，做最终的聚合。

这里第一个关键点就出来了：map-side combine 是reduceByKey相对groupByKey最大的优势。假设你有 1 亿条记录，key 只有 1000 个，那reduceByKey在 Map 端就能把 1 亿条压成最多 1000 条（每个 key 一条中间结果），网络传输量直接降了几个数量级。而groupByKey不做这个预聚合，1 亿条原封不动地走网络。这个差距，在小数据上你根本感知不到，数据一上来就是小时级的差别。

再往下，Shuffle 文件是落在磁盘上的，这就涉及磁盘 I/O。Spark 默认的spark.shuffle.file.buffer是 32KB，这个缓冲区越大，写磁盘的次数越少，但太大会占内存。还有spark.reducer.maxSizeInFlight，控制 Reduce 端一次拉多少数据，默认 48MB，网络条件好的集群可以适当调大，减少拉取轮次。

2.2 用一组数字感受 Shuffle 的代价

光说原理不够直观，我拿一组模拟数据给你算笔账。假设一个任务处理 100GB 的原始数据，经过 Map 端处理后参与 Shuffle 的数据量是 60GB，集群有 20 个 executor，每个 executor 配 4 核。

如果不做任何优化，走默认配置，Shuffle 写磁盘 60GB，读磁盘 60GB，网络传输 60GB。假设磁盘顺序写速度 200MB/s、网络 1GB/s，光 Shuffle 这一段的理想耗时就是：写 300 秒 + 读 300 秒 + 网络 60 秒，加起来超过 10 分钟，这还没算序列化反序列化的 CPU 开销和可能的磁盘争用。而如果通过 map-side combine 把 Shuffle 数据压到 6GB，上面这些数字全部除以 10，一分钟出头就完事。

这组数字是我为了说明问题做的简化估算，真实环境受磁盘类型（SSD 还是机械盘）、网络拓扑、数据压缩比影响很大，但量级关系是成立的。你要养成的习惯是：看到任何一个会触发 Shuffle 的算子，先在心里估一下它的数据量级，再决定要不要优化、怎么优化。这个估算能力，比记住十个参数名有用得多。

2.3 分区数：最容易被忽视的那个旋钮

spark.sql.shuffle.partitions这个参数，默认值是 200。这个默认值坑了无数人。200 个分区意味着什么？如果你的 Shuffle 数据只有 2GB，那每个分区 10MB，task 数量不多不少还算合理；但如果你的 Shuffle 数据有 500GB，每个分区就是 2.5GB，单个 task 处理 2.5GB 数据，内存稍不注意就 OOM，而且 task 数量太少，集群的并行度根本吃不满，一堆核在那闲着。

反过来，分区数也不是越大越好。分区太多，每个 task 处理的数据量很小，task 启动和调度的开销占比就上来了，而且会产生大量小文件。我一般的经验法则是：让每个 Shuffle 分区的数据量落在 100MB 到 200MB 之间。按这个标准反推，500GB 的 Shuffle 数据，分区数应该在 2500 到 5000 之间。你可以用下面这个思路动态设置：

// 根据 Shuffle 数据量动态推算分区数，经验值：每分区 128MB val shuffleDataSizeMB = 500L * 1024 // 假设 500GB val targetPartitionSizeMB = 128 val partitions = math.max( (shuffleDataSizeMB / targetPartitionSizeMB).toInt, spark.sparkContext.defaultParallelism * 2 ) spark.conf.set("spark.sql.shuffle.partitions", partitions.toString)

注意这里我加了一个下限，取defaultParallelism * 2，防止数据量小的时候分区数太少导致并行度不足。这个下限的取法不是死的，你可以根据集群核数调整，核心思想是保证每个核至少能分到一两个 task，别让资源空转。

提示：spark.sql.shuffle.partitions只对 Spark SQL 和 DataFrame 的 Shuffle 生效，如果你用的是 RDD 的reduceByKey这类算子，分区数是由算子本身或者spark.default.parallelism决定的，别搞混了。

3. 算子选型：同样的逻辑，差出十倍性能

3.1 groupByKey、reduceByKey、aggregateByKey 的取舍

这三个算子经常被拿来比较，我把它们的差异整理成一张表，你对着看就清楚了。

groupByKey的问题不只是不做预聚合，更致命的是它会把同一个 key 的所有 value 拉到一个迭代器里。如果某个 key 对应几百万条数据，这个迭代器就能把 executor 的内存撑爆。我见过太多人用groupByKey之后接一个map去做聚合，这等于把reduceByKey的活手动干了一遍，还多付了全量网络传输的代价。

aggregateByKey是我个人最推荐的通用聚合算子，因为它足够灵活。举个求平均值的例子，用reduceByKey做不了（因为要同时维护 sum 和 count），但aggregateByKey可以：

val data = sc.parallelize(Seq(("a", 10.0), ("a", 20.0), ("b", 5.0), ("b", 15.0))) // 初始值 (sum, count)，分区内聚合，分区间聚合 val avg = data.aggregateByKey((0.0, 0))( (acc, v) => (acc._1 + v, acc._2 + 1), (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2) ).mapValues { case (sum, cnt) => sum / cnt }

这段代码的关键在于两个函数：第一个是分区内的聚合逻辑，第二个是跨分区的合并逻辑。两个函数都只维护一个固定大小的二元组，不管某个 key 有多少条数据，内存占用都是常数级的。这就是aggregateByKey相比groupByKey的本质优势——聚合状态的大小与数据量无关。

3.2 join 的三种形态与选择逻辑

join是另一个 Shuffle 重灾区。Spark 里常见的 join 实现有三种：Shuffle Hash Join、Sort Merge Join、Broadcast Join。前两种都要走 Shuffle，第三种不走。选择逻辑其实很简单：如果参与 join 的一张表足够小，能塞进每个 executor 的内存，就用 Broadcast Join。

Broadcast Join 的原理是把小表完整地广播到每个 executor 上，然后大表的每个分区在本地和小表做 join，完全避免了 Shuffle。触发方式有两种，一种是显式用broadcast()函数包一下，另一种是依赖spark.sql.autoBroadcastJoinThreshold参数（默认 10MB）自动判断。我一般倾向于显式指定，因为自动判断有时候会因为统计信息不准而失效：

import org.apache.spark.sql.functions.broadcast val bigTable = spark.read.parquet("/path/to/big") val smallTable = spark.read.parquet("/path/to/small") val result = bigTable.join(broadcast(smallTable), Seq("user_id"), "left")

这里有个坑要提醒：广播的表不能太大，默认上限是 8GB（spark.sql.broadcastTimeout和 executor 内存共同决定），超过这个量级广播本身就会 OOM。判断标准是：小表大小乘以 executor 数量，不能超过集群总内存的合理比例。比如小表 500MB，20 个 executor，那就要占 10GB 内存，如果每个 executor 只有 4GB 内存，这就很危险了。

如果两张表都很大，没法广播，那就只能走 Sort Merge Join。这时候优化的重点就回到分区数上——确保 join key 的分区数合理，避免某个分区数据量过大。另外，join 之前先做过滤，把不必要的数据干掉，这是最朴素也最有效的优化，很多人却总是忘。

3.3 一个真实的算子替换案例

我拿一个模拟项目里的场景来说。原始代码是这样的：

// 原始写法：groupByKey 后手动聚合 val userActions = rdd.groupByKey().mapValues { actions => val total = actions.map(_.amount).sum val count = actions.size (total, count) }

这段代码在数据量小的时候跑得挺欢，数据一上来就出问题。我把它改成aggregateByKey：

// 优化写法：aggregateByKey 预聚合 val userActions = rdd.aggregateByKey((0.0, 0L))( (acc, action) => (acc._1 + action.amount, acc._2 + 1), (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2) )

改动很小，但效果立竿见影。在一个模拟的 200GB 数据集上，原写法跑了将近 40 分钟，改完之后 12 分钟左右。差距的来源就是 map-side combine 把网络传输量压下去了。这个案例我反复讲，是因为它太典型了——很多性能问题不需要什么高深技巧，就是把算子用对。

4. 数据倾斜：那个让 task 卡几小时的元凶

4.1 怎么判断是不是数据倾斜

数据倾斜的典型症状是：绝大多数 task 几分钟就跑完了，但总有那么几个 task 卡在那里，进度条半天不动，最后要么超时失败，要么跑几个小时才结束。你在 Spark UI 的 Stage 页面看 task 的 Duration 分布，如果最大值和中间值差出一个数量级，基本可以确诊。

确诊之后，下一步是找到是哪个 key 在捣乱。最直接的办法是把 Shuffle 后的数据按 key 采样，统计每个 key 的记录数：

// 采样统计 key 分布，找出热点 key val sampled = rdd.sample(false, 0.01) // 采样 1% val keyCounts = sampled.map(x => (x._1, 1L)).reduceByKey(_ + _) val topKeys = keyCounts.takeOrdered(10)(Ordering.by(-_._2)) topKeys.foreach { case (k, c) => println(s"key=$k, sampledCount=$c") }

采样比例根据数据量调整，数据量特别大的时候 1% 甚至 0.1% 就够了。拿到热点 key 之后，你还要判断它是“真热点”还是“假热点”。真热点是这个 key 本身业务上就对应大量数据，比如某个默认值、某个测试账号；假热点往往是数据质量问题，比如大量记录的 key 是 null 或者空字符串。这两种情况的处理方式不一样，后面会分别讲。

4.2 两阶段聚合：给热点 key 加随机前缀

两阶段聚合是我处理热点 key 最常用的手段，思路是给热点 key 加上随机前缀，把它打散到多个分区先做一次局部聚合，然后再去掉前缀做全局聚合。打个比方，就像一个大队伍过独木桥，桥一次只能过一个人，那就把队伍拆成十个小队，分别从十个入口过，最后再汇合。

val prefixCount = 10 // 打散成 10 份 // 第一阶段：加随机前缀，局部聚合 val stage1 = rdd.map { case (key, value) => val prefix = scala.util.Random.nextInt(prefixCount) (s"${prefix}_$key", value) }.reduceByKey(_ + _) // 第二阶段：去掉前缀，全局聚合 val stage2 = stage1.map { case (keyWithPrefix, value) => val key = keyWithPrefix.split("_", 2)(1) (key, value) }.reduceByKey(_ + _)

这里prefixCount的取值有讲究。取太小，打散效果不够；取太大，会产生大量小 task，调度开销上来了。我的经验是取集群核数的 1 到 2 倍，比如 40 核的集群取 40 到 80。另外要注意，这个方法只适用于聚合类操作（求和、计数），如果是join或者需要保留明细的场景，得换思路。

4.3 空 key 和异常 key 的清洗

前面说的“假热点”，最常见的来源就是 null key 和空字符串 key。这类 key 往往占了总数据量的一大半，但它们对应的业务含义可能是“无效数据”。处理方式很简单，在 Shuffle 之前先过滤掉，或者给它们单独赋一个随机值：

// 方案一：直接过滤（如果业务上允许丢弃） val cleaned = rdd.filter { case (key, _) => key != null && key.nonEmpty } // 方案二：给空 key 加随机后缀，打散到不同分区 val scattered = rdd.map { case (key, value) => if (key == null || key.isEmpty) { (s"null_${scala.util.Random.nextInt(100)}", value) } else { (key, value) } }

方案一适合那些空 key 确实没用的场景，简单粗暴。方案二适合空 key 的数据还需要保留、只是不想让它们挤在一起的情况。我一般会先跟业务方确认空 key 的语义，别自作主张把数据丢了，这个坑我踩过——曾经把一个“空 key 代表默认配置”的数据过滤掉了，结果下游统计全错。

4.4 倾斜排查速查表

我把实际排查中遇到的典型情况和对应处理整理成一张表，方便你对照。

这张表不是万能的，但它覆盖了我遇到的大部分情况。排查的核心思路永远是：先定位是哪个 key、哪个分区出了问题，再对症下药，别上来就瞎调参数。

5. 缓存与序列化：内存这笔账要算清楚

5.1 cache 和 persist 的存储级别怎么选

cache()本质上是persist(StorageLevel.MEMORY_AND_DISK)的简写，但很多人不知道这一点，以为cache()就是纯内存。实际上当内存不够时，它会溢写到磁盘。存储级别的选择，核心是权衡“内存占用”和“重算代价”。

选择逻辑其实不复杂。如果数据集能完整放进集群内存，用MEMORY_ONLY，反序列化的对象访问最快。如果放不下，用MEMORY_AND_DISK_SER，序列化能省不少内存，代价是每次访问要反序列化，多花点 CPU。我一般默认用MEMORY_AND_DISK_SER，因为生产环境很少有“内存随便用”的好事。

这里有个关键判断：这个 RDD 或 DataFrame 到底会不会被复用。如果只用一次，缓存纯属浪费内存，反而拖慢整体速度。只有被多次 action 触发的数据集，缓存才有意义。我见过有人给每个中间结果都加cache()，结果内存被塞满，频繁 GC，性能比不缓存还差。

5.2 Kryo 序列化：一个几乎必开的开关

Spark 默认用的是 Java 序列化，体积大、速度慢。Kryo 序列化能把序列化后的体积压到 Java 序列化的几分之一，速度也快不少。开启方式很简单：

val conf = new SparkConf() .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .set("spark.kryo.registrationRequired", "false")

但有个细节要注意：如果你有自定义的类需要序列化，最好注册一下，否则 Kryo 每次都要写类名，反而浪费空间：

conf.registerKryoClasses(Array(classOf[MyRecord], classOf[MyConfig]))

spark.kryo.registrationRequired设为 true 会强制要求注册，没注册的类直接报错，这在开发阶段能帮你发现遗漏，但生产环境我一般设为 false，避免因为漏注册导致任务失败。这个取舍看你的团队规范，没有绝对的对错。

5.3 广播变量：小数据的大用处

除了 join 时广播小表，广播变量在别的场景也很有用。比如你有一份配置字典或者维度映射表，需要在每个 task 里查，如果每个 task 都去读一遍文件或者查一次数据库，开销就大了。用广播变量把它分发到每个 executor，只读一次，后续都从内存拿：

val dimMap = Map("001" -> "分类A", "002" -> "分类B") val broadcastDim = sc.broadcast(dimMap) val result = rdd.map { case (code, value) => val category = broadcastDim.value.getOrElse(code, "未知") (category, value) }

广播变量的坑在于：它一旦广播出去就不能修改，而且会一直占着 executor 的内存直到任务结束。所以广播的数据量要控制，别把几百 MB 的表随便广播，那会把 executor 内存吃光。

6. 一套能落地的调优流程与踩坑记录

6.1 调优的正确顺序：先看再调

很多人调优的顺序是反的——上来就改参数，改完发现没用，再改另一个，来回折腾。正确的顺序应该是：先定位瓶颈，再针对性调整，最后验证效果。

第一步，看 Spark UI。重点看三个地方：Stage 页面的 task 耗时分布（判断有没有倾斜）、Shuffle 的读写数据量（判断 Shuffle 是不是瓶颈）、Executor 页面的 GC 时间和内存使用（判断有没有内存压力）。这三个信息基本能定位百分之八十的问题。

第二步，根据定位结果选手段。如果是倾斜，走第 4 节的两阶段聚合；如果是 Shuffle 数据量大，检查算子选型，看能不能用reduceByKey替代groupByKey，能不能加 map-side combine；如果是内存压力大，考虑序列化、调整存储级别、增大分区数。

第三步，小数据量验证。改完之后别直接上全量数据跑，先用 1% 到 10% 的采样数据验证逻辑正确性和性能趋势，确认没问题再上全量。这一步能帮你省下大量等待时间。

6.2 我踩过的几个典型坑

第一个坑：盲目调大 executor 内存。早期遇到 OOM，我的第一反应是加内存，从 4G 加到 8G 加到 16G，结果还是 OOM。后来才发现根本原因是数据倾斜，某个 key 的数据量就是那么大，加多少内存都不够。正确的做法是先解决倾斜，而不是无脑加内存。

第二个坑：分区数设得过大。有一次为了“充分利用并行度”，把spark.sql.shuffle.partitions设成了 20000，结果每个 task 处理的数据只有几 MB，task 调度和启动的开销占了总时间的一大半，而且产生了海量小文件，下游读取的时候又慢又卡。分区数要匹配数据量，不是越大越好。

第三个坑：缓存了不该缓存的数据。有个任务里我对一个中间 DataFrame 加了cache()，但这个 DataFrame 其实只被用了一次，缓存反而占用了本该留给 Shuffle 的内存，导致频繁溢写磁盘。后来去掉cache()，性能反而提升了。缓存之前一定要问自己：这个数据真的会被复用吗？

第四个坑：忽略了数据本地性。Spark UI 里有个 Data Locality 的指标，如果大量 task 是ANY级别（意味着数据不在本地，要走网络），说明数据分布和计算分布不匹配。这种情况可以通过调整spark.locality.wait参数，让 Spark 多等一会儿本地数据，或者用repartition重新分布数据来改善。

6.3 参数配置的参考基线

最后给一份我常用的参数基线，注意这只是起点，具体值要根据你的集群规格和数据量调整，别照抄。

# 序列化 spark.serializer=org.apache.spark.serializer.KryoSerializer # Shuffle 相关 spark.sql.shuffle.partitions=1000 spark.shuffle.file.buffer=64k spark.reducer.maxSizeInFlight=96m spark.shuffle.compress=true spark.shuffle.spill.compress=true # 内存与 GC spark.memory.fraction=0.6 spark.memory.storageFraction=0.5 # 广播 spark.sql.autoBroadcastJoinThreshold=50m # 本地性等待 spark.locality.wait=3s

这份配置里，spark.shuffle.file.buffer从默认 32k 调到 64k，减少写磁盘次数；spark.reducer.maxSizeInFlight从 48m 调到 96m，减少网络拉取轮次；spark.sql.autoBroadcastJoinThreshold从 10m 调到 50m，让更多中等大小的表能走广播 join。每一项调整背后都有明确的意图，你调的时候也要清楚自己在调什么，而不是复制粘贴。

注意：spark.memory.fraction和spark.memory.storageFraction这两个参数直接影响执行内存和存储内存的划分，调错了容易引发频繁 GC 或者缓存失效。如果你不确定，先保持默认，等定位到明确的内存问题再动。

我个人在实际操作中的体会是，Spark 调优这件事，八成的收益来自把算子用对、把数据倾斜解决掉，剩下两成才是参数微调。很多人本末倒置，花大量时间研究参数，却忽略了最基础的算子选型。先把groupByKey换成reduceByKey或aggregateByKey，先把热点 key 打散，这些做完，性能往往已经有质的提升，参数调优只是锦上添花。另外，每次调优都要有对照，改之前记录一下耗时和资源使用，改之后再对比，别凭感觉说“好像快了”，数据不会骗人。