目录

• Overview
• Quick Example
• Programming Model
  • Basic Concepts
  • Handling Event-time and Late Data
  • Fault Tolerance Semantics
• API using Datasets and DataFrames
  • Creating streaming DataFrames and streaming Datasets
    • Input Sources
    • Schema inference and partition of streaming DataFrames/Datasets
  • Operations on streaming DataFrames/Datasets
    • Basic Operations - Selection, Projection, Aggregation
    • Window Operations on Event Time
    • Handling Late Data and Watermarking
    • Join Operations
    • Streaming Deduplication
    • Arbitrary Stateful Operations
    • Unsupported Operations
  • Starting Streaming Queries
    • Output Modes
    • Output Sinks
    • Using Foreach
  • Managing Streaming Queries
  • Monitoring Streaming Queries
    • Reading Metrics Interactively
    • Reporting Metrics programmatically using Asynchronous APIs
    • Reporting Metrics using Dropwizard
  • Recovering from Failures with Checkpointing
• Additional Information

1. Overview 概述

结构化流是一个建立在Spark SQL引擎上的可扩展性的并且容错的流处理引擎。你可以以你在静态数据上表示批量计算的相同方式表达你的流式计算。Spark SQL引擎将会不断地运行，并且当流数据持续到达时更新最终结果。你可以使用Scala、Java、Python或者R的Dataset/DataFrame API去表示流聚合、event-time窗口、stream-to-batch连接等。在同一优化的Spark SQL引擎中执行计算。最终，系统通过checkpointing和Write Ahead Logs（预写日志）确保端到端正好一次的容错保证。简而言之，结构化流提供快速、可扩展、容错、端到端一次性流处理，而用户不需要推断流。

在本指南中，我们将要学些编程模型和API。首先，让我们以一个简单示例开始——一个流word count。

2. Quick Example 快速学习示例

假设你想要维护从一个监听TCP socket的数据服务端接收到的文本数据的单词统计字数程序。你怎样使用结构化流表示这个呢。你可以查看Scala/Java/Pytho/R的完整代码。如果你下载Spark，你可以直接运行该示例。在任何情况下，我们一步一步地学习该示例并了解它是如何工作的。首先，我们必须引入必要的类并创建一个本地的SparkSession，这是Spark所有功能的开始点。

import org.apache.spark.sql.functions._

import org.apache.spark.sql.SparkSession

val spark = SparkSession

  .builder

  .appName("StructuredNetworkWordCount")

  .getOrCreate()

import spark.implicits._

接着，让我们创建流DataFrame，它代表监听localhost:9999的服务端接收到的文本数据，并转换为DataFrame以统计单词字数。

// 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”的字符串列，并且流文本数据的每一行成为表的一行。注意，目前没有接收到任何数据，因为我们正在设置转换，还没有启动。接下来，我们使用.as[String]将DataFrame转换为字符串Dataset（数据集），因此我们可以应用flatMap操作去将每一行分割为多个单词。组合成的单词Dataset（数据集）包含所有单词。最终，通过在Dataset（数据集）中的唯一值分组并统计它们数量，我们定义DataFrame。注意，这是流式DataFrame，它表示流的运行单词数量。

我们现在已经设置了对流式数据的查询。剩下的就是实际开始接收数据和计算计数。为了执行，我们将其设置为每次更新时在控制台上打印完整的一组计数（由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()

当上述代码执行之后，流式计算已经在后台启动啦。查询对象是活跃状态的流式查询的句柄，并且我们决定使用awaitTermination()等待查询终止，以防止在查询处于活动状态时进程退出。

为了实际执行该示例代码，你可以在你自己的Spark应用程序编译代码，或者当你已经下载Spark后简单运行该示例。稍后展示。你首先需要通过命令将Netcat作为数据服务端运行，该命令为：

$ nc -lk 9999

然后，在不同的终端，你可以通过命令启动示例：

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

然后，运行netcat服务器的终端上打印的任何行每秒都将会被统计并在屏幕上打印出来。如图所示：

3. Programming Model 编程模型

结构化流中的关键思想是将实时数据流视为一个不断附加的表。这导致了一个新的类似于批处理模型的流处理模型。你把你的流式计算表示为标准批量查询，就像在一个静态表上一样，Spark将它作为在无边界输入表上的增量查询运行。

3.1 Basic Concepts 基本概念

将输入数据流视为“输入表”。到达流的每个数据项就像添加到输入表的新行。

在输入生成的“结果表”上的查询。每个触发间隔（比如说，每隔1秒），新行会被附加到输入表中，最终更新结果表。无论什么时候结果表更新，我们都想要将改变的结果行写入一个外部sink（接收器）。

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

• Complete Mode 完全模式 —— 整个更新的结构表将会被写入到外部存储器。由存储连接器决定如何处理整个表的写入。
• Append Mode 附加模式 —— 自上次触发以来，只有附加到结果表的新行才会被写入到外部存储器。这仅适用于结果表中的现有行不期望被更改的查询。
• Update Mode 更新模式 ——自上次触发以来，只有附加到结果表的行才会被写入到外部存储器（自Spark2.1.1可用）。注意这与完全模式不同，这个模式只输出自上次触发更改的行。如果查询没有包含聚合，则它将等同于“追加”模式。

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

为了说明这个模式的用法，让我们了解前面快速示例上下文中的模型。第一行DataFrame是输入表，而最终的wordCounts DataFrame是结果表。注意在流lines DataFrame上生成wordCounts的查询与静态DataFrame完全相同。然而，当这个查询启动，Spark将会从socket连接不断地检查新数据。如果有新数据，Spark将会运行一个“增量”查询，将以前的运行计数与新数据结合起来计算更新的计算，如下所示：

注意结构化流不会实现整个表格。它从流数据源读取最新可用的数据，逐步处理它以更新结果，然后丢失源数据。它只是根据需要保留最小中间状态来更新结果（例如，在前面的例子中的中间计数）。

这个模式与许多其他流处理引擎明显不同。很多流系统要求用户自己维护运行聚合，因此不得不推理容错性和数据一致性（至少一次，或者多次，或者正好一次）。在这个模式下，当有新数据时，Spark有责任更新结果表，从而减轻用户对它的推理。作为一个例子，让我们看看这个模式如何处理基于事件时间的处理和迟到数据。

3.2 Handling Event-time and Late Data 处理Event-time和迟来的数据

Event-time是嵌入数据本身的时间。对于很多应用程序，你可能想要在事件时间上操作。例如，如果你想要得到每分钟由物联网硬件设备生成的事件数，那么你可能想要使用生成数据的时间（即数据中的事件时间），而不是Spark接收他们的时间。在这个模式中可以非常自然地表示事件时间 —— 每个来自硬件设备的事件都是表中的一行，并且事件时间是一行中的列值。这允许基于窗口的聚合（例如每分钟的事件数）仅仅是事件时间列上分组和聚合的特殊类型 —— 每个时间窗口是一个组，并且每一行可以属于多个窗口/组。因此，这个基于事件时间窗口的聚合查询可以在静态数据集和数据流上一致地定义，从而使用户的生活变得更加容易。

进一步，这个模式很自然地处理根据事件时间晚于预期到达的数据。因为Spark正在更新结果表，所以当有延迟的数据时，它完全控制更新旧的聚合，同时清除旧的聚合以限制中间状态数据的大小。自从Spark 2.1，我们已经支持水印，允许用户指定后期数据的阀值，并允许引擎相应地清除旧状态。详情参阅Window Operations 部分。

3.3 Fault Tolerance Semantics 容错语义

提供端到端正好一次的语义是结构化流设计背后的关键目标之一。为了实现它，我们设计结构化流来源，sinks和执行引擎，以便可靠地跟踪处理的确切进程，以便通过重新启动或者重新处理来处理任何类型的故障。每个流来源假定有偏移量（类似于Kafka偏移量，或者Kinesis序列化数字）去跟踪在流中读取的位置。引擎使用checkpointing和write ahead logs（预写日志）去记录每次触发处理的数据的偏移量范围。流式sinks被设计为处理重新处理的幂等。通过使用可重放来源和幂等sinks，结构化数据流可以确保任何失败下端到端正好一次的语义。

4 API using Datasets and DataFrames 使用Datasets和DataFrames的API

自从Spark 2.0，DataFrames和Datasets可以表示静态的有边界数据以及流式无边界数据。类似于静态的Datasets/DataFrames，你可以使用公共入口点SparkSession（Scala/Java/Python/R文档）从流式来源创建流式DataFrames/Datasets，并且在它们上应用与静态DataFrames/Datasets一样的相同操作。如果你不熟悉Datasets/DataFrames，强烈建议你使用DataFrame/Dataset编程指南熟悉它们。

4.1 Creating streaming DataFrames and streaming Datasets 创建流式DataFrames和流式Datasets

流式DataFrames可以通过SparkSession.readStream()返回的DataStreamReader接口（Scala/Java/Python）创建。在R中，是通过使用read.stream()方法。类似于读取接口以创建静态DataFrame，你可指定来源的详情 —— 数据格式，模式，选项等等。

4.1.1 Input Sources 输入来源

这里有几个构建的来源：

• File source 文件来源 —— 以数据流的形式读取目录中写入的文件。支持文件格式是text，csv,，json和parquet。参阅DataStreamReader接口的文档已获取更新的列表，以及每种文件格式的支持选项。注意文件必须以原子方式防止在给定的目录中，在大多数文件系统中，这可以通过文件移动操作来实现。
• Kafka source Kafka来源 —— 从Kafka读取数据。它已经兼容Kafka broker版本0.10.0或更高的版本。详情参阅Kafka Integration Guide。
• Socket 来源（用于测试）—— 从socket连接读取UTF8文本数据。监听服务器socket在driver端。注意这应该只能被用于测试因为它不提供端到端容错保证。
• Rate source（用于测试）—— 按每秒指定的行数生成数据，每个输出行包含一个timestamp（时间戳）和值。其中timestamp是一个包含消息分派时间的Timestamp类型，值为包含消息数的Long类型，从0开始作为第一行。这个来源用于测试和基准测试。

一些来源不是容错的，因为它们不保证在故障发生后数据可以通过checkpoint偏移量被重放。详情参阅早期关于容错语义的节点。

这里有些示例：

这些例子生成没有类型的流DataFrames，意味着DataFrame的模式在编译时不被检查，当查询提交时只在运行时间被检查。一些操作如map，flatMap等等，需要在编译时知道流DataFrames的类型。为此，我们可以使用与静态DataFrame相同的方法将这些无类型的流DataFrames转换为有类型的流数据集Datasets。详情参阅SQL Programming Guide。另外，更多细节可以在本文支持的流来源上参阅。

4.1.2 Schema inference and partition of streaming DataFrames/Datasets 模式推断和流式DataFrames/Datasets的分区

默认的，从基于来源的文件获取到的结构化流要求指定模式，而不是依赖于Spark自动推断。这个限制确保了一致的模式将用于流式查询，即使在发生故障的情况下。对于临时用例，你可以通过设置spark.sql.streaming.schemaInference为true来重新启用模式推断。

当名为/key=value/的子目录存在时，会发生分区发现，并且列表将自动递归到这些目录中。如果这些列出现在用户提供的模式中，则它们将由Spark根据正在读取的文件的路径填写。组成分区模式的目录在查询开始时必须存在，并且必须保持静态。例如，当/data/year=2015/存在时，添加/data/year=2016/是ok的，但是更改分区列是无效的（如通过创建目录/data/date=2016-04-17/）。

4.2 Operations on streaming DataFrames/Datasets 在流式DataFrames/Datasets上的操作

你可以将所有类型的操作应用于流DataFrames/Datasets上 —— 从无类型化的，类SQL操作（例如select，where，groupBy），到类型化的RDD类操作（例如map，filter，flatMap）。详情参阅SQL programming guide。让我们查看几个示例操作。

4.2.1 Basic Operations - Selection,Projection,Aggregation 基本操作——选择，投影，聚合

在DataFrame/Dataset上的大多数普通操作是支持流式的。在这个节点稍后讨论几个不支持的操作。

你也可以将流式DataFrame/Dataset注册为一个临时视图，然后在该视图上应用SQL命令。

注意，你可以通过使用df.isStreaming来唯一标识DataFrame/Dataset是否有流式数据。

4.2.2 Window Operations on Event Time 在事件时间上的窗口操作

滑动事件时间窗口的聚合对于结构化流式处理来说是很简单，并且与分组聚合很相似。在分组聚合中，聚合值（如counts）是通过在用户指定的分组列中的每个唯一值来维护的。在基于window的聚合的情况下，聚合值是通过行所在的事件时间所在的每个窗口来维护的。让我们举例说明。

想像我们的快速学习示例被修改后，现在流包含行和生成行的时间。替换运行单词统计，我们想要在10分钟的窗口内对单词进行计数，每5分钟更新一次。也就是说，对在10分钟窗口12:00 - 12:10，12:05 - 12:15，12:10 - 12:20等之间接收的单词进行单词统计。注意，12:00 - 12:10意味着数据是在12:00之后并在12:10之前接收的。现在，思考一个在12:07时接收的单词。对应于两个窗口12:00 - 12:10和12:05 - 12:15，这个单词数目应该增加。因此，计数将由两个分组键（即单词）和窗口（可以从事件时间计算）索引。

结果表将会如下所示。

由于窗口类似于分组，因此在代码中，你可以使用groupBy()和window()操作来 表达窗口化聚合。你可以在Scala/Java/Python查看完整的代码。

4.2.3 Handling Late Data and Watermarking 处理晚期数据和水印

现在思考如果其中一个事件晚到达应用程序会发生什么。例如，假设，一个在12:04时生成的单词（例如事件时间）可以在12:11被应用程序接收。应用程序应该使用时间12:04而不是12:11去更新窗口12:00 - 12:10的老旧统计数据。这个在我们基于窗口分组里很自然发生 —— 结构化流可以在很长一段时间内保持部分聚合的中间状态，以便后期数据可以正确更新旧窗口的聚合，如下所示。

然而，为了在几天内运行此查询，系统必须限制其累积的中间内存状态量。这意味着系统需要知道一个旧聚合什么时候可以从内存状态中移除掉，因为应用程序将不在为该聚合接收后期数据。为了实现这一点，在Spark2.1中，我们引入了watermarking（水印），它可以让引擎自动跟踪在数据中的当前事件时间，并相应地尝试清除旧状态。你可以通过指定事件时间列和阀值来定义查询的watermark（水印），阀值为数据预期在事件时间方面有多迟。对于在时间T开始的特定窗口，引擎将会保持状态并允许后期数据更新状态，直到（引擎看到的最大事件时间 - 延迟阀值 > T）。换而言之，在阀值中的后期数据将会被聚合，但是比阀值更迟的数据将会丢失。让我们通过一个示例来说明。我们可以使用withWatermark()来在前面示例中很容易定义watermarking（水印），如下。

在这个示例中，我们正在“timestamp”列的值上定义查询的watermark（水印），并且还定义了“10分钟”作为数据允许的最晚时间的阀值。如果查询在更新输出模式中运行（在输出模式节点介绍），那么引擎将会不断在结果表中更新窗口的数目，直到该窗口比水印更老旧，该水印滞后于列“timestamp”10分钟中的当前事件时间。这是说明。

如同说明中展示的，引擎跟踪的最大事件时间是蓝色标识的线条，并且在每次触发开始时watermark（水印）设置为（最大事件时间 - ‘10 mins’）为红线。例如，当引擎监控数据（12:14，dog）时， 它设置下次触发的水印为12:04。该水印让引擎维持额外10分钟的中间状态以允许后期数据能被统计。例如，数据（12:19，cat）是序列之外和延迟的，并且它落在windows12:00 - 12:10和12:05 - 12:15。因为，它仍然在触发器中水印12:04之前，所以引擎仍将中间计数保持为状态并正确更新相关窗口的计数。然而，当水印更新为12:11时，窗口（12:00 - 12:10）的中间状态将会被清除，并且所有后续数据（例如(12:04，donkey)）被认为是“太晚”并且因此被忽略。注意，在每次触发之后，根据更新模式的指示，更新后的计数（即紫色行）被写入到sink作为触发输出。一些sinks（如files）可能不支持更新模式要求的fine-grained updates（细粒度更新）。为了与它们一起工作，我们也支持Append Mode（追加模式），其中只有最后的计数被写入到sink。如下所示。

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

类似于前面的更新模式，引擎为每个窗口维持中间计数。然而，部分计数不会更新到结果表，也没写入到sink。引擎为后期数据等待10mins用于统计，然后丢失窗口<水印的中间状态，并且添加最终计数到结果表/sink。例如，只有在水印更新为12:11后，才将最终计数的窗口12:00 - 12:10附加到结果表中。

 水印清除聚合状态的条件。需要注意的是，水印清除聚合查询中的状态必须满足以下条件（截至Spark2.1.1，将来可能会发生变化）

• Output mode must be Append or Update输出模式必须是附加或更新的. 完整模式要求所有聚合数据都要保留，因此不能使用水印降低中间状态。详情参阅Output Modes节点以了解每个输出模式。
• 聚合必须有事件时间列或者事件时间列上的窗口。
• 在作为在聚合中使用的时间戳列的相同列上必须调用withWatermark。例如，df.withWatermark("time","1 min").groupBy("time2").count()在附加输出模式是无效的，因为水印在与聚合列不同的列上定义的。
• 在聚合之前必须调用withWatermark以使用水印细节。例如，df.groupBy("time").count().withWatermark("time","1 min")在附加输出模式中是无效的。

4.2.4 Join Operations 加入操作

流DataFrames可以加入到静态DataFrames以创建新的流DataFrames。这里有几个示例：

4.2.5 Streaming Deduplication 流式重复数据删除

使用事件中的唯一标识符你可以对数据流中的记录进行重复数据删除。这与使用唯一标识符列在静态中进行重复数据删除是一样的。查询将会存储来自前面记录中的必要数量的数据以便可以过滤重复的记录。类似于聚合，你可以使用带有或不带有水印的重复数据删除。

With watermark带有水印 —— 如果有重复记录可能迟到的上限，那么你可以在事件时间列上定义水印，并使用guid和事件时间列进行数据重复删除。查询将会使用水印移除来自过去记录的旧状态数据，这些记录不再希望有重复数据的。这限制了查询必须维护的状态量。

Without watermark不带有水印 —— 由于重复记录可能到达时没有界限，那么查询将所有来自过去记录的数据存储为状态。

4.2.6 Arbitrary Stateful Operations 任意有状态操作

很多用例要求比聚合更高级的有状态操作。例如，在许多用例中，你必须从事件的数据流中跟踪sessions（会话）。为了进行这种会话，你必须将任意类型的数据保存为状态，并在每次触发中使用数据流事件对状态执行任意操作。自从Spark2.2，这可以使用操作mapGroupsWithState和更强大的操作flatMapGroupsWithState来完成的。这两个操作允许你将用户定义的代码应用在分组数据集上以更新用户定义的状态。获取更多详情，可以在API文档（Scala/Java）和示例（Scala/Java）查看。

4.2.7 Unsupported Operations 不支持的操作

这里有几个不支持流式DataFrames/Datasets的DataFrame/Dataset操作。它们中的一些如下：

• 流式数据集还没有支持多个流式聚合(例如在流式DF上的聚合链)。
• 在流式数据集上不支持限制和取前N个行。
• 在流式数据集上的不同操作是不支持的。
• 只在完整输出模式内并在聚合之后，流式数据集才支持排序操作。

流式和静态数据集之间的外连接是有条件支持的。

• 不支持使用流式数据集的完全外连接。
• 不支持右侧流式数据集的左外连接。
• 不支持左侧流式数据集的右外连接。
• 两个流数据集之间的任何类型的连接尚未支持。

另外，还有一些数据集方法不适用于流式数据集。它们是马上运行查询并返回结果的操作，这对流式数据集没有意义。相反，这些功能可以通过明确开始流式查询来完成（查看下节点）。

• count() —— 不能从流式数据集返回单一计数。相反，使用ds.groupBy().count()，它返回包含运行计数的流式数据集。
• foreach() —— 替代使用ds.writeStream.foreach(...)（看下节点）
• show() —— 替代使用控制台sink（看下节点）

如果你尝试所有这些操作，你将会看到AnalysisException，如“operation XYZ is not supported with streaming DataFrames/Datasets”。在将来的Spark版本可能会支持其中一些操作，但还有其他一些基本上很难有效率实现流式传输数据。例如，不支持对输入流进行排序，因为它需要跟踪在流中接收到的所有数据。因此这基本上是难以有效执行的。

4.3 Starting Streaming Queries 开始流式查询

一旦你已经定义最终结果DataFrame/Dataset，剩下的就是让你开始流式计算。为了做到这一点，你必须使用通过Dataset.writeStream()返回的DataStreamWriter(Scala/Java/Python文档)。你必须在界面中指定以下一项或多项。

输出sinks的细节：数据格式，地点等等。

输出模式：指定什么写入到输出sink。

查询名字：可选地，指定查询的唯一名称进行标识。

触发间隔：可选地，指定触发间隔。如果没有指定，当前一处理已经完成时，系统将会立即检查新数据的可用性。如果由于之前的处理尚未完成而导致触发时间错过，则系统将立即触发处理。

Checkpoint地点：对于一些可以保证端到端容错的输出sinks，指定系统将写入所有checkpoint信息的地点。这应该是与HDFS兼容的容错文件系统中的目录。更多checkpointing语义参阅下节点。

4.3.1 Output Modes 输出模式

这里有几个类型的输出模式。

• Append mode（默认）附加模式 —— 这是默认的模式，自从上次触发以来只有添加到结果表中的新行才会输出到sink。这只支持添加到结果表的行不会更改的查询。因此，这个模式保证每一行将仅输出一次（假定容错sink）。例如，只有select,where,map,flatMap,filter,join等查询会支持附加模式。
• Complete mode 完整模式 —— 整个结果表在每次触发之后将输出到sink。支持聚合查询。
• Update mode 更新模式 —— （从Spark2.1.1可用）只有结果表中自上次触发更新的的行会输出到sink。更多信息在未来版本中添加。

不同类型的流式查询支持不同输出模式。这是兼容性矩阵。

4.3.2 Output Sinks 输出sinks

这里有几种类型的内置输出sinks。

File sink - 将输出存储到目录中。

Kafka sink - 将输出存储到Kafka的一个或多个主题中。

Foreach sink - 对输出中的记录进行任意计算。有关更多详细信息，请参阅后面的部分。

Console sink(for debugging) - 每次触发时将输出打印到控制台/标准输出。附加和完整输出模式都支持。这应该用于低数据量的调试目的，因为每次触发后整个输出被收集并存储在driver端的内存中。

Memory sink(for debugging) - 输出以内存表的形式存储在内存中。附加和完整输出模式，都是支持的。这应该用于低数据量的调试目的，因为每次触发后整个输出被收集并存储在driver端的内存中。因此，小心谨慎使用它。

一些sinks不是容错的，因为它们不保证输出的持久化，仅用于调试目的。在前面节点参阅fault-tolerance semantics。这是Spark所有sinks的细节。

注意你必须调用start()以实际启动查询的执行。这将返回一个StreamingQuery对象，它是持续运行的执行句柄。你可以使用这个对象去管理查询，这将会在下一个子节点讨论。现在，让我们通过几个例子了解。

// ========== DF with no aggregations ==========

val noAggDF = deviceDataDf.select("device").where("signal > 10")   

// Print new data to console

noAggDF

  .writeStream

  .format("console")

  .start()

// Write new data to Parquet files

noAggDF

  .writeStream

  .format("parquet")

  .option("checkpointLocation", "path/to/checkpoint/dir")

  .option("path", "path/to/destination/dir")

  .start()

// ========== DF with aggregation ==========

val aggDF = df.groupBy("device").count()

// Print updated aggregations to console

aggDF

  .writeStream

  .outputMode("complete")

  .format("console")

  .start()

// Have all the aggregates in an in-memory table

aggDF

  .writeStream

  .queryName("aggregates")    // this query name will be the table name

  .outputMode("complete")

  .format("memory")

  .start()

spark.sql("select * from aggregates").show()   // interactively query in-memory table

4.3.3 Using Foreach 使用Foreach

foreach操作允许在输出数据上进行任意操作计算。从Spark2.1开始，这只对Scala和Java可用。为了使用它，你将必须实现ForeachWriter接口（Scala/Java文档），当触发器产生了一系列作为输出的行时，这个接口就会调用方法。注意以下重要的点：

• writer必须序列化，因为它将会被序列化并发送到executors以执行。
• 所有三个方法，open、process和close在executors上被调用。
• 只有当open方法调用时，writer必须进行所有初始化（如打开连接，启动事务等等）。注意，如果对象创建后类中有任何初始化，那么将在driver中进行初始化，这可能不是你想要的。
• version和partition是open中的两个参数，open唯一代表一组需要发送的行。version是随着每次触发而增加的单调递增的id。partition是代表输出分区的id，因为输出是分布式的并且在多个executors上处理的。
• open可以使用version和partition来选择是否需要写入行的序列。相应地，它可以返回true（写入处理）或者false（不需要写入）。如果返回false，那么process将不会被任何行调用。例如，发生部分故障之后，失败触发器的某些输出分区可能已经被提交到数据库。根据数据库中的元数据，writer可以识别出已经提交的分区，然后相应地返回false来跳过再次提交它们。
• 无论什么时候调用open，close也将会被调用（除非由于一些错误导致JVM退出）。这是true即使open返回false。如果在处理和写数据时发生错误，close将会被调用并返回错误。你有责任去清除在open中已经创建的状态（例如连接，事务等等），以免发生资源泄漏。

4.4 Managing Streaming Queries 管理流式查询

当查询开始时创建的StreamingQuery对象可以用于监控和管理查询。

你可以在单例SparkSession中启动任意数量的查询。它们将全部同时运行并共享集群资源。你可以使用sparkSession.streams()去获取StreamingQueryManager(Scala/Java/Python文档)，它会用于管理当前活跃的查询。

4.5 Monitoring Streaming Queries 监控流式查询

这里有多种方式用于监控活跃的流式查询。你可以使用Spark的Dropwizard Metrics支持把metrics推送到外部系统，也可以通过编程方式访问它们。

4.5.1 Reading Metrics Interactively 以交互方式读取度量标准

你可以使用streamingQuery.lastProgress()和streamingQuery.status()直接获取活跃查询的当前状态和指标。lastProgress()在Scala和Java中返回StreamingQueryProgress对象以及在Python中具有相同字段的字典。它具有在流的上次触发中所取得进展的所有信息 —— 处理了哪些数据，处理的速率，延迟等等。还有streamingQuery.recentProgress，它返回最后几个进度的数组。另外，streamingQuery.status()在Scala和Java中返回StreamingQueryStatus对象以及在Python中具有相同字段的字典。它提供有关查询立即执行的信息 —— 触发器处于活动状态，正在处理数据等。

这里有几个示例：

val query: StreamingQuery = ...

println(query.lastProgress)

/* Will print something like the following.

{

  "id" : "ce011fdc-8762-4dcb-84eb-a77333e28109",

  "runId" : "88e2ff94-ede0-45a8-b687-6316fbef529a",

  "name" : "MyQuery",

  "timestamp" : "2016-12-14T18:45:24.873Z",

  "numInputRows" : 10,

  "inputRowsPerSecond" : 120.0,

  "processedRowsPerSecond" : 200.0,

  "durationMs" : {

    "triggerExecution" : 3,

    "getOffset" : 2

  },

  "eventTime" : {

    "watermark" : "2016-12-14T18:45:24.873Z"

  },

  "stateOperators" : [ ],

  "sources" : [ {

    "description" : "KafkaSource[Subscribe[topic-0]]",

    "startOffset" : {

      "topic-0" : {

        "2" : 0,

        "4" : 1,

        "1" : 1,

        "3" : 1,

        "0" : 1

      }

    },

    "endOffset" : {

      "topic-0" : {

        "2" : 0,

        "4" : 115,

        "1" : 134,

        "3" : 21,

        "0" : 534

      }

    },

    "numInputRows" : 10,

    "inputRowsPerSecond" : 120.0,

    "processedRowsPerSecond" : 200.0

  } ],

  "sink" : {

    "description" : "MemorySink"

  }

}

*/

println(query.status)

/*  Will print something like the following.

{

  "message" : "Waiting for data to arrive",

  "isDataAvailable" : false,

  "isTriggerActive" : false

}

*/

4.5.2 Reporting Metrics programmatically using Asynchronous APIs 以编程方式使用异步API报告度量

您还可以通过附加StreamingQueryListener（Scala / Java文档）来异步监控与SparkSession相关的所有查询。 一旦将自定义的StreamingQueryListener对象与sparkSession.streams.attachListener（）相关联，您将在查询开始和停止时以及在活动查询中取得进展时获得回调。 这里是一个例子

4.5.3 Reporting Metrics using Dropwizard 使用Dropwizard报告度量

Spark支持使用Dropwizard依赖包报告度量。要同时报告结构化流式查询的度量标准，您必须在SparkSession中显式启用配置spark.sql.streaming.metricsEnabled。

启用此配置后，在SparkSession中启动的所有查询都将通过Dropwizard向任何接收器配置（例如Ganglia，Graphite，JMX等）报告指标。

4.5.4 Recovering from Failures with Checkpointing 通过checkpointing从故障中恢复

如果发生故障或者故意关机，你可以恢复先前查询的进度和状态，并从它停止的地方继续。这是通过使用checkpointing和预写日志来完成的。你可以使用checkpoint位置配置查询，并且查询会将所有进度信息（即每个触发器中处理的偏移范围）和正在运行的聚合（例如，快递示例中的字数）保存到checkpoint位置。此checkpoint位置必须是HDFS兼容文件系统中的路径，并且可以在启动查询时再DataStreamWriter中设置为选项。

5. Additional Information

Further Reading

• See and run the Scala/Java/Python/R examples.
  • Instructions on how to run Spark examples
• Read about integrating with Kafka in the Structured Streaming Kafka Integration Guide
• Read more details about using DataFrames/Datasets in the Spark SQL Programming Guide
• Third-party Blog Posts
  • Real-time Streaming ETL with Structured Streaming in Apache Spark 2.1 (Databricks Blog)
  • Real-Time End-to-End Integration with Apache Kafka in Apache Spark’s Structured Streaming (Databricks Blog)
  • Event-time Aggregation and Watermarking in Apache Spark’s Structured Streaming (Databricks Blog)

Talks

• Spark Summit 2017 Talk - Easy, Scalable, Fault-tolerant Stream Processing with Structured Streaming in Apache Spark
• Spark Summit 2016 Talk - A Deep Dive into Structured Streaming