基于本文回答

播面 播面

文解图解播客视频，多维讲透八股文
0
评论
请从左侧选择一个分类

SparkSQL中Executor 频繁磁盘溢写（Spill to Disk）导致吞吐量极低

在 Spark SQL 中，Executor 频繁出现磁盘溢写（Spill to Disk）是一个非常典型的性能瓶颈。

什么是 Spill to Disk？

当 Spark 执行 Shuffle、关联（Join）、排序（Sort）或聚合（Aggregation）等操作时，需要在 Executor 的 Execution Memory（执行内存）中暂存和处理数据。如果当前 Task 处理的数据量超出了该 Task 分配到的执行内存上限，为了防止发生内存溢出（OOM），Spark 会启动保护机制：将内存中一部分数据序列化并写入本地磁盘临时目录（Spill），待需要时再从磁盘读取、反序列化并合并。

因为磁盘 I/O 读写以及大量的序列化与反序列化操作，其性能比纯内存计算慢数个数量级，因此频繁的 Spill 会导致任务运行极慢，集群整体吞吐量急剧下降。

一、 频繁 Spill 的常见诱因

  1. Shuffle 分区数（Partition）设置过小
    默认的 spark.sql.shuffle.partitions 为 200。如果处理的数据量达到百 GB 甚至 TB 级别，每个分区承载的数据量（特别是反序列化成 JVM 对象后）将非常庞大，极易撑爆执行内存。
  2. 每个 Executor 的 Core（核）数过多，导致内存竞争
    如果设置了较高的 spark.executor.cores（如 8 或 16），这意味着一个 Executor 内部会同时并行运行 8 或 16 个 Task，它们会共享同一个 JVM 堆中的执行内存。平均到每个 Task 的内存极其微小，极易触发 Spill。
  3. 数据倾斜（Data Skew）
    如果部分 Key 的数据量远超其他 Key，会导致某些特定 Task 分配到的数据量超载，造成“长尾效应”，并在这些 Task 上发生严重的 Spill。
  4. 内存配置比例不合理
    如果使用了大量的 cache() 或 persist() 占用了过多的 Storage Memory（存储内存），会导致留给 Shuffle/Sort 运算的 Execution Memory 被严重挤压。
  5. 代码层面的“内存膨胀”操作
    使用 explode() 展开大数组、使用了无边界的窗口函数（Window Func）、或者在 RDD 算子中误用 groupByKey（未在 Map 端预聚合）等，会导致内存瞬间爆满。

二、 针对性的解决与优化方案

1. 调整 Shuffle 分区数，减小单 Task 数据量

提升分区数是解决 Spill 最直接、低成本的方法：

  • 手动增加分区数：根据总体 Shuffle 数据量调整 spark.sql.shuffle.partitions。通常建议将单分区的数据量控制在 100MB ~ 200MB 左右。
    plaintext 
    spark.sql.shuffle.partitions = 1000  # 根据实际数据量适当调大
  • 利用 AQE 自动合并/拆分分区：确保开启 AQE 的自适应分区合并功能，使 Spark 能够根据运行时实际数据大小自动分配合理的分区数。
    plaintext 
    spark.sql.adaptive.enabled = true
spark.sql.adaptive.coalescePartitions.enabled = true

2. 优化 Executor 的内存与 Core 的配比

避免单 Executor 运行过多并发 Task，降低内存竞争：

  • 控制 spark.executor.cores：通常建议将每个 Executor 的核数控制在 3 到 5 个（过高会导致单个 Task 内存过小且 GC 压力大，过低则无法利用 JVM 内部共享数据的优势）。
  • 合理增大 spark.executor.memory。
  • 调整内存划分比例：如果任务中几乎不使用 cache()，可以适当降低 Storage 内存的占比，将更多空间留给 Execution 内存：
    • spark.memory.fraction（默认 0.6）：统一内存占 JVM 堆的比例。
    • spark.memory.storageFraction（默认 0.5）：Storage 占统一内存的比例。若不需要缓存，可尝试将此值调低（例如 0.2 或 0.1），把空间释放给 Execution。

3. 缓解与治理数据倾斜

如果 Spark UI 显示只有少数 Task 发生了大规模 Spill，则极大概率存在数据倾斜：

  • 开启 AQE 倾斜关联优化：
    plaintext 
    spark.sql.adaptive.skewJoin.enabled = true
    开启后，Spark 会在运行期检测倾斜的分区，并将其拆分成多个子分区进行 Join，能非常有效地缓解 Spill。
  • 大表 Join 小表时使用广播（Broadcast）：
    通过 Broadcast Hash Join 避免触发 Shuffle，从而消除由于 Shuffle 引起的 Spill。
  • 加盐（Salting）法：对倾斜的 Key 手动拼接随机前缀（如 0_key, 1_key），将数据打散后再进行聚合或关联，最后去除前缀。

4. 优化代码，避免高内存消耗算子

  • 慎用 explode()：特别是对包含大 Map 或大 Array 的列，explode 会导致行数呈指数级增加，在内存中产生巨量对象。
  • 优化 Window 函数的边界：
    避免使用无边界的窗口（如 ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING），尽量明确窗口边界（如 ROWS BETWEEN 10 PRECEDING AND CURRENT ROW），防止整组数据必须一次性载入内存。
  • 用 reduceByKey/aggregateByKey 代替 groupByKey：
    如果在 SQL 之外使用了 RDD 算子，尽量避免使用 groupByKey。reduceByKey 会在 Map 端进行本地预聚合（Map-side Combine），能极大减少溢写到磁盘的数据量。

5. 开启更高效的序列化器

默认的 Java 序列化效率低、占用空间大。切换到 Kryo 序列化器可以降低对象在内存和溢写磁盘时的占用空间，间接缓解内存压力：

plaintext 
spark.serializer = org.apache.spark.serializer.KryoSerializer

三、 总结排查步骤

  1. 观察 Spark UI：打开产生溢写的 Stage，对比 Spill (Memory) 和 Spill (Disk)。
  2. 判断是“整体过载”还是“单点倾斜”：
    • 若所有 Task 均有 Spill，说明整体分区数太小或 Executor 内存/核数配比不合理（优先尝试增加 Shuffle 分区数和调整 Executor 内存）。
    • 若仅有个别 Task 出现极大的 Spill，其他 Task 正常，说明存在数据倾斜（优先尝试AQE 倾斜优化或广播关联）。