Iceberg表Upsert时列缺失/新增无法合并问题求助

在AWS Glue Job v4（Spark 3.3.0-amzn-1 + Iceberg v1.0.0）环境下，创建Iceberg表时已设置write.spark.accept-any-schema="true"，表能正常创建且可在Glue/Lake Formation、Athena中查询，但执行Upsert的MERGE语句时，因源数据缺失目标表的部分列，抛出如下错误：

AnalysisException: cannot resolve my_column in MERGE command given columns [updates.col1, updates.col2, ...

创建表代码：

df.writeTo(f'glue_catalog.{DATABASE_NAME}.{TABLE_NAME}') \
    .using('iceberg') \
    .tableProperty("location", TABLE_LOCATION) \
    .tableProperty("write.spark.accept-any-schema", "true") \
    .tableProperty("format-version", "2") \
    .createOrReplace()

Upsert执行代码：

df_upsert.createOrReplaceTempView("upsert_items")
upsert_query = f"""
MERGE INTO glue_catalog.{DATABASE_NAME}.{TABLE_NAME} target
USING (SELECT * FROM upsert_items) updates
ON {join_condidtion}
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
"""
spark.sql(upsert_query)

原因说明 #

write.spark.accept-any-schema参数仅针对**批量写入（如append、overwrite）**生效，MERGE属于SQL DML操作，Iceberg不会自动为MERGE语句中缺失的列填充NULL，必须保证源表（updates）和目标表（target）的列完全匹配，或者显式指定列映射。

解决方法 #

方法1：显式补全缺失列 #

在USING子查询中，为目标表存在但源数据缺失的列显式赋值NULL，确保列数和列名完全匹配。
示例（静态补全）：
假设目标表有col1, col2, my_column，而源数据只有col1, col2，修改USING子查询：

MERGE INTO glue_catalog.{DATABASE_NAME}.{TABLE_NAME} target
USING (
    SELECT col1, col2, CAST(NULL AS STRING) AS my_column 
    FROM upsert_items
) updates
ON {join_condidtion}
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

若需动态适配目标表结构，可通过Spark元数据自动生成补全逻辑：

# 获取目标表的所有列及类型
target_df = spark.table(f'glue_catalog.{DATABASE_NAME}.{TABLE_NAME}')
target_cols = target_df.columns
source_cols = df_upsert.columns

# 生成缺失列的补全语句
missing_cols = [
    f"CAST(NULL AS {target_df.schema[c].dataType.simpleString()}) AS {c}" 
    for c in target_cols if c not in source_cols
]
select_clause = ", ".join(source_cols + missing_cols)

# 构建最终MERGE语句
upsert_query = f"""
MERGE INTO glue_catalog.{DATABASE_NAME}.{TABLE_NAME} target
USING (SELECT {select_clause} FROM upsert_items) updates
ON {join_condidtion}
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
"""
spark.sql(upsert_query)

方法2：显式指定列映射 #

不使用UPDATE SET *和INSERT *，而是明确列出需要更新/插入的列，对缺失列手动赋值：

MERGE INTO glue_catalog.{DATABASE_NAME}.{TABLE_NAME} target
USING (SELECT * FROM upsert_items) updates
ON {join_condidtion}
WHEN MATCHED THEN UPDATE SET 
    target.col1 = updates.col1,
    target.col2 = updates.col2,
    target.my_column = target.my_column -- 匹配时保留原列值，也可设为NULL
WHEN NOT MATCHED THEN INSERT (col1, col2, my_column)
VALUES (updates.col1, updates.col2, NULL)

方法3：升级Iceberg版本（可选） #

Iceberg v1.1.0及以上版本对MERGE的Schema兼容性有优化，可尝试升级Iceberg版本（需确认Glue Job支持对应版本），但需注意：即使升级，MERGE操作默认仍不会自动补全缺失列，仍需显式处理。

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