Spark SQL+Iceberg技术问题：MERGE与INSERT操作未复用缓存DataFrame而是重新扫描源数据

我之前在做Iceberg SCD2开发时也碰到过一模一样的问题，咱们来拆解下根源，再给出针对性的解决办法：

为什么缓存没被复用？ #

1. TempView的执行链路被优化器展开了 #

你把缓存好的staged_delta注册成TempView后，后续的MERGE/INSERT SQL在解析阶段，Spark Catalyst优化器会自动把TempView的原始定义（也就是那个LEFT JOIN的SQL逻辑）直接展开到最终执行计划里，而不是引用已经物化好的缓存DataFrame。这就相当于跳过了缓存节点，重新走一遍源数据扫描+Join的流程。

2. Iceberg MERGE的特殊执行逻辑绕开了缓存 #

Iceberg的MERGE操作有专属的执行路径（比如处理快照、分区更新、元数据校验等），Spark的通用缓存机制和这套路径的整合性不好，优化器会优先选择Iceberg专属的优化策略，自然就忽略了已缓存的中间结果。

3. 缓存可能被隐性清理或优化器认为重算更高效 #

虽然你用count()触发了缓存物化，但如果缓存的DataFrame分区数、数据分布和后续MERGE/INSERT的需求不匹配，或者优化器发现可以把谓词下推到源表减少扫描量，它会直接跳过缓存，认为重新计算的成本更低。

具体解决方案 #

方案1：改用DataFrame API执行MERGE和INSERT #

放弃Spark SQL语句，直接用DataFrame的原生API操作，这样能强制Spark复用缓存的staged_delta：

from pyspark.sql.functions import current_timestamp, col

# 4) 过滤出需要关闭的记录（直接用缓存的DataFrame）
close_source = staged_delta.filter(col("delta_type") == "U")

# 执行MERGE更新
target_df = spark.read.table(target_table)
(target_df
 .merge(close_source, 
        (target_df.key_a == close_source.key_a) &
        (target_df.key_b == close_source.key_b) &
        (target_df.period_key == close_source.period_key) &
        (target_df.is_current == True))
 .whenMatchedUpdate(set={
     "valid_to": close_source.batch_ts,
     "is_current": False,
     "ingestion_ts": current_timestamp(),
     "batch_id": close_source.batch_id
 })
 .execute())

# 5) 过滤出需要插入的记录并写入
insert_df = staged_delta.filter(col("delta_type").isin("I", "U"))
(insert_df
 .select(
     col("key_a"), col("key_b"), col("period_key"), col("attr_1"), col("attr_2"),
     col("batch_ts").alias("valid_from"),
     None.cast("timestamp").alias("valid_to"),
     col("true").alias("is_current"),
     col("change_hash"),
     current_timestamp().alias("ingestion_ts"),
     col("batch_id")
 )
 .write
 .mode("append")
 .saveAsTable(target_table))

方案2：用SQL提示强制禁用视图重写 #

如果想保留Spark SQL的写法，可以在查询staged_delta的语句中加入/*+ NO_REWRITE */提示，阻止优化器展开视图的执行链路：

-- 生成关闭记录的视图时加提示
CREATE OR REPLACE TEMP VIEW close_source AS
SELECT /*+ NO_REWRITE */ key_a, key_b, period_key, batch_ts, batch_id, delta_type
FROM staged_delta
WHERE delta_type = 'U'

-- 插入数据时同样加提示
INSERT INTO {target_table} (
    key_a, key_b, period_key, attr_1, attr_2,
    valid_from, valid_to, is_current, change_hash,
    ingestion_ts, batch_id
)
SELECT /*+ NO_REWRITE */
    s.key_a, s.key_b, s.period_key, s.attr_1, s.attr_2,
    s.batch_ts, CAST(NULL AS TIMESTAMP), true, s.change_hash,
    current_timestamp(), s.batch_id
FROM staged_delta s
WHERE s.delta_type IN ('I', 'U')

方案3：将缓存DataFrame写入临时物理表 #

把缓存好的staged_delta写入临时物理表，后续操作直接查询这个表，彻底切断和源数据的链路：

# 将缓存数据写入临时表（支持内存+磁盘存储）
staged_delta.write.mode("overwrite").saveAsTable("staged_delta_temp")

# MERGE时直接查询临时表
spark.sql(f"""
    MERGE INTO {target_table} t
    USING (SELECT * FROM staged_delta_temp WHERE delta_type = 'U') s
    ON t.key_a = s.key_a
    AND t.key_b = s.key_b
    AND t.period_key = s.period_key
    AND t.is_current = true
    WHEN MATCHED THEN
        UPDATE SET
            t.valid_to = s.batch_ts,
            t.is_current = false,
            t.ingestion_ts = current_timestamp(),
            t.batch_id = s.batch_id
""")

# 插入操作同理
spark.sql(f"""
    INSERT INTO {target_table} (...)
    SELECT * FROM staged_delta_temp WHERE delta_type IN ('I', 'U')
""")

额外验证步骤 #

在执行MERGE/INSERT前先确认缓存状态，避免缓存未生效的情况：

# 检查缓存是否激活
print(f"staged_delta 已缓存: {spark.catalog.isCached('staged_delta')}")
# 查看缓存的详细信息（分区数、存储大小等）
spark.catalog.cacheStatus().show(truncate=False)

小建议 #

• 你的repartition(500)可以根据数据量调整，建议每个分区控制在128MB-256MB之间，避免过多小分区增加Shuffle开销；
• 打开Spark UI（默认http://localhost:4040）查看执行计划，对比缓存前后的Stage变化，能直观确认缓存是否被复用。

内容的提问来源于stack exchange，提问作者fabrik5k