Spark 3.3.0+Iceberg 1.1.0执行MERGE INTO报错求助

版本兼容性检查 #

Iceberg 1.1.0对Spark MERGE INTO的支持有明确限制：

• 仅支持基于Hive Catalog的Iceberg表，Dataproc元存储服务属于托管Hive元存储，理论上符合要求，但需确认集群是否正确绑定了Hive Catalog类型的Iceberg配置。
• 若使用其他Catalog类型（如自定义Catalog），Iceberg 1.1.0可能暂不支持MERGE INTO操作。

Spark配置校验 #

检查集群创建命令中的Iceberg相关配置是否齐全：

• 必须加载对应版本的Iceberg Spark运行时包：

  --properties spark.jars.packages=org.apache.iceberg:iceberg-spark-runtime-3.3_2.12:1.1.0
  

• 必须配置Iceberg扩展及正确的Catalog类型：

  --properties spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog,spark.sql.catalog.spark_catalog.type=hive
  

  注意：spark_catalog.type=hive是关键，确保元存储关联的是Hive类型Catalog。

MERGE INTO语法校验 #

Iceberg的MERGE INTO语法需严格遵循规范，避免语法错误：

• 必须包含明确的ON匹配条件，且条件需关联Iceberg表的主键（若表定义了主键）。
• 正确语法示例：

  MERGE INTO your_db.your_iceberg_table t
  USING (SELECT * FROM parquet.`gs://your-bucket/parquet-path`) s
  ON t.id = s.id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
  

• 若通过临时视图引用Parquet数据，需确保视图正确注册：

  spark.read.parquet("gs://your-bucket/parquet-path").createOrReplaceTempView("parquet_source")
  

临时替代方案 #

若以上排查后仍无法解决，可使用Iceberg提供的merge API实现UPSERT逻辑，Python代码示例：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 读取Parquet源数据
source_df = spark.read.parquet("gs://your-bucket/parquet-path")
# 加载目标Iceberg表
target_table = spark.table("your_db.your_iceberg_table")

# 执行Upsert操作
target_table.alias("t").merge(
    source_df.alias("s"),
    "t.id = s.id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

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