使用spark流式读写Iceberg

流式读取

Iceberg支持在Spark结构化流处理任务中读取增量数据，可以从指定的历史时间戳开始：

val df = spark.readStream
     .format("iceberg")
     .option("stream-from-timestamp", Long.toString(streamStartTimestamp))
     .load("database.table_name") 

Iceberg只支持从追加（append）快照读取数据。覆盖（overwrite）快照无法处理，默认会抛出异常。可以通过设置streaming-skip-overwrite-snapshots=true来忽略覆盖操作。同样，删除（delete）快照默认也会抛出异常，可以通过设置streaming-skip-delete-snapshots=true来忽略删除操作。

流式写入

要将流式查询的结果写入Iceberg表，请使用DataStreamWriter：

data.writeStream
     .format("iceberg")
     .outputMode("append")
     .trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
     .option("checkpointLocation", checkpointPath)
     .toTable("database.table_name") 

Iceberg支持append和complete两种输出模式：

• append：将每个微批次的行追加到表中
• complete：用每个微批次的内容替换整个表的内容

在启动流式查询之前，请确保已创建表。请参阅SQL创建表文档了解如何创建Iceberg表。

Iceberg不支持实验性的连续处理模式，因为该模式没有提供"提交"输出的接口。

分区表

Iceberg要求在写入数据之前按分区对每个任务的数据进行排序。在Spark中，任务按Spark分区进行划分。对于批处理查询，建议进行显式排序以满足要求，但这种方法会增加额外延迟，因为重新分区和排序对于流式工作负载来说是重量级操作。为了避免额外延迟，您可以启用fanout writer来消除这一要求。

data.writeStream
     .format("iceberg")
     .outputMode("append")
     .trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
     .option("fanout-enabled", "true")
     .option("checkpointLocation", checkpointPath)
     .toTable("database.table_name") 

Fanout writer为每个分区值打开文件，并在写入任务完成之前不会关闭这些文件。应避免在批处理写入中使用fanout writer，因为对输出行进行显式排序对于批处理工作负载来说成本较低。

流式表的维护

流式写入会快速创建新的表版本，产生大量表元数据来跟踪这些版本。通过调整提交频率、过期旧快照和自动清理元数据文件来维护元数据是非常推荐的。

调整提交频率

高频率的提交会产生数据文件、清单文件和快照，从而增加维护工作量。建议至少设置1分钟的触发间隔，并根据需要增加间隔。

过期旧快照

写入表的每个批次都会产生一个新的快照。Iceberg会跟踪表元数据中的快照，直到它们过期。频繁提交会导致快照快速累积，因此强烈建议对流式查询写入的表进行定期维护。快照过期是移除不再需要的元数据和任何数据文件的过程。默认情况下，该过程会过期超过五天的快照。

合并小文件

流式写入的数据量通常很小，这会导致表元数据跟踪大量小文件。将小文件合并成大文件可以减少表所需的元数据，并提高查询效率。Iceberg和Spark提供了rewrite_data_files存储过程。

重写清单文件

为了优化流式工作负载的写入延迟，Iceberg可以使用"快速追加"方式写入新快照，而不会自动合并清单文件。这可能会导致大量小清单文件。Iceberg可以重写清单文件数量以提高查询性能。Iceberg和Spark提供了rewrite_manifests存储过程。