Apache Spark™3.1针对结构化流的新特性

通过任命李，Shixiong朱而且张老板

2021年4月27日在工程的博客

分享这篇文章

除了提供基于Spark Core和SQL API的流处理能力外，结构化流是Apache Spark™最重要的组件之一。在这篇博文中，我们总结了最新3.1版本中Spark Streaming的显著改进，包括一个新的流表API，支持流-流连接和多个UI增强。此外，模式验证和对Apache Kafka数据源的改进提供了更好的可用性。最后，对FileStream源/接收器进行了各种增强，以提高读/写性能。

新的流表API

在启动结构化流时，连续数据流被认为是无界表。因此，Table api提供了一种更自然、更方便的方式来处理流查询。在Spark 3.1中，我们增加了对DataStreamReader和DataStreamWriter的支持。最终用户现在可以直接使用API来读取和写入流数据帧作为表。请看下面的例子:

＃创建流数据帧src＝spark.readStream.format(“率”).option(“rowPerSecond”,10) .load ()#写入流数据帧来一个表格src.writeStream。选项(“checkpointLocation checkpointLoc1) .toTable(“myTable”)＃检查的表格结果spark.read.table(“myTable”)。显示（截断＝30.）+-----------------------+-----+|时间戳|价值|+-----------------------+-----+|2021-01年-19年07：45：23.122|42||2021-01年-19年07：45：23.222|43||2021-01年-19年07：45：23.322|44|.．.

此外，使用这些新选项，用户可以转换源数据集并写入一个新表:

#写来一个新表格与转换spark.readStream.table(“myTable”)。选择(“价值”)\.writeStream。选项("checkpointLocation"， checkpointLoc2) \.format(“铺”).toTable(“newTable”)＃检查的表格结果spark.read.table(“newTable”)。显示（）+-----+|价值|+-----+|1214||1215||1216|.．.

Databricks建议使用三角洲湖格式与流表api，这允许您

紧凑的小文件产生的低延迟摄取并发。
使用多个流(或并发批处理作业)维护“恰好一次”的处理。
当使用文件作为流的源时，有效地发现哪些文件是新文件。

新支持流-流连接

在Spark 3.1之前，流-流连接只支持内连接、左外连接和右外连接。在最新的版本中，我们实现了完全的外部和左侧半流连接，使结构化流在更多的场景中有用。

左半流-流连接(火星- 32862）
完整的外部流-流连接(火星- 32863）

Kafka数据源的改进

在Spark 3.1中，我们已经将Kafka依赖升级到2.6.0 (火星- 32568)，允许用户迁移到新的Kafka偏移量检索API (adminclient . listoffset)。它解决了问题(火星- 28367)的Kafka连接器在使用旧版本时无限等待。

模式验证

模式是结构化流查询的基本信息。在Spark 3.1中，我们为用户输入模式和内部状态存储添加了模式验证逻辑:

在查询重启之间引入状态模式验证(火星- 27237）

通过此更新，键和值模式将存储在流开始时的模式文件中。然后，在重新启动查询时，将根据现有的键和值模式验证新的键和值模式的兼容性。当字段数量相同且每个字段的数据类型相同时，状态模式被认为是“兼容的”。注意，这里不检查字段名，因为Spark允许重命名。

这将防止运行具有不兼容状态模式的查询，从而减少不确定行为的机会，并提供更多信息的错误消息。

为流状态存储引入模式验证(火星- 31894）

以前，结构化流直接将检查点(用UnsafeRow表示)放入StateStore，而不需要任何模式验证。当升级到新的Spark版本时，检查点文件将被重用。如果没有模式验证，与聚合函数相关的任何更改或错误修复都可能导致随机异常，甚至是错误的答案(例如火星- 28067)．现在Spark根据模式验证检查点，并在迁移期间重用检查点时抛出InvalidUnsafeRowException。值得一提的是，这项工作还帮助我们找到了阻断剂，火星- 31990: Spark 3.0.1版本中，流的状态存储兼容性被打破。