求助：Iceberg单分区写入多文件失败，如何兼顾性能？

问题背景 #

使用Iceberg写入数据时，单分区场景下代码正常执行，但多分区写入时触发报错；移除advertising_id后写入恢复正常，但单分区数据集中到单个执行器，性能严重下降。

执行的写入代码 #

df.repartition(
  partitions, col("exposure_id"), col("event_date"), col("advertising_id"))
.sortWithinPartitions(col("advertising_id"), col("timestamp"))
.writeTo(fullTableName)
.append()

触发的错误信息 #

Caused by: java.lang.IllegalStateException: Incoming records violate the writer assumption that records are clustered by spec and by partition within each spec. Either cluster the incoming records or switch to fanout writers.
Encountered records that belong to already closed files:
partition 'exposure_id=10/event_date=2024-06-28' in spec [
1000: exposure_id: identity(13)
1001: event_date: identity(14)
]

核心原因 #

Iceberg默认writer要求同一分区的记录是连续写入的，但把advertising_id加入repartition后，同一分区的数据会被分散到多个task中。当后续task再写入该分区时，之前的分区文件已经被关闭，就会触发上述报错。而移除advertising_id后，同一分区的数据会被分配到单个task，导致单执行器负载过高。

解决方案 #

1. 启用Fanout Writers（推荐） #

Fanout Writers允许单个task同时写入多个分区的文件，打破了默认writer对分区连续性的要求，完美适配当前的repartition逻辑。

可以通过会话配置全局启用：

spark.conf.set("spark.sql.iceberg.write.fanout.enabled", "true")

也可以在写入时单独指定选项：

df.repartition(
  partitions, col("exposure_id"), col("event_date"), col("advertising_id"))
.sortWithinPartitions(col("advertising_id"), col("timestamp"))
.writeTo(fullTableName)
.option("write.fanout.enabled", "true")
.append()

2. 调整排序逻辑保证分区连续性 #

如果不想启用Fanout，可以先按Iceberg的分区列（exposure_id、event_date）全局排序，确保同一分区的数据集中在连续的task中，避免writer遇到已关闭的分区文件：

df.repartition(
  partitions, col("exposure_id"), col("event_date"), col("advertising_id"))
.sort(col("exposure_id"), col("event_date"), col("advertising_id"), col("timestamp"))
.writeTo(fullTableName)
.append()

这里用全局sort替代sortWithinPartitions，先把同一分区的数据聚在一起，再做分区内排序。

3. 优化repartition参数 #

• 调整列顺序：把Iceberg分区列放在repartition的最前面，再加上advertising_id，让同一分区+同一advertising_id的数据落在同一个task，减少同一分区跨task的情况。
• 增大分区数：适当调高partitions参数值，让每个task处理的数据量更小，降低同一分区被拆分到多个task的概率。

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