3. 数据倾斜的定位方法：通过Spark UI/MapReduce UI观察、查看Stage耗时、分析Task数据量、使用Accumulator定位

数据倾斜这个问题，说白了就是“木桶效应”的极端体现——整个集群都在等那个最慢的任务跑完。我见过不少团队，代码写得挺漂亮，一跑起来就卡在某个Stage上动弹不得。怎么快速定位？别急，我一步步说。

3.1 通过Spark UI / MapReduce UI观察

这是最直观的方法。我个人习惯，任务一提交就先打开UI页面盯着看。

Spark UI怎么看？

打开Spark UI，直奔Stages页面。你会看到每个Stage的进度条。如果某个Stage的进度条卡在99%不动，或者明显比其他Stage慢很多——嗯，大概率就是倾斜了。

点进那个慢的Stage，看Tasks列表。你会看到一堆任务，大部分几秒就跑完了，但有一两个任务耗时特别长。我遇到过最夸张的一次，99个任务跑了2秒，最后一个任务跑了40分钟。你想想看，这差距有多大。

关键指标：

  • Shuffle Read Size / Records：看每个Task读取的数据量。正常情况大家差不多，如果某个Task读取的数据量是其他Task的10倍以上，那就是倾斜。
  • Duration：看耗时。单个Task耗时远超中位数，基本可以确认。
  • GC Time：如果某个Task的GC时间特别长，说明它处理的数据量太大，内存吃紧。

MapReduce UI怎么看？

MapReduce的UI稍微老派一点，但原理一样。打开JobTracker或ResourceManager的页面，找到你的Job。看Map阶段和Reduce阶段的进度。

Reduce阶段最容易出问题。如果大部分Reducer都完成了，就剩一两个卡在33%、66%或者99%——这就是典型的倾斜。点进去看，那个Reducer处理的记录数是不是比其他多很多？

我的小技巧：在Spark UI里，按Shuffle Read Size排序，降序排列。最大的那个Task如果比第二名大一个数量级，不用怀疑，就是它了。

3.2 查看Stage耗时

Stage耗时是另一个重要线索。我一般会看两个东西：总耗时和Task耗时分布。

Stage总耗时

在Spark UI的Stages页面，每个Stage都有个Duration列。如果某个Stage的耗时占了整个Job的80%以上，那问题就出在这个Stage里。

举个例子：一个Job有5个Stage，前4个加起来跑了2分钟，第5个Stage跑了30分钟。你说问题在哪？

Task耗时分布

点进那个慢的Stage，看Tasks列表。注意看Duration这一列。我习惯用眼睛扫一下最大值和中位数。

  • 中位数：代表大多数任务的正常耗时
  • 最大值：代表那个倒霉蛋的耗时

如果最大值是中位数的几十倍甚至上百倍，那就是倾斜。我记得有一次，中位数是3秒，最大值是1200秒。差了400倍。当时我就知道，肯定是某个Key的数据量爆炸了。

注意：Stage耗时长不一定就是数据倾斜。也可能是数据量大、计算逻辑复杂、或者资源不足。要结合Task数据量一起看，才能下结论。

3.3 分析Task数据量

这一步是实锤。光看耗时还不够，得看数据量。数据量才是倾斜的根源。

怎么看数据量？

在Spark UI的Task详情里，有几个关键指标：

指标 含义 倾斜时的表现
Shuffle Read Size / Records Task从Shuffle中读取的数据量 某个Task的数据量是其他Task的10倍以上
Input Size / Records Task从数据源读取的数据量 某个Task的输入数据量异常大
Output Size / Records Task输出的数据量 某个Task的输出数据量异常大

我一般先看Shuffle Read Size。因为大多数倾斜都发生在Shuffle阶段。比如groupBy、join这些操作，最容易产生倾斜。

具体操作步骤

  1. 打开Spark UI，进入慢的Stage
  2. 点击Tasks标签
  3. 找到Shuffle Read Size列，点击排序
  4. 看最大值和最小值，算一下比例

如果比例超过10倍，基本可以确认倾斜。如果比例超过100倍，那问题很严重了。

我曾经遇到过一个案例：一个join操作，左边表有1亿条数据，右边表有100万条数据。按理说应该很快。结果跑了一个小时还没结束。一看UI，有个Task的Shuffle Read Size是2GB，其他Task只有几十MB。后来发现，右边表里有个Key是空字符串，占了80%的数据。这就是典型的倾斜。

3.4 使用Accumulator定位

有时候UI上的信息不够细。比如你想知道具体是哪个Key导致的数据倾斜。这时候就要用Accumulator了。

Accumulator是什么？

Accumulator是Spark提供的一种累加器。可以在Executor端累加数据，然后在Driver端读取。说白了，就是分布式环境下的计数器。

怎么用Accumulator定位倾斜？

我一般会写一个自定义的Accumulator，用来统计每个Key出现的次数。代码大概长这样：

// 自定义Accumulator，统计Key的分布
class KeyCountAccumulator extends AccumulatorV2[String, java.util.Map[String, Long]] {
  private val map = new java.util.HashMap[String, Long]()

  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[String, java.util.Map[String, Long]] = {
    val newAcc = new KeyCountAccumulator()
    newAcc.map.putAll(map)
    newAcc
  }

  override def reset(): Unit = map.clear()

  override def add(key: String): Unit = {
    map.put(key, map.getOrDefault(key, 0L) + 1L)
  }

  override def merge(other: AccumulatorV2[String, java.util.Map[String, Long]]): Unit = {
    other.value.forEach((k, v) => map.put(k, map.getOrDefault(k, 0L) + v))
  }

  override def value: java.util.Map[String, Long] = map
}

// 使用示例
val keyCountAcc = new KeyCountAccumulator()
sparkContext.register(keyCountAcc, "keyCountAcc")

df.foreach(row => {
  val key = row.getAs[String]("key")
  keyCountAcc.add(key)
})

// 在Driver端读取结果
val result = keyCountAcc.value
result.forEach((k, v) => println(s"Key: $k, Count: $v"))

跑完之后，看哪个Key的Count特别大。那个Key就是罪魁祸首。

我的建议：Accumulator不要在生产环境的大数据集上直接用。因为把所有Key都统计一遍，可能会把Driver端的内存撑爆。我一般先抽样，或者只统计Top N的Key。

另一种方式：用groupBy + count

如果你不想写Accumulator，也可以用Spark SQL的groupBy + count来统计Key的分布：

df.groupBy("key")
  .count()
  .orderBy(desc("count"))
  .show(10)  // 只看Top 10

这样也能快速找到倾斜的Key。不过要注意，如果数据量特别大，这个操作本身也可能触发Shuffle，导致新的倾斜。所以建议先过滤或者抽样。

避坑指南：我曾经在几十亿条数据上直接跑groupBy + count，结果又触发了一次倾斜，把集群搞崩了。后来学乖了，先抽样10%的数据，找到倾斜的Key，再针对性地处理。

总结一下定位流程

我一般按这个顺序来：

  1. 看UI：打开Spark UI，找到慢的Stage
  2. 看耗时：确认某个Task的耗时异常
  3. 看数据量：确认那个Task的Shuffle Read Size异常
  4. 找Key：用Accumulator或groupBy找到具体的倾斜Key

这四步走完，基本就能定位到问题所在。下一步就是怎么解决了——那是下一章的内容。

嗯，定位这一步其实不难，难的是你愿不愿意花时间去UI里翻一翻。很多人一看到任务慢，就想着加资源、加并行度。其实很多时候，找到那个倾斜的Key，对症下药，比加100个节点都管用。

← 上一章 📖 返回目录 下一章 →