Spark3.2.1StructuredStreaming编程指南

Structured Streaming是一个可扩展、容错的流处理引擎，建立在Spark SQL引擎之上。开发者可以用离线批处理数据相同的表示来表示流计算的逻辑，并且保持其逻辑的一致性(流批一体)。Spark SQL引擎会处理好增量连续运行，并随着流式数据的接收持续更新最终结果。开发者可以使用Dataset/DataFrame API ，使用Scala，Java，Python或者R的方式编程，表达 streaming 聚合，事件时间窗口，流批Join等。计算逻辑在Spark SQL引擎上执行，充分利用Spark SQL引擎的优势。最后，系统通过Checkpoint及Write-Ahead Log保证端到端的exactly-once容错机制。简单来说，Structured Streaming提供高性能，可扩展，容错，端到端exactly-once的流处理

，开发人员无需对流进行额外的考虑。

内部实现上，Structured Streaming 的query默认采用micro-batch处理引擎，将数据流当作一系列小的批任务处理，端到端延时最低能到100毫秒并且保证exactly=once容错机制。在Spark2.3中，加入了一个名为Continuous的新的低延时处理模式，可以达到端到端延时1ms并且保证at-least-once。在两种模式间切换时无需修改Dataset/DataFrame的处理逻辑。

在本指南中，会详细介绍变成模型和API。大部分概念会使用默认的micro-batch处理模式来进行解释，随后再讨论Continuous处理模式。接下来，我们先从一个简单的Structured Streaming query--流式word count开始。

假设你需要维护一个持续从TCP socket接收文本数据并word count的程序。看看如何使用Structured Streaming来表达。下面是Scala的例子，如果下载Spark，还可以直接运行该示例。让我们通过示例一步步的理解其工作原理。

首先，我们需要import必须的类并创建一个local模式的SparkSession，这是所有与Spark相关的功能的起点。

import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
val spark = SparkSession
.builder
.appName("StructuredNetworkWordCount")
.getOrCreate()

import spark.implicits._

接下来，创建一个Streaming DataFrame，它表示从本地主机9999端口接收的文本数据，并转换DataFrame来计算wordcount。

// Create DataFrame representing the stream of input lines from connection to localhost:9999
val lines = spark.readStream
.format("socket")
.option("host", "localhost")
.option("port", 9999)
.load()
// Split the lines into words
val words = lines.as[String].flatMap(_.split(" "))
// Generate running word count
val wordCounts = words.groupBy("value").count()

lines DataFrame表示包含流式文本数据的无界表。此表包含一列名为“value”的字符串，流式文本数据中的每一行都成为表中的一行。请注意，由于我们只是在设置转换，还没有开始转换，所以目前还没有收到任何数据。接下来，我们使用将DataFrame转换为字符串数据集。作为[String]，这样我们就可以应用flatMap操作将每一行拆分为多个单词。结果单词数据集包含所有单词。最后，我们通过按数据集中的唯一值分组并计数来定义wordCounts DataFrame。请注意，这是一个流DataFrame，表示流的wordcount。

我们现在已经设置了对流数据的query。剩下的就是开始接收数据并计算计数。为了做到这一点，我们将其设置为在每次更新时将完整的计数集（由outputMode（“complete”）指定）打印到控制台。然后使用start()启动流计算。

// Start running the query that prints the running counts to the console
val query = wordCounts.writeStream
.outputMode("complete")
.format("console")
.start()
query.awaitTermination()

执行此代码后，流计算将在后台启动。query对象是活动流式查询的句柄，我们使用waitTermination（）等待查询的终止，以防止进程在查询活动时退出。

要实际执行此示例代码，您可以在自己的Spark应用程序中编译代码，或者在下载Spark后运行示例。我们正在展示后者。首先，您需要使用

$ nc -lk 9999

然后，在另一个终端中，可以使用

$ ./bin/run-example org.apache.spark.examples.sql.streaming.StructuredNetworkWordCount localhost 9999

然后，在运行netcat服务器的终端中键入的任何行都将每秒计数并打印在屏幕上。如下所示：

# TERMINAL 1:
# Running Netcat
$ nc -lk 9999
apache spark
apache hadoop

# TERMINAL 2: RUNNING StructuredNetworkWordCount
$ ./bin/run-example org.apache.spark.examples.sql.streaming.StructuredNetworkWordCount localhost 9999
-------------------------------------------
Batch: 0
-------------------------------------------
+------+-----+
| value|count|
+------+-----+
|apache| 1|
| spark| 1|
+------+-----+
-------------------------------------------
Batch: 1
-------------------------------------------
+------+-----+
| value|count|
+------+-----+
|apache| 2|
| spark| 1|
|hadoop| 1|
+------+-----+
...

Structured Streaming 的关键思想是将实时数据流视为一个不断追加的表。这种思想创造了一种新的流处理模型，它与批处理模型非常相似。您将把流计算表示为静态表上的标准批处理查询，Spark将其作为无界输入表上的增量查询运行。让我们更详细地了解这个模型。

3.1 基本概念

将输入数据流视为“输入表”。流中的每个数据项都像是一个新行被追加到输入表中。

对输入的查询将生成“结果表”。每个触发间隔（比如，每1秒）都会向输入表追加新行，最终更新结果表。无论何时更新结果表，我们都希望将更改后的结果行写入外部接收器。

“Output”定义为写入外部存储器的内容。可以在不同的模式下定义输出：

Complete Mode-整个更新的结果表将写入外部存储器。由存储连接器决定如何处理整个表的写入。

Append Mode-只有自上次触发器以来追加到结果表中的新行才会写入外部存储器。这仅适用于结果表中的现有行预计不会更改的查询。

Update Mode-只有自上次触发后在结果表中更新的行才会写入外部存储器（自Spark 2.1.1起可用）。请注意，这与Complete Mode的不同之处在于，此模式仅输出自上次触发以来已更改的行。如果查询不包含聚合，则相当于追加模式。

请注意，每种模式都适用于某些类型的查询。这将在后面详细讨论。

ote that the query on streaming lines DataFrame to generate wordCounts is exactly the same as it would be a static DataFrame.

为了说明该模型的使用，让我们结合上面的快速示例来理解该模型。第一行DataFrame是输入表，最后一行wordCounts DataFrame是结果表。请注意，streaming DataFrame上lines到wordCounts的query与static DataFrame完全相同。但是，当该查询启动时，Spark将持续检查来自socket连接的新数据。如果有新数据，Spark将运行“增量”查询，将以前运行的计数与新数据结合起来，以计算更新的计数，如下所示。

需要注意的是，Structured Streaming并不能物化整个表。它从流数据源读取最新的可用数据，增量处理以更新结果，然后丢弃源数据。它只保留更新结果所需的最小中间状态数据（如前面示例中的中间计数）。

此模型与许多其他流处理引擎有显著不同。许多流媒体系统要求用户自己维护正在运行的聚合，因此必须考虑容错性和数据一致性（至少一次、最多一次或恰好一次）。在该模型中，Spark负责在有新数据时更新结果表，从而避免用户对其进行推理。作为一个例子，让我们看看这个模型如何处理基于事件时间的处理和延迟到达的数据。

3.2 处理事件时间和延迟数据

事件时间是数据本身产生的时间，包含在数据内容中。对于许多应用程序，您可能希望在事件时间上进行操作。例如，如果希望获得每分钟由物联网设备生成的事件数，那么最好是能使用生成数据的时间（即数据中的事件时间），而不是Spark接收数据的时间。这个事件时间在这个模型中非常自然地表示出来——来自设备的每个事件都是表中的一行，而事件时间是行中的一列值。这允许基于窗口的聚合（例如，每分钟的事件数）只是事件时间列上的一种特殊类型的分组和聚合——每个时间窗口是一个组，每一行可以属于多个窗口/组。因此，可以在离线数据（例如已经收集好的设备事件日志）以及实时数据流上一致地定义这种基于事件时间窗口的聚合查询，这种流批一体的处理方式能够帮助开发人员减轻大量的重复开发工作。

此外，该模型天然的支持根据事件时间处理比预期晚到的数据。由于Spark自己更新结果表，因此它可以完全控制在有延迟数据时更新旧聚合，以及清理旧聚合以限制中间状态数据的大小。从Spark 2.1开始，我们就支持Watermark，它允许用户指定延迟数据的阈值，并允许引擎相应地清除旧状态。稍后将在“窗口操作”一节中详细解释这些操作。

3.3 容错语义

提供端到端的exactly-once语义是Structured Streaming设计背后的关键目标之一。为了实现这一点，我们设计了Structured Streaming sources, sinks 和execution engine，以可靠地跟踪处理的确切进度，从而可以通过重新启动和/或重新处理来处理任何类型的故障。假设每个流媒体源都有偏移量（类似于Kafka偏移量或Kinesis序列号）来跟踪流中的读取位置。引擎使用检查点和WAL来记录每个Trigger中正在处理的数据的偏移范围。Sinks设计为幂等重复处理。总之，使用可重放Source和幂等Sink，Structured Streaming 可以确保在任何故障情况下端到端只执行一次语义。

4.1 创建Streaming DataFrame和Streaming Dataset

Streaming DataFrame可以通过SparkSession.readStream()返回的DataStreamReader接口创建。与创建静态DataFrame的读取接口类似，可以指定source的详细信息——data format, schema, options等。

4.1.1 Input Source

内置的Input Source如下：

File source-读取写入到目录中的文件作为数据流。文件会按照文件修改时间的顺序进行处理。如果设置了latestFirst，则顺序将颠倒。支持的文件格式有text、CSV、JSON、ORC和Parquet。请参阅DataStreamReader的文档，以获取最新的列表，以及每种文件格式支持的option。请注意，文件必须以原子方式放置在给定的目录中，在大多数文件系统中，可以通过文件move操作来实现。

Kafka source-从卡夫卡读取数据。它与Kafka broker版本0.10.0或更高版本兼容。有关更多详细信息，请参阅《Kafka 集成指南》。

Socket source（用于测试）-从socket连接读取UTF8文本数据。socket接收器位于Driver端。请注意，这只应用于测试，因为它不提供端到端的容错保证。

Rate source（用于测试）-以每秒指定的行数生成数据，每个输出行包含时间戳和值。其中，timestamp是包含消息分派时间的时间戳类型，value是包含消息计数的Long类型，从0开始作为第一行。此源用于测试和基准测试。

有些数据源是不容错的，因为它们不能保证在发生故障后可以使用checkpoint偏移量重放数据。请参阅前面关于容错语义的部分。以下是Spark中所有Source的详细信息。

File source（容错）参数列表：

path：输入目录的路径，对所有文件格式通用。

maxFilesPerTrigger：每个Trigger中包含的最大新文件数（默认值：无最大值）

latestFirst：是否首先处理最新的新文件，在有大量积压文件时很有用（默认值：false）

filenameOnly：是否仅基于文件名而非完整路径检查新文件（默认值：false）。设置为“true”时，以下文件将被视为同一文件，因为它们的文件名“dataset.txt”是相同的：

"file:///dataset.txt"

“s3://a/dataset.txt”

“s3n://a/b/dataset.txt”

“s3a://a/b/c/dataset.txt”

maxFileAge：目录中文件的最长期限,超过该期限将被忽略。对于第一批，所有文件都将被视为有效。如果latestFirst设置为'true'，并设置了maxFilesPerTrigger，则此参数将被忽略，因为可能会忽略有效且应处理的旧文件。最大期限是根据最新文件的时间戳而不是当前系统的时间戳来指定的。（默认值：1周）

cleanSource：处理后清理已完成文件的选项。

可用选项有"archive", "delete", "off"。如果未提供该选项，则默认值为“off”。

当提供“archive”时，还必须提供附加选项sourceArchiveDir。“sourceArchiveDir”的值在深度上不能与source的通配符（根目录中的目录数）匹配，其中深度是两个path上的最小深度。这将确保存档文件永远不会作为新的源文件。

例如，假设使用/hello?/spark/*作为 source 通配符，“/hello1/spark/archive/dir”不能用作“sourceArchiveDir”的值，如/hello?/spark/*和'/hello1/spark/archive'将匹配。/archived/here可以，因为它不匹配。

Spark将根据源文件自身的路径移动源文件。例如，如果源文件的路径是/a/b/dataset.txt，存档目录的路径为/archive/here，文件将被移动到/archive/here/a/b/dataset.txt。

注1：存档（通过移动）或删除已完成的文件都会在每个微批处理中引入开销（降低速度，即使它发生在单独的线程中），因此在启用此选项之前，您需要了解文件系统中每个操作的成本。另一方面，启用此选项的好处是将降低list源文件这一高消耗操作的成本。

可以使用spark配置已完成文件清理器中使用的线程数。具体配置是spark.sql.streaming.fileSource.cleaner.numThreads（默认值：1）。

注2：启用此选项时，确保该source仅被一处输入使用，且未被其他输出使用。

注3：删除和移动操作都是尽可能的操作。未能删除或移动文件不会导致流式查询失败。Spark在某些情况下可能无法清理某些源文件，例如，应用程序没有正常关闭，有太多文件排队清理。

有关特定于文件格式的选项，请参阅DataStreamReader中的相关方法。例如，有关“parquet”格式的选项，请参阅DataStreamReader.parquet（）。

此外，还有一些session配置会影响某些文件格式。有关更多详细信息，请参阅《SQL编程指南》。例如，“parquet”见parquet配置。

Socket Source(非容错)：

支持的配置有host和port，均为必填。

Rate Source（容错）：

rowsPerSecond（例如100，默认值：1）：每秒应该生成多少行。

rampUpTime（例如5s，默认值：0s）：在生成速度变为rowsPerSecond之前，需要多长时间才能加速。使用小数点将被截断为整数秒。

numPartitions（例如10，默认值：Spark的default parallelism）：生成的行的分区号。

source将尽最大努力达到rowsPerSecond，但可能会受到资源限制，可以调整numPartitions以帮助达到所需的速度。

Kafka Source（容错）：

参考《Kafka Integration Guide》

其他Source：

Pulsar：参考https://github.com/streamnative/pulsar-spark

示例：

val spark: SparkSession = ...
// Read text from socket
val socketDF = spark
.readStream
.format("socket")
.option("host", "localhost")
.option("port", 9999)
.load()
socketDF.isStreaming // Returns True for DataFrames that have streaming sources

socketDF.printSchema
// Read all the csv files written atomically in a directory
val userSchema = new StructType().add("name", "string").add("age", "integer")
val csvDF = spark
.readStream
.option("sep", ";")
.schema(userSchema) // Specify schema of the csv files
.csv("/path/to/directory") // Equivalent to format("csv").load("/path/to/directory")

该示例生成untyped的Streaming DataFrame，这意味着DataFrame的Schema不会在编译时被检查，只有在提交查询时才会在运行时被检查。有些操作，如map、flatMap等，需要在编译时就知道类型。为此，可以使用与Static DataFrame相同的方法将这些untyped Streaming DataFrame转换为typed Streaming DataFrame。有关更多详细信息，请参阅《SQL编程指南》。此外，本文后面将讨论有关受支持的Streaming Source的更多详细信息。

从Spark 3.1开始，还可以使用DataStreamReader.table()从表中创建Streaming DataFrame。

4.1.2 Schema推理和Partition

默认情况下，在Structured Streaming中，File Source的需要手动指定Schema，而不是依靠Spark自动推断。此限制确保流式处理使用一致的Schema，即使在失败的情况下也是如此。对于Ad-hoc的场景，可以通过设置spark.sql.streaming.schemaInference=true来启用自动推断。

如果存在名为/key=value/的子目录则会触发分区发现，并且list操作将自动递归到这些目录中，。如果这些列出现在程序指定的shema中，Spark将根据正在读取的文件的路径来填充它们。当处理开始时，组成分区Schema的目录必须存在，并且必须保持静态。例如，当/data/year=2015/存在时可以添加/data/year=2016/，但更改分区列（即创建目录/data/date=2016-04-17/）是无效的。

4.2 操作 Streaming DataFrame/Dataset

在Streaming DataFrame/Dataset上可以进行各种操作，从untyped的类SQL操作（如select、where、groupBy）到typed的RDD类操作（例如map、filter、flatMap）。有关更多详细信息，请参阅《SQL编程指南》。让我们来看几个可以使用的示例操作。

4.2.1 基本操作-Selection, Projection, Aggregation

Streaming支持对DataFrame/Dataset的大多数常见操作。本节后面将讨论不受支持的几个操作。

case class DeviceData(device: String, deviceType: String, signal: Double, time: DateTime)
val df: DataFrame = ... // streaming DataFrame with IOT device data with schema { device: string, deviceType: string, signal: double, time: string }
val ds: Dataset[DeviceData] = df.as[DeviceData] // streaming Dataset with IOT device data
// Select the devices which have signal more than 10
df.select("device").where("signal > 10") // using untyped APIs
ds.filter(_.signal > 10).map(_.device) // using typed APIs
// Running count of the number of updates for each device type
df.groupBy("deviceType").count() // using untyped API
// Running average signal for each device type
import org.apache.spark.sql.expressions.scalalang.typed
ds.groupByKey(_.deviceType).agg(typed.avg(_.signal)) // using typed API

还可以将Streaming DataFrame/Dataset注册为临时视图，然后对其应用SQL命令。

df.createOrReplaceTempView("updates")
spark.sql("select count(*) from updates") // returns another streaming DF

可以通过下面方式判断DataFrame是否是Streaming类型的：

df.isStreaming

调试过程中可能需要检查查询计划，因为Spark会在解释处理Streaming Dataset的SQL语句时注入带状态的操作。一旦带状态的操作被注入到查询计划中，您可能需要使用有状态操作中的注意事项来检查查询。（例如输出模式、水印、状态存储大小维护等）

4.2.2 基于事件时间的窗口函数

滑动事件时间窗口上的聚合对于Structured Streaming来说非常简单，与分组聚合非常相似。在分组聚合中，为用户指定的分组列中的每个唯一值维护聚合值（例如计数）。如果是基于窗口的聚合，则为事件时间所在的每个窗口维护聚合值。让我们通过一个例子来理解这一点。

修改一下快速示例，流现在包含行以及生成行的时间。我们希望以10分钟为窗口统计一次单词，并且每5分钟更新一次。也就是说，在12:00-12:10、12:05-12:15、12:10-12:20等10分钟窗口之间接收数据的wordcount。请注意，12:00-12:10表示在12:00之后但在12:10之前到达的数据。现在，考虑一个在12:07收到的单词。这个词应该增加对应于两个窗口12:00-12:10和12:05-12:15的计数。因此，计数将同时由分组键（即单词）和窗口（可根据事件时间计算）索引。

如下是结果集的情况：

window操作和group非常类似，因此在代码中可以通过groupBy和window两个函数进行表达：

import spark.implicits._
val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
// Group the data by window and word and compute the count of each group
val windowedCounts = words.groupBy(
window($"timestamp", "10 minutes", "5 minutes"),
$"word"
).count()

4.2.2.1 处理滞后数据和水印

现在考虑一下如下场景，如果其中一个事件迟到了会发生什么。例如，应用程序可以在12:11接收12:04（即事件时间）生成的单词。应用程序应该使用时间12:04而不是12:11来更新窗口12:00-12:10的旧计数。这种场景在基于窗口的分组中自然而然的就被处理了——Structured Streaming可以在很长一段时间内保持部分聚合的中间状态，以便后期数据可以正确地更新旧窗口的聚合，如下所示。

然而，考虑到系统会持续运行很多天，系统必须限制累积的中间内存状态的数量。这意味着系统需要知道何时可以从内存状态中删除旧聚合，因为应用程序将不再接收该聚合的延迟数据。为了实现这一点，我们在Spark 2.1中引入了水印，它让引擎自动跟踪数据中的当前事件时间，并尝试相应地清除旧状态。您可以通过指定事件时间列和阈值来定义查询的水印，该阈值根据事件时间来确定数据的预期延迟时间。对于在时间T结束的特定窗口，引擎将一直保留状态，并允许延迟数据更新状态，直到（引擎收到的最大事件时间-延迟阈值>T）。换句话说，阈值内的延迟数据将被聚合，但阈值之后的数据将开始被删除（有关确切的保证，请参阅本节后面的内容）。让我们用一个例子来理解这一点。在前面的示例中，我们可以使用withWatermark（）轻松定义水印，如下所示。

import spark.implicits._
val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }
// Group the data by window and word and compute the count of each group
val windowedCounts = words
.withWatermark("timestamp", "10 minutes")
.groupBy(
window($"timestamp", "10 minutes", "5 minutes"),
$"word")
.count()

在本例中，我们将根据列“timestamp”的值定义水印，并将“10分钟”定义为允许数据延迟的阈值。如果在update输出模式下运行此查询（稍后在“输出模式”一节中讨论），引擎将不断更新结果表中窗口的计数，直到窗口比水印早，水印比“timestamp”列中的当前事件时间滞后10分钟。如下图：

如图所示，引擎跟踪的最大事件时间是蓝色虚线，每个触发器开始处设置为（最大事件时间-“10分钟”）的水印是红线。例如，当引擎收到数据（12:14，dog）时，它会将下一个触发器的水印设置为12:04。此水印允许引擎在额外10分钟内保持中间状态，以便对延迟数据进行计数。例如，数据（12:09，cat）是无序且延迟的，它在windows 12:00-12:10和12:05-12:15中出现。由于它仍在触发器中的水印12:04之前，引擎仍将中间计数保持为状态，并正确更新相关窗口的计数。但是，当水印更新到12:11时，窗口的中间状态（12:00-12:10）被清除，所有后续数据（例如（12:04，donkey））被认为“太晚”，因此被忽略。请注意，在每次触发后，更新的计数（即紫色行）将被写入Sink作为触发输出，这由update的输出模式决定。

某些Sink（例如文件）可能不支持Update模式所需的细粒度的更新。为了使用它们，我们还支持Append模式，在这种模式下，只有最后的计数被写入Sink。这一点如下所示。

请注意，在非流数据集上使用withWatermark是无效的。由于水印不应以任何方式影响任何批处理查询，我们将直接忽略它。

与之前的Update模式类似，引擎为每个窗口保持中间计数。但是，部分计数不会更新到结果表，也不会写入接收器。引擎等待“10分钟”来计算延迟的数据，然后删除窗口<水印的中间状态，并将最终计数附加到结果表/接收器。例如，只有在水印更新到12:11之后，窗口12:00-12:10的最终计数才会附加到结果表中。

4.2.2.2 三类时间窗口

Spark支持三种时间窗口：滚动（固定）、滑动和会话。

滚动窗口是一系列固定大小、不重叠且连续的时间间隔。一个输入只能绑定到一个窗口。

从“固定大小”的角度来看，滑动窗口与翻滚窗口类似，但如果滑动的持续时间小于窗口的持续时间，窗口可以重叠，在这种情况下，输入可以绑定到多个窗口。

翻滚和滑动窗口使用窗口功能，这已在上述示例中描述。

与前两种类型相比，会话窗口具有不同的特性。会话窗口具有窗口长度的动态大小，具体取决于输入。会话窗口从一个输入开始，如果在间隔时间内收到以下输入，则会自行扩展。对于固定的间隔持续时间，当在收到最新输入后间隔持续时间内没有收到输入时，会话窗口关闭。

会话窗口使用session_window 函数。该函数的用法与窗口函数类似。

import spark.implicits._
val events = ... // streaming DataFrame of schema { timestamp: Timestamp, userId: String }
// Group the data by session window and userId, and compute the count of each group
val sessiOnizedCounts= events
.withWatermark("timestamp", "10 minutes")
.groupBy(
session_window($"timestamp", "5 minutes"),
$"userId")
.count()

除了静态值，我们还可以提供一个表达式，根据输入行动态指定间隙持续时间。请注意，间隙持续时间为负或为零的行将从聚合中筛选出来。

使用动态间隙持续时间，会话窗口的关闭不再依赖于最新的输入。会话窗口的范围是所有事件范围的并集，这些范围由事件开始时间和任务执行期间计算的间隔持续时间决定。

import spark.implicits._
val events = ... // streaming DataFrame of schema { timestamp: Timestamp, userId: String }

val sessionWindow = session_window($"timestamp", when($"userId" === "user1", "5 seconds")
.when($"userId" === "user2", "20 seconds")
.otherwise("5 minutes"))
// Group the data by session window and userId, and compute the count of each group
val sessiOnizedCounts= events
.withWatermark("timestamp", "10 minutes")
.groupBy(
Column(sessionWindow),
$"userId")
.count()

请注意，在流式任务中使用会话窗口时有一些限制，如下所示：

不支持“Update”作为输出模式。

在分组键中，除了session_窗口外，还应至少有一列。这一限制在批式任务中没有。

4.2.2.3 水印清除聚合状态的条件

需要注意的是，必须满足以下条件，水印才能清除聚合查询中的状态（从Spark 2.1.1开始，可能会在将来更改）。

输出模式必须为追加或更新。完整模式要求保留所有聚合数据，因此不能使用水印删除中间状态。有关每个输出模式的语义的详细解释，请参见“输出模式”部分。

聚合必须具有事件时间列或事件时间列上的窗口。

withWatermark必须在与聚合中使用的时间戳列相同的列上调用。例如，df.withWatermark("time", "1 min").groupBy("time2").count() 在追加输出模式下无效，因为水印是在聚合列以外的列上定义的。

必须在聚合之前调用withWatermark，才能使用水印。例如，df.groupBy("time").count().withWatermark("time", "1 min")在追加输出模式下无效。

4.2.2.4 水印聚合的语义保证

水印延迟（使用withWatermark设置）为“2小时”，可确保引擎不会丢弃延迟小于2小时的任何数据。换言之，在此之前处理的最新数据（就事件时间而言）落后2小时以内的任何数据都保证被聚合。

然而，保证仅在一个方向上是严格的。延迟超过2小时的数据不保证被删除；它可能会聚合，也可能不会聚合。数据越延迟，引擎处理数据的可能性就越小。

4.2.3 Join操作(留空)

4.2.4

流式重复数据消除

您可以使用事件中的唯一标识符消除数据流中的重复记录。这与使用唯一标识符列进行静态重复数据消除完全相同。查询将存储来自以前记录的必要数量的数据，以便可以过滤重复记录。与聚合类似，您可以使用带或不带水印的重复数据消除。

使用水印-如果对重复记录的到达时间有上限，则可以在事件时间列上定义水印，并使用guid和事件时间列进行重复数据消除。查询将使用水印从过去的记录中删除旧的状态数据，这些记录预计不会再获得任何副本。这限制了查询必须维护的状态量。

没有水印——由于重复记录的到达时间没有界限，因此查询将所有过去记录的数据存储为状态。

val streamingDf = spark.readStream. ... // columns: guid, eventTime, ...
// Without watermark using guid column
streamingDf.dropDuplicates("guid")
// With watermark using guid and eventTime columns
streamingDf
.withWatermark("eventTime", "10 seconds")
.dropDuplicates("guid", "eventTime")

4.2.5 处理多个水印(留空)

4.2.6 任意状态化的操作

许多用例需要比聚合更高级的有状态操作。例如，在许多用例中，您必须从事件的数据流中跟踪session。要执行这种session跟踪，必须将任意类型的数据保存为状态，并使用每个触发器中的数据流事件对状态执行任意操作。自Spark 2.2以来，可以使用操作mapGroupsWithState和更强大的操作flatMapGroupsWithState来实现这一点。这两种操作都允许在分组的数据集上应用用户定义的代码来更新用户定义的状态。有关更多具体细节，请查看API文档（Scala/Java）和示例（Scala/Java）。

尽管Spark无法检查并强制执行它，但状态函数应该根据输出模式的语义来实现。例如，在更新模式下，Spark不希望状态函数发出的行比当前水印加上允许的延迟记录延迟早，而在追加模式下，状态函数可以发出这些行。

（在session_window出来之前(Spark3.2)flatMapGroupsWithState是最常使用的替代方案，但常被开发人员诟病，Flink早已支持session_window）

4.2.7 不支持的操作

-流式数据集不支持多个流式聚合（即流式DF上的聚合链）。

-流式数据集不支持limit和获取前N行。

-不支持对流式数据集执行distinct的操作。

-在流式数据集上聚合后不支持重复数据消除操作。

-只有在完全输出模式,并聚合后，流式数据集才支持排序操作。

-流数据集上不支持几种类型的外部联接。有关更多详细信息，请参阅“连接操作”部分中的支持矩阵。

如果，您认为阅读这篇博客让您有些收获，不妨点击一下右下角的【推荐】。

如果，您希望更容易地发现我的新博客，不妨点击一下左下角的【关注我】。

如果，您对我的博客所讲述的内容有兴趣，请继续关注我的后续博客，我是【Arli】。

本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。