Iceberg从入门到精通系列之二十四：Spark Structured Streaming

本文主要是介绍Iceberg从入门到精通系列之二十四：Spark Structured Streaming，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Iceberg从入门到精通系列之二十四：Spark Structured Streaming

一、Streaming Reads
二、Streaming Writes
三、Partitioned table
四、流表的维护

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”)。

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：将每个微批次的行追加到表中
complete：替换每个微批次的表内容

在开始流式查询之前，请确保您创建了表。请参阅 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")