Apache Iceberg多Spark流写入同表执行Merge时冲突报错求助

多个Spark Streaming任务向同一张Iceberg表的不同字段写入数据，Iceberg文档说明支持基于乐观并发的多并发写入，但执行Merge操作时出现如下报错：

Caused by: org.apache.iceberg.exceptions.ValidationException: Found conflicting files that can contain records matching true

Spark Merge语句 #

spark.sql(
  f"""
  MERGE INTO datahub.replicacao.pefin_table tgt
  USING (select nu_documento, co_cadus, aud_enttyp, nu_particao from pefin_pf) src
  ON tgt.nu_documento = src.nu_documento and src.nu_particao in ('1', '2', '4')
  WHEN MATCHED AND src.aud_enttyp = 'D' THEN DELETE
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
""")

Spark Session配置 #

val spark = SparkSession.builder()
.master("local[*]")
.config("spark.sql.catalog.datahub", "org.apache.iceberg.spark.SparkSessionCatalog")
.config("spark.sql.catalog.datahub.type", "hadoop")
.config("spark.sql.catalog.datahub", "org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.datahub.warehouse", "file:///C:/dev/warehouse")
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.getOrCreate()

原因分析 #

• 乐观并发冲突：Iceberg的乐观并发控制依赖快照版本对比，当多个Merge任务同时操作同一批数据行（通过nu_documento匹配）时，第一个任务提交生成新快照后，第二个任务基于旧快照执行Merge，会检测到数据文件冲突，触发报错。
• Merge条件范围重叠：当前Merge的ON条件包含src.nu_particao in ('1', '2', '4')，多个Streaming任务可能处理不同分区的相同nu_documento记录，导致不同任务的Merge命中同一目标行，引发冲突。
• Catalog配置错误：Spark Session配置中重复设置spark.sql.catalog.datahub，先设为SparkSessionCatalog又覆盖为SparkCatalog，可能导致Catalog初始化异常，影响并发控制逻辑的正确性。

解决方法 #

1. 修复Catalog配置：移除重复的Catalog配置项，统一使用正确的Catalog类型，比如：

val spark = SparkSession.builder()
.master("local[*]")
.config("spark.sql.catalog.datahub", "org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.datahub.type", "hadoop")
.config("spark.sql.catalog.datahub.warehouse", "file:///C:/dev/warehouse")
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.getOrCreate()

2. 缩小Merge冲突范围：
   • 给每个Streaming任务分配独立的分区范围，比如任务1处理nu_particao='1'，任务2处理nu_particao='2'，修改对应任务的Merge ON条件为专属分区，确保不同任务不会操作同一行数据。
   • 或者在Merge的ON条件中增加任务专属标识字段，避免跨任务的行冲突。
3. 添加冲突重试逻辑：在Spark Streaming任务中捕获乐观并发冲突异常，配置自动重试机制，利用Iceberg的快照版本重试Merge操作，直到执行成功。
4. 优化Merge粒度：避免全表Merge，基于分区做增量Merge，减少不同任务操作重叠数据的概率。

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