Apache Iceberg Spark Merge未自动新增列，求动态Schema演化方案

一、Iceberg“完整Schema演化”的含义 #

Iceberg的完整Schema演化并非指自动感知并修改表结构，而是指它支持原子、无锁、不重写数据的Schema修改能力，涵盖以下操作：

• 新增/删除列
• 重命名列
• 修改兼容的列类型（比如int转bigint、string转varchar）
• 调整列顺序
• 添加/删除嵌套字段

这些修改仅在Iceberg的元数据层完成，不会触动现有数据文件，同时保证新旧Schema的双向兼容：旧数据读取新Schema时，新增列会自动填充默认值（如null）；新数据读取旧Schema时，会自动忽略新增列，完全不影响数据读写的连续性。

二、实现Merge操作时动态新增列的方法 #

你的场景中Merge未自动新增列，是因为Spark默认禁用了Merge操作的Schema演化，需要开启对应配置并调整语句写法：

1. 开启Merge Schema演化配置 #

在Spark作业中添加以下配置（二选一即可）：

• SparkSession初始化时设置：

  val spark = SparkSession.builder()
    .appName("IcebergMergeDemo")
    .config("spark.sql.iceberg.merge.schema.evolution.enabled", "true")
    // 若涉及时间类型无时区的情况，可额外添加：
    // .config("spark.sql.iceberg.handle-timestamp-without-timezone", "true")
    .getOrCreate()
  

• SQL语句中临时设置：

  SET spark.sql.iceberg.merge.schema.evolution.enabled = true;
  

2. 调整Merge语句写法 #

确保WHEN NOT MATCHED THEN INSERT明确指定插入所有列（写成INSERT *），让Iceberg能识别源表的新增列：

df.createOrReplaceTempView("myview")
spark.sql("""
MERGE INTO catalog.mydb.test2 as t 
USING (SELECT * FROM myview) as s 
ON t.id = s.id 
WHEN MATCHED THEN UPDATE SET * 
WHEN NOT MATCHED THEN INSERT *
""")

3. 验证结果 #

执行Merge后，通过以下SQL查看表结构，确认新增列已生效：

DESCRIBE TABLE catalog.mydb.test2;

注意事项 #

• 新增列默认会被添加到Schema末尾，若需指定位置，可通过Iceberg的API手动修改Schema（但一般无需额外操作）
• 源表与目标表的列类型必须兼容，否则Schema演化会失败（比如不能将int类型列改为string）
• 历史数据中的新增列会自动填充null（若列定义了默认值，则使用默认值）

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