前言：从"能跑"到"跑得好"

写Spark代码分三个境界：

第一境界：代码能跑起来（入门）
第二境界：代码跑得又快又稳（进阶）
第三境界：代码跑崩了知道怎么修（高手）

大多数人停留在第一境界，能跑就行，跑不动就加内存。

但当你的任务从处理1GB变成处理1TB，从跑1分钟变成跑1天，你就会发现：不懂原理的Spark，就像不懂发动机的老司机，迟早要在高速上抛锚。

本篇是Spark系列的第三篇，我们将深入Spark内核，从Job到Task，从Shuffle到内存管理，彻底搞懂Spark是怎么跑的，以及怎么让它跑得更快。

一、Spark作业执行全流程

1.1 从代码到执行

1.2 Job、Stage、Task的关系

关系总结：

• 一个Application包含多个Job
• 一个Job由一个Action操作触发
• 一个Job包含多个Stage
• Stage之间以Shuffle为边界划分
• 一个Stage包含多个Task
• Task数量 = 该Stage最后一个RDD的分区数

1.3 Stage划分原理

Stage划分的关键：宽依赖（Shuffle）

窄依赖：一个父分区对应一个子分区（map、filter、union）
宽依赖：一个父分区对应多个子分区（groupByKey、reduceByKey、join）

// 示例：这段代码会产生几个Stage？
val result = sc.textFile("data.txt")      // Stage 1 开始
  .flatMap(_.split(" "))                   // 窄依赖
  .map(word => (word, 1))                  // 窄依赖
  .reduceByKey(_ + _)                      // 宽依赖！Stage 1 结束，Stage 2 开始
  .filter(_._2 > 10)                       // 窄依赖
  .collect()                               // Action，触发Job

// 答案：2个Stage
// Stage 1: textFile -> flatMap -> map
// Stage 2: reduceByKey -> filter -> collect

1.4 DAGScheduler工作流程

二、Shuffle深度剖析

2.1 什么是Shuffle？

Shuffle是分布式计算的性能杀手，它涉及：

• 数据序列化
• 磁盘IO
• 网络传输
• 数据反序列化

2.2 Sort-Based Shuffle

Spark 2.0后默认使用Sort-Based Shuffle。

2.3 Shuffle Write详解

2.4 Shuffle Read详解

2.5 Shuffle调优参数

// 优化配置示例
spark.conf.set("spark.shuffle.file.buffer", "64k")
spark.conf.set("spark.reducer.maxSizeInFlight", "96m")
spark.conf.set("spark.shuffle.io.maxRetries", "6")
spark.conf.set("spark.shuffle.compress", "true")

三、内存管理机制

3.1 Executor内存布局

3.2 内存计算示例

假设：spark.executor.memory = 10GB

预留内存 = 300MB（固定）
可用内存 = 10GB - 300MB ≈ 9.7GB

统一内存 = 9.7GB × 0.6 ≈ 5.8GB
├── 执行内存（初始）= 5.8GB × 0.5 ≈ 2.9GB
└── 存储内存（初始）= 5.8GB × 0.5 ≈ 2.9GB

用户内存 = 9.7GB × 0.4 ≈ 3.9GB
└── 用户代码中创建的对象

3.3 统一内存管理

关键规则：

• 执行内存和存储内存可以互相借用
• 存储内存被执行内存借用后，可以被强制释放
• 执行内存被存储内存借用后，不能被强制释放（等任务完成）

3.4 堆外内存

// 开启堆外内存
spark.conf.set("spark.memory.offHeap.enabled", "true")
spark.conf.set("spark.memory.offHeap.size", "4g")

3.5 OOM问题排查

四、数据倾斜深度解决

4.1 什么是数据倾斜？

4.2 如何发现数据倾斜？

4.3 八大解决方案

方案一：过滤异常Key

// 场景：大量NULL或异常值
// 解决：直接过滤

val df = spark.read.parquet("data")
  .filter($"key".isNotNull)
  .filter($"key" =!= "")
  .filter($"key" =!= "NULL")

方案二：提高并行度

// 场景：数据量大但分区少
// 解决：增加分区数

// 方式1：设置全局参数
spark.conf.set("spark.sql.shuffle.partitions", "1000")

// 方式2：repartition
df.repartition(1000)

// 方式3：重新分区到特定列
df.repartition(1000, $"key")

方案三：双重聚合（加盐）

// 场景：聚合操作数据倾斜
// 解决：两阶段聚合

// 原始倾斜代码
val result = df.groupBy("city").agg(sum("amount"))

// 优化后
import scala.util.Random

val salted = df.withColumn("salted_city", 
  concat($"city", lit("_"), lit(Random.nextInt(100))))

val partialAgg = salted
  .groupBy("salted_city")
  .agg(sum("amount").as("partial_sum"))

val finalResult = partialAgg
  .withColumn("city", split($"salted_city", "_")(0))
  .groupBy("city")
  .agg(sum("partial_sum").as("total_amount"))

方案四：Broadcast Join

// 场景：大表Join小表，小表倾斜
// 解决：广播小表，避免Shuffle

import org.apache.spark.sql.functions.broadcast

// 小表（< 10MB）
val smallDF = spark.table("dim_city")
// 大表
val largeDF = spark.table("fact_orders")

// 广播小表
val result = largeDF.join(broadcast(smallDF), Seq("city_id"))

// 调整广播阈值
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100MB")

方案五：倾斜Key单独处理

// 场景：只有少数Key倾斜
// 解决：分离处理

// 找出倾斜的Key
val skewedKeys = df.groupBy("key")
  .count()
  .filter($"count" > 1000000)
  .select("key")
  .collect()
  .map(_.getString(0))
  .toSet

val skewedKeysBC = spark.sparkContext.broadcast(skewedKeys)

// 分离数据
val normalData = df.filter(!$"key".isin(skewedKeys.toSeq: _*))
val skewedData = df.filter($"key".isin(skewedKeys.toSeq: _*))

// 正常数据正常处理
val normalResult = normalData.groupBy("key").agg(sum("value"))

// 倾斜数据特殊处理（如加盐、或提高并行度）
val skewedResult = skewedData
  .repartition(1000)
  .groupBy("key")
  .agg(sum("value"))

// 合并结果
val finalResult = normalResult.union(skewedResult)

方案六：使用Map端聚合

// 场景：reduceByKey类操作
// 解决：使用reduceByKey而非groupByKey

// ❌ 不推荐：groupByKey（无Map端聚合）
rdd.groupByKey().mapValues(_.sum)

// ✅ 推荐：reduceByKey（有Map端聚合）
rdd.reduceByKey(_ + _)

// ✅ 推荐：aggregateByKey
rdd.aggregateByKey(0)(
  (acc, value) => acc + value,  // 分区内聚合
  (acc1, acc2) => acc1 + acc2   // 分区间聚合
)

方案七：调整Join策略

// Skew Join（Spark 3.0+ AQE）
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

方案八：自定义Partitioner

// 场景：需要精确控制数据分布
// 解决：自定义分区器

class CustomPartitioner(numParts: Int) extends Partitioner {
  override def numPartitions: Int = numParts
  
  override def getPartition(key: Any): Int = {
    val k = key.asInstanceOf[String]
    if (k == "热点Key") {
      // 热点Key随机分配到多个分区
      Random.nextInt(numParts / 10)
    } else {
      // 其他Key正常Hash
      Math.abs(k.hashCode % numParts)
    }
  }
}

rdd.partitionBy(new CustomPartitioner(1000))

4.4 方案选择指南

五、Spark UI深度解读

5.1 UI页面结构

5.2 关键指标解读

Jobs页面

Job 0 (成功)
├── 描述: count at MyApp.scala:25
├── 提交时间: 2024-01-15 10:30:00
├── 持续时间: 2.5 min
├── Stages: 3/3 (成功)
└── Tasks: 1000/1000 (成功)

Stages页面

Stage 2 (成功)
├── 描述: collect at MyApp.scala:30
├── 持续时间: 1.5 min
├── Input: 10 GB / 1000 records
├── Output: 100 MB / 100 records
├── Shuffle Read: 5 GB
├── Shuffle Write: 2 GB
└── Tasks: 
    ├── 成功: 200
    ├── 失败: 0
    ├── 总计: 200
    └── 时间分布:
        ├── 最小: 2s
        ├── 中位数: 30s
        ├── 最大: 5min  ← 可能数据倾斜!
        └── 75%分位: 45s

5.3 问题诊断流程

5.4 SQL页面解读

Query 0
├── 描述: SELECT city, sum(amount) FROM orders GROUP BY city
├── 提交时间: 2024-01-15 10:30:00
├── 持续时间: 30s
├── 扫描行数: 1,000,000,000
├── 输出行数: 100
└── 执行计划:
    *(2) HashAggregate(keys=[city], functions=[sum(amount)])
    +- Exchange hashpartitioning(city, 200)
       +- *(1) HashAggregate(keys=[city], functions=[partial_sum(amount)])
          +- *(1) FileScan parquet [city, amount]
             PartitionFilters: []
             PushedFilters: []
             ReadSchema: struct<city:string,amount:decimal(10,2)>

关键点：

• Exchange表示Shuffle
• HashAggregate分两阶段（partial_sum是Map端聚合）
• PushedFilters显示下推的过滤条件

六、实战调优案例

6.1 案例一：百亿数据Join优化

问题：两个大表Join，跑了6小时还没完成

// 原始代码
val orders = spark.table("orders")      // 100亿条
val users = spark.table("users")        // 10亿条
val result = orders.join(users, Seq("user_id"))

分析：

1. 两张都是大表，无法Broadcast
2. Shuffle数据量巨大
3. 存在数据倾斜（大V用户订单多）

优化方案：

// 1. 开启AQE
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

// 2. 增加Shuffle分区
spark.conf.set("spark.sql.shuffle.partitions", "2000")

// 3. 分离倾斜Key（大V用户）
val topUsers = orders.groupBy("user_id")
  .count()
  .filter($"count" > 1000000)
  .select("user_id")

val topUsersBroadcast = broadcast(
  users.join(topUsers, Seq("user_id"))
)

// 普通用户正常Join
val normalOrders = orders.join(
  topUsers.select($"user_id", lit(1).as("is_top")),
  Seq("user_id"),
  "left_anti"  // 排除大V
)
val normalResult = normalOrders.join(users, Seq("user_id"))

// 大V用户Broadcast Join
val topOrders = orders.join(
  topUsers.select($"user_id", lit(1).as("is_top")),
  Seq("user_id"),
  "left_semi"  // 只保留大V
)
val topResult = topOrders.join(topUsersBroadcast, Seq("user_id"))

// 合并结果
val finalResult = normalResult.union(topResult)

效果：从6小时优化到30分钟

6.2 案例二：OOM问题解决

问题：Executor频繁OOM

java.lang.OutOfMemoryError: Java heap space

分析步骤：

// 1. 查看数据量
df.count()  // 10亿条
df.rdd.partitions.size  // 只有100个分区！

// 2. 计算每个分区数据量
10亿 / 100 = 1000万条/分区
// 假设每条1KB，每个分区约10GB！

优化方案：

// 1. 增加分区数
df.repartition(2000)  // 增加到2000个分区

// 2. 增加Executor内存
--executor-memory 8g

// 3. 增加Executor数量
--num-executors 100

// 4. 减少单个Executor的核数（减少并发）
--executor-cores 2

// 5. 使用更高效的序列化
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

6.3 案例三：小文件问题

问题：读取10万个小文件，任务启动就花了30分钟

分析：

• 每个小文件一个Task
• Task调度开销远大于实际处理时间

优化方案：

// 1. 合并小文件读取
spark.conf.set("spark.sql.files.maxPartitionBytes", "128MB")
spark.conf.set("spark.sql.files.openCostInBytes", "4MB")

// 2. 预处理：合并小文件
val df = spark.read.parquet("small_files/*")
df.coalesce(100).write.parquet("merged_files")

// 3. 使用CombineFilesInputFormat（底层优化）
spark.conf.set("spark.hadoop.mapreduce.input.fileinputformat.split.minsize", "134217728")

七、调优参数速查表

7.1 资源配置

7.2 Shuffle配置

7.3 内存配置

7.4 SQL配置

八、写在最后

调优是一门艺术，没有银弹，只有不断尝试。

记住几个原则：

1. 先跑通，再优化：别一上来就调参数
2. 数据说话：用Spark UI分析，不要猜
3. 从大到小：先解决最大的瓶颈
4. 记录变化：每次调优记录参数和效果

最后送大家一句话：

调优的最高境界是不用调优——设计好的数据架构和代码，比任何参数都重要。

本文作者：一个与OOM搏斗多年的老兵

最惨经历：调了一天参数，最后发现是代码死循环

附录：面试高频题

1. Spark的Stage是怎么划分的？

   从后往前回溯，遇到宽依赖（Shuffle）就切分Stage。窄依赖在同一Stage内。

2. Spark的内存管理是怎样的？

   统一内存管理：执行内存和存储内存共享，可互相借用。执行内存可强制释放存储内存。

3. 什么是数据倾斜？怎么解决？

   某个Key数据量远大于其他Key，导致Task执行时间不均衡。解决：加盐、Broadcast Join、AQE等。

4. Shuffle的过程是怎样的？

   Map端：分区→排序→溢写→合并；Reduce端：拉取→合并→聚合。

5. 如何定位Spark性能问题？

   通过Spark UI查看Job/Stage/Task执行情况，关注时间分布、Shuffle数据量、GC时间等指标。

6. AQE有什么作用？

   运行时动态优化：合并小分区、切换Join策略、处理数据倾斜。