# 使用spark流式读写Iceberg
Iceberg使用Apache Spark的DataSourceV2 API来实现数据源和目录（catalog）功能。Spark DSv2是一个不断演进的API，在不同版本的Spark中支持程度有所不同。
#### 流式读取
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存储过程。