Iceberg 使用 Apache Spark 的 DataSourceV2 API 来实现数据源和目录。 Spark DSv2 是一个不断发展的 API，在 Spark 版本中提供不同级别的支持。
一、Streaming Reads
Iceberg 支持处理从历史时间戳开始的 Spark 结构化流作业中的增量数据：
val df = spark.readStream
.format("iceberg")
.option("stream-from-timestamp", Long.toString(streamStartTimestamp))
.load("database.table_name")
Iceberg 仅支持从追加快照中读取数据。覆盖快照无法处理，默认会引发异常。通过设置streaming-skip-overwrite-snapshots=true 可以忽略覆盖。同样，删除快照默认会引发异常，通过设置streaming-skip-delete-snapshots=true可以忽略删除。
二、Streaming Writes
要将流式查询中的值写入 Iceberg 表，请使用 DataStreamWriter：
data.writeStream
.format("iceberg")
.outputMode("append")
.trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
.option("checkpointLocation", checkpointPath)
.toTable("database.table_name")
如果您使用的是 Spark 3.0 或更早版本，则需要使用 .option(“path”, “database.table_name”).start()，而不是 .toTable(“database.table_name”)。
对于基于目录的 Hadoop 目录：
data.writeStream
.format("iceberg")
.outputMode("append")
.trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
.option("path", "hdfs://nn:8020/path/to/table")
.option("checkpointLocation", checkpointPath)
.start()
Iceberg 支持追加和完整输出模式：
- append：将每个微批次的行追加到表中
- ***plete：替换每个微批次的表内容
在开始流式查询之前，请确保您创建了表。请参阅 SQL 创建表文档以了解如何创建 Iceberg 表。
Iceberg 不支持实验性连续处理，因为它不提供“提交”输出的接口。
三、Partitioned table
Iceberg 需要在写入数据之前按每个任务的分区对数据进行排序。在 Spark 中，任务按 Spark 分区进行分割。针对分区表。对于批量查询，建议您进行显式排序来满足要求（请参阅此处），但该方法会带来额外的延迟，因为重新分区和排序被视为流工作负载的繁重操作。为了避免额外的延迟，您可以启用扇出编写器来消除这一要求。
data.writeStream
.format("iceberg")
.outputMode("append")
.trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))
.option("fanout-enabled", "true")
.option("checkpointLocation", checkpointPath)
.toTable("database.table_name")
扇出写入器打开每个分区值的文件，并且在写入任务完成之前不会关闭这些文件。避免使用扇出写入器进行批量写入，因为对输出行进行显式排序对于批量工作负载来说成本较低。
四、流表的维护
流式写入可以快速创建新的表版本，创建大量表元数据来跟踪这些版本。强烈建议通过调整提交率、使旧快照过期以及自动清理元数据文件来维护元数据。
调整提交率
高提交率会产生数据文件、清单和快照，从而导致额外的维护。建议触发间隔至少为 1 分钟，并根据需要增加间隔。
使旧快照过期
写入表的每个批次都会生成一个新快照。 Iceberg 跟踪表元数据中的快照，直到它们过期。快照会随着频繁提交而快速积累，因此强烈建议定期维护流式查询写入的表。快照过期是删除元数据和任何不再需要的数据文件的过程。默认情况下，该过程将使超过五天的快照过期。
压缩数据文件
从流处理写入的数据量通常很小，这可能会导致表元数据跟踪大量小文件。将小文件压缩为大文件可以减少表所需的元数据，并提高查询效率。 Iceberg 和 Spark 附带了 rewrite_data_files 过程。
重写清单
为了优化流工作负载的写入延迟，Iceberg 可以使用不会自动压缩清单的“快速”附加写入新快照。这可能会导致大量小的清单文件。 Iceberg可以重写清单文件的数量来提高查询性能。 Iceberg 和 Spark 附带了 rewrite_manifests 过程。
