Apache Iceberg Merge into冲突文件问题的合规修复方案咨询

错误信息 #

org.apache.iceberg.exceptions.ValidationException: Found conflicting files that can contain records matching true: [file_1, file_2, file_3]

问题背景 #

• 已知直接删除底层文件会破坏Iceberg元数据，需安全修复方式
• 问题跨会话持续，排除目录缓存导致的临时异常
• 触发场景：通过AWS Athena执行update table1 SET column1 = NULL清空列后，在AWS Glue Spark作业中运行Merge into语句填充列时触发该错误；怀疑Athena的Update操作未正确标记文件为过时，回滚至更早快照无法解决问题

修复步骤 #

1. 核查快照与文件状态 #

先确认冲突文件的归属快照及是否被标记为删除：

• 通过Athena查询元数据：

SELECT snapshot_id, file_path, is_deleted 
FROM "your_database"."table1$files" 
WHERE file_path IN ('file_1', 'file_2', 'file_3')

• 通过Glue Spark作业查询：

val table = spark.table("your_database.table1")
  .asInstanceOf[org.apache.spark.sql.catalyst.TableIdentifier]
  .getTable.asInstanceOf[org.apache.iceberg.spark.SparkTable].table()

// 打印所有快照信息
table.snapshots().forEach(s => println(s"快照ID: ${s.snapshotId()}, 生成时间: ${s.timestamp()}"))

// 打印冲突文件的状态
table.files().forEach(f => 
  if (Seq("file_1", "file_2", "file_3").contains(f.path().toString)) {
    println(s"文件路径: ${f.path()}, 所属快照: ${f.snapshotId()}, 是否已删除: ${f.deleted()}")
  }
)

2. 安全标记冲突文件为已删除 #

若确认冲突文件属于已完成的更新操作且应被标记为过时，通过Iceberg API手动标记删除：

import org.apache.iceberg.operations.DeleteFile

val table = spark.table("your_database.table1")
  .asInstanceOf[org.apache.spark.sql.catalyst.TableIdentifier]
  .getTable.asInstanceOf[org.apache.iceberg.spark.SparkTable].table()

// 筛选出冲突的DataFile
val conflictingDataFiles = table.files()
  .filter(f => Seq("file_1", "file_2", "file_3").contains(f.path().toString))
  .map(_.asInstanceOf[org.apache.iceberg.DataFile])
  .toList

// 构建删除操作并提交
val deleteOperations = conflictingDataFiles.map(f => 
  DeleteFile.builder(table.spec()).fromDataFile(f).build()
)

table.newRewrite().deleteFiles(deleteOperations).commit()

3. 清理无效快照与优化表 #

若回滚快照无效，说明快照链存在异常，执行快照清理与表优化：

val table = spark.table("your_database.table1")
  .asInstanceOf[org.apache.spark.sql.catalyst.TableIdentifier]
  .getTable.asInstanceOf[org.apache.iceberg.spark.SparkTable].table()

// 清理7天前的快照，保留最近3个有效快照
table.expireSnapshots()
  .expireOlderThan(System.currentTimeMillis() - 7*24*3600*1000)
  .retainLast(3)
  .commit()

// 执行表优化，整理文件布局
spark.sql("OPTIMIZE your_database.table1 REWRITE DATA USING BIN_PACK")

4. 后续规避方案 #

由于Athena对Iceberg的Update操作可能存在元数据同步延迟或异常，后续改用Glue Spark作业执行列清空操作：

// 在Glue作业中执行Update
spark.sql("UPDATE your_database.table1 SET column1 = NULL")

// 执行后立即清理过期快照
val table = spark.table("your_database.table1")
  .asInstanceOf[org.apache.spark.sql.catalyst.TableIdentifier]
  .getTable.asInstanceOf[org.apache.iceberg.spark.SparkTable].table()

table.expireSnapshots()
  .expireOlderThan(System.currentTimeMillis() - 3600*1000) // 清理1小时前的快照
  .commit()

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