Spark SQL+Iceberg：MERGE/INSERT未复用缓存DataFrame反而重扫源数据

架构老林

2026-6-1

问题

环境配置：Python 3.10、Spark 3.3.2，Spark驱动内存12g、本地模式local[6]，目标表为Iceberg表，源数据集3.2GB，源分区数12。

在优化Spark SQL的SCD2流程时，通过关联得到中间DataFrame staged_delta，以MEMORY_AND_DISK级别持久化，调用count()物化缓存后创建临时视图staged_delta。之后基于该视图执行两个操作：

MERGE操作关闭delta_type = 'U'的现有记录；
INSERT操作插入delta_type IN ('I','U')的新版本记录。

但运行时发现，MERGE和INSERT阶段并未复用缓存的staged_delta，反而重新扫描源数据，出现与源分区数一致的扫描阶段，推测是重新计算或回退到血缘依赖。

简化后的代码如下：

from pyspark import StorageLevel


def apply_scd2(spark, staged_df, target_table):
    # 1) Prepare source snapshot for this batch
    staged_df = staged_df.repartition(500)
    staged_df.createOrReplaceTempView("source_staged")

    # 2) Current target slice
    spark.sql(f"""
        CREATE OR REPLACE TEMP VIEW current_target AS
        SELECT key_a, key_b, period_key, change_hash
        FROM {target_table}
        WHERE is_current = true
    """)

    # 3) Build delta and cache it
    staged_delta = spark.sql("""
        SELECT
          s.*, 
          t.change_hash AS current_hash,
          CASE
            WHEN t.key_a IS NULL THEN 'I'
            WHEN t.change_hash <> s.change_hash THEN 'U'
            ELSE 'N'
          END AS delta_type
        FROM source_staged s
        LEFT JOIN current_target t
          ON t.key_a = s.key_a
         AND t.key_b = s.key_b
         AND t.period_key = s.period_key
    """).persist(StorageLevel.MEMORY_AND_DISK)

    # Materialize cache
    staged_delta.count()
    staged_delta.createOrReplaceTempView("staged_delta")

    # 4) Close rows (updates)
    spark.sql("""
        CREATE OR REPLACE TEMP VIEW close_source AS
      SELECT key_a, key_b, period_key, batch_ts, batch_id, delta_type
        FROM staged_delta
        WHERE delta_type = 'U'
    """)

    spark.sql(f"""
        MERGE INTO {target_table} t
        USING close_source 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 AND s.delta_type = 'U' THEN UPDATE SET
          t.valid_to = s.batch_ts,
          t.is_current = false,
          t.ingestion_ts = current_timestamp(),
          t.batch_id = s.batch_id
    """)

    # 5) Insert open rows (inserts)
    spark.sql(f"""
        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
          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')
    """)

预期行为：staged_delta仅缓存一次，MERGE和INSERT操作复用缓存，避免昂贵的源重扫与Shuffle。
实际行为：MERGE/INSERT阶段触发类似源扫描的操作，存在与源分区数（12）一致的阶段。

分析与解决方案

核心原因

临时视图与缓存DataFrame的解耦
通过staged_delta.createOrReplaceTempView()创建的临时视图是Spark SQL的逻辑视图，仅保存查询语句而非关联到缓存的DataFrame对象。Spark解析后续SQL时，会重新解析视图的血缘关系，回溯到source_staged和current_target，导致重新执行整个关联逻辑，而非读取缓存数据。
Iceberg操作的执行计划优化
Iceberg的MERGE/INSERT操作会触发Spark优化器重新规划执行计划，优化器可能认为直接扫描源数据并关联的计划更高效，或者未识别到缓存的依赖关系，从而绕过缓存。

解决办法

1. 直接基于缓存DataFrame生成下游数据，减少视图嵌套

避免通过SQL视图中转，直接对缓存的staged_delta进行过滤、选择操作生成下游DataFrame，再注册为临时视图：

# 替换原close_source的SQL创建逻辑
close_source = staged_delta.filter(staged_delta.delta_type == 'U').select(
    "key_a", "key_b", "period_key", "batch_ts", "batch_id", "delta_type"
)
close_source.createOrReplaceTempView("close_source")

# INSERT操作直接用缓存的DataFrame生成数据源
insert_df = staged_delta.filter(staged_delta.delta_type.isin('I', 'U')).select(
    "key_a", "key_b", "period_key", "attr_1", "attr_2",
    "batch_ts", 
    spark.sql("CAST(NULL AS TIMESTAMP)").alias("valid_to"),
    spark.sql("true").alias("is_current"),
    "change_hash",
    spark.sql("current_timestamp()").alias("ingestion_ts"),
    "batch_id"
)
insert_df.write.insertInto(target_table)

2. 使用SQL的`CACHE TABLE`语句直接缓存视图

改用SQL的CACHE TABLE来缓存staged_delta视图，让Spark SQL直接管理缓存，后续查询会优先读取缓存：

# 替换原DataFrame persist和count的逻辑
spark.sql("""
    CREATE OR REPLACE TEMP VIEW staged_delta AS
    SELECT
      s.*, 
      t.change_hash AS current_hash,
      CASE
        WHEN t.key_a IS NULL THEN 'I'
        WHEN t.change_hash <> s.change_hash THEN 'U'
        ELSE 'N'
      END AS delta_type
    FROM source_staged s
    LEFT JOIN current_target t
      ON t.key_a = s.key_a
     AND t.key_b = s.key_b
     AND t.period_key = s.period_key
""")
# 缓存视图并物化
spark.sql("CACHE TABLE staged_delta")
spark.sql("SELECT COUNT(*) FROM staged_delta")

3. 临时关闭自适应执行计划（AQE）

Spark 3.x的自适应执行计划可能会调整执行计划导致缓存失效，可在SCD2流程段临时关闭：

# 临时关闭AQE
original_aqe = spark.conf.get("spark.sql.adaptive.enabled")
spark.conf.set("spark.sql.adaptive.enabled", "false")

# 执行SCD2流程...

# 恢复原配置
spark.conf.set("spark.sql.adaptive.enabled", original_aqe)

4. 确认缓存状态

执行count()后，可通过以下方式确认缓存是否成功物化：

# 查看缓存的RDD
print(spark.sparkContext.getPersistentRDDs())
# 查看DataFrame的存储级别
print(staged_delta.storageLevel)

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

Spark SQL+Iceberg：MERGE/INSERT未复用缓存DataFrame反而重扫源数据

核心原因 #

解决办法 #

1. 直接基于缓存DataFrame生成下游数据，减少视图嵌套 #

2. 使用SQL的CACHE TABLE语句直接缓存视图 #

3. 临时关闭自适应执行计划（AQE） #