PAYATHON 2026

Spark SQL 写入 Iceberg 时为何疑似重算而未复用缓存？

数据阿明

结论

这段代码中的 persist(StorageLevel.MEMORY_AND_DISK) 和 count() 是常见的缓存物化写法。后续 stage 再次出现 12 个分区，不能据此认定 Spark 忽略了缓存。

可以优先排查 staged_delta 的上游读取了 target_table，而随后的 MERGE 又修改了同一张表。写入后的缓存刷新可能导致这个依赖目标表的中间结果被重新计算。persist() 无法保证两个写入操作始终使用写入前的固定批次结果。

目前缺少 Iceberg 版本、物理执行计划和运行指标。这个排查方向有代码依据，但还不能认定为本次运行的最终根因。

为什么缓存可能重新计算

原示例中的依赖关系是：

source_staged ──┐
               ├─ LEFT JOIN → staged_delta → MERGE / INSERT
target_table ──┘                               │
      ↑                                       │
      └───────────────────────────────────────┘

current_target 是普通临时视图，创建视图并没有冻结目标表的数据。staged_delta 虽然被缓存，但它的逻辑计划仍包含对目标表的读取。

Spark 3.3.2 的 CacheManager 按逻辑计划的 sameResult 匹配缓存。它的 recacheByPlan() 会查找依赖指定计划的缓存，清除缓存数据并重建缓存计划。临时视图名称相同、Python 变量仍然存在，都不能说明已有缓存块一直有效。Spark 3.3.2 CacheManager 源码

Spark 的 DataSource V2 写入路径包含写入后刷新缓存的机制。以具体版本为例，Iceberg 1.3.1 的 Spark 3.3 扩展也会为行级写入调用 recacheByPlan()。这支持上述排查方向，但输入没有提供 Iceberg 版本，不能直接认定实际执行路径与该示例完全一致。Spark 写入执行源码、Iceberg 1.3.1 扩展源码

还需要区分缓存块丢失的情况。MEMORY_AND_DISK 允许把放不进内存的分区保存在磁盘上，所以“内存不够”不等于必然重算；如果可用缓存块丢失，Spark 仍可能沿 lineage 恢复数据。Spark 持久化说明

先确认扫描发生在哪里

缓存物化后、执行任何写入前，查看一个新查询的执行计划：

spark.sql("""
SELECT key_a, key_b, period_key, batch_ts, batch_id, delta_type
FROM staged_delta
WHERE delta_type = 'U'
""").explain(mode="formatted")

重点查看消费 staged_delta 的分支是否出现 InMemoryTableScan。随后执行原 MERGE，并在 INSERT 之前检查下面这个新查询：

spark.sql("""
SELECT *
FROM staged_delta
WHERE delta_type IN ('I', 'U')
""").explain(mode="formatted")

同时查看 Spark UI 的 SQL 执行详情和 Storage 页面：

  • 缓存物化后，预期分区是否已经缓存完整。
  • MERGE 前后，缓存块或缓存占用是否发生变化。
  • 可疑扫描读取的是原始源路径，还是 Iceberg 目标表。
  • 原始源扫描节点是否在该次执行中实际产生了读取指标。
  • 是否重新执行了构造 staged_delta 的 LEFT JOIN，而不是 MERGE 自身用于匹配目标记录的 join。

InMemoryTableScan 表明查询采用了缓存读取路径，但不能单独证明所有缓存块都已存在。InMemoryRelation 展示的缓存构建计划中也可能保留原始 scan 和 join；看到这些文字，不能说明它们在当前操作中重新执行了。

12 个分区同样只是一条线索。缓存物化本来就需要扫描源数据；MERGE 还需要读取目标记录，写入可能需要新的 Exchange 或排序。即使缓存了上游结果，整个写入过程仍可能发生扫描或 shuffle。

如果写入前的普通 SELECT 已经不能匹配缓存，再检查视图是否被替换、查询计划是否改变，以及是否调用过 unpersist()、clearCache() 或刷新操作。

让两个写入使用固定的中间结果

对于这个 SCD2 流程，可以在 MERGE 之前切断 staged_delta 对目标表的 lineage，再让两个写入读取该结果。

方案一：使用可靠 checkpoint

DataFrame.checkpoint(eager=True) 会立即写出 checkpoint，并返回截断逻辑计划的 DataFrame。必须使用返回的新对象。PySpark 3.3.2 checkpoint 文档

在原代码完成 staged_delta.count() 后、创建 close_source 前，加入：

from pyspark import StorageLevel

# Example directory for local mode.
spark.sparkContext.setCheckpointDir(
    "file:///tmp/spark-scd2-checkpoints"
)

# Complete the checkpoint before modifying the target table.
delta_snapshot = staged_delta.checkpoint(eager=True)

# Release the original cache after checkpoint succeeds.
staged_delta.unpersist(blocking=True)

# Optional: cache the checkpoint-backed result for both consumers.
delta_snapshot = delta_snapshot.persist(StorageLevel.MEMORY_AND_DISK)
delta_snapshot.count()
delta_snapshot.createOrReplaceTempView("staged_delta")

然后创建原来的 close_source，再执行 MERGE 和 INSERT。两者都通过新的 staged_delta 视图读取 checkpoint 结果。

示例路径适用于具备写权限的本地模式；集群运行时应改为所有执行节点都可访问的可靠存储路径。固定上游结果依靠已经完成的 checkpoint，之后再次缓存是可选的性能优化。

方案二：将批次结果落盘后重新读取

如果中间结果还需要用于任务重启、审计或重试，可以在修改目标表前写入独立的批次目录：

# Replace with a unique path for this batch.
stage_path = "file:///tmp/scd2-stage/batch-unique-id"

staged_delta.write.mode("errorifexists").parquet(stage_path)

fixed_delta = spark.read.parquet(stage_path)
fixed_delta.createOrReplaceTempView("staged_delta")

写入完成后，再创建 close_source 并执行原来的两条写入语句。新读取的 DataFrame 依赖暂存文件，不再通过原 LEFT JOIN 读取目标表。两个操作全部成功前，应保留这批文件；路径需要按批次隔离。

SCD2 流程还要注意什么

固定中间结果既能减少重算，也能保持批次分类一致。如果 MERGE 关闭当前记录后，staged_delta 重新读取更新后的 current_target，原先标为 U 的记录可能变成 I。当前 INSERT 同时接受这两类，但批次分类已经发生变化，缓存不能保证批次快照固定。

此外，两次独立的 SQL 写入不会因为共享 checkpoint 而自动成为一个事务。MERGE 成功、INSERT 失败时，仍需设计恢复或幂等重试方式。

repartition(500) 本身也会引入 shuffle，不能保证缓存得到复用。先确认是否真正重扫源数据，再根据任务耗时和写入需求调整分区数；目前的 3.2GB 源数据大小与 12g 配置不足以判断中间结果是否能全部放入内存。

备注：内容仅供参考。