Structured Streaming 是一个可拓展,容错的,基于 Spark SQL 执行引擎的流处理引擎。使用小量的静态数据模拟流处理。伴随流数据的到来,Spark SQL 引擎会逐渐连续处理数据并且更新结果到最终的 Table 中。你可以在 Spark SQL 上引擎上使用 DataSet/DataFrame API 处理流数据的聚集,事件窗口,和流与批次的连接操作等。最后 Structured Streaming 系统快速,稳定,端到端的恰好一次保证,支持容错的处理。
- import org.apache.spark.sql.functions._
- import org.apache.spark.sql.SparkSession
- val spark = SparkSession
- .builder
- .appName("StructuredNetworkWordCount")
- .getOrCreate()
- import spark.implicits._
- 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()
- val query = wordCounts.writeStream
- .outputMode("complete")
- .format("console")
- .start()
- query.awaitTermination()
结构化流的关键思想是将实时数据流视为一个连续附加的表
将输入的数据当成一个输入的表格,每一个数据当成输入表的一个新行。
"Output" 是写入到外部存储的写方式,写入方式有不同的模式:
事件时间是嵌入在数据本身中的时间。对于许多应用程序,您可能希望在此事件时间操作。例如,如果要获取 IoT 设备每分钟生成的事件数,则可能需要使用生成数据的时间(即数据中的事件时间),而不是 Spark 接收的时间他们。此事件时间在此模型中非常自然地表示 - 来自设备的每个事件都是表中的一行,事件时间是该行中的一个列值。这允许基于窗口的聚合(例如每分钟的事件数)仅仅是偶数时间列上的特殊类型的分组和聚合 - 每个时间窗口是一个组,并且每一行可以属于多个窗口 / 组。因此,可以在静态数据集(例如,来自收集的设备事件日志)以及数据流上一致地定义这种基于事件时间窗的聚合查询,使得用户的生活更容易。
此外,该模型自然地处理基于其事件时间比预期到达的数据。由于 Spark 正在更新结果表,因此当存在延迟数据时,它可以完全控制更新旧聚合,以及清除旧聚合以限制中间状态数据的大小。由于 Spark 2.1,我们支持水印,允许用户指定后期数据的阈值,并允许引擎相应地清除旧的状态。稍后将在 "窗口操作" 部分中对此进行详细说明。
提供端到端的一次性语义是结构化流的设计背后的关键目标之一。为了实现这一点,我们设计了结构化流源,接收器和执行引擎,以可靠地跟踪处理的确切进展,以便它可以通过重新启动和 / 或重新处理来处理任何类型的故障。假定每个流源具有偏移量(类似于 Kafka 偏移量或 Kinesis 序列号)以跟踪流中的读取位置。引擎使用检查点和预写日志来记录每个触发器中正在处理的数据的偏移范围。流接收器被设计为用于处理再处理的幂等。结合使用可重放源和幂等宿,结构化流可以确保在任何故障下的端到端的一次性语义。
从 Spark 2.0 开始,DataFrames 和 Datasets 可以表示静态,有界数据,以及流式,无界数据。与静态 DataSets/ DataFrames 类似,您可以使用公共入口点 SparkSession(Scala / Java / Python 文档)从流源创建流 DataFrames /DataSets,并对它们应用与静态 DataFrames / Datasets 相同的操作。如果您不熟悉 Datasets / DataFrames,强烈建议您使用 DataFrame / Dataset 编程指南熟悉它们。
Streaming DataFrames 可以通过 SparkSession.readStream()返回的 DataStreamReader 接口(Scala / Java / Python docs)创建。类似于用于创建静态 DataFrame 的读取接口,您可以指定源 - 数据格式,模式,选项等的详细信息。
在 Spark 2.0,有几个内置的数据源:
这些示例生成无类型的流式 DataFrames,这意味着在编译时不检查 DataFrame 的模式,仅在提交查询时在运行时检查。一些操作,如 map,flatMap 等,需要在编译时知道类型。要做到这些,你可以使用与静态 DataFrame 相同的方法将这些无类型的流 DataFrames 转换为类型化流数据集。有关更多详细信息,请参阅 SQL 编程指南。此外,有关支持的流媒体源的更多详细信息将在文档中稍后讨论。
默认情况下,基于文件的源的结构化流要求您指定模式,而不是依靠 Spark 自动推断。此限制确保即使在发生故障的情况下,一致的模式也将用于流式查询。对于临时用例,可以通过将 spark.sql.streaming.schemaInference 设置为 true 来重新启用模式推断。
当名为 / key = value / 的子目录存在时,发生分区发现,并且列表将自动递归到这些目录中。如果这些列出现在用户提供的模式中,它们将由 Spark 根据正在读取的文件的路径填充。当查询开始时,组成分区方案的目录必须存在,并且必须保持静态。例如,可以添加 / data / year = 2016 / when / data / year = 2015 / 存在,但是更改分区列是无效的(即通过创建目录 / data / date = 2016-04-17 /)。
您可以对流式 DataFrames / 数据集应用各种操作 - 从无类型,类似 SQL 的操作(例如 select,where,groupBy)到类型化的 RDD 类操作(例如 map,filter,flatMap)。有关更多详细信息,请参阅 SQL 编程指南。让我们来看看一些你可以使用的示例操作。
- case class DeviceData(device: String, type: String, signal: Double, time: DateTime)
- val df: DataFrame = ... // streaming DataFrame with IOT device data with schema { device: string, type: 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("type").count() // using untyped API
- // Running average signal for each device type
- import org.apache.spark.sql.expressions.scalalang.typed._
- ds.groupByKey(_.type).agg(typed.avg(_.signal)) // using typed API
滑动事件时间窗口上的聚合通过结构化流直接进行。理解基于窗口的聚合的关键思想与分组聚合非常相似。在分组聚合中,为用户指定的分组列中的每个唯一值维护聚合值(例如计数)。在基于窗口的聚合的情况下,对于行的事件时间落入的每个窗口维持聚合值。让我们用插图来理解这一点。
想象一下,我们的快速示例被修改,流现在包含行以及生成行的时间。我们不想运行字数,而是计算 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 中查看以下示例的完整代码。
现在考虑如果事件中的一个到达应用程序的迟到会发生什么。例如,例如,在 12:04(即事件时间)生成的词可以由应用在 12:11 接收到。应用程序应使用时间 12:04 而不是 12:11 来更新窗口 12:00 - 12:10 的旧计数。这在我们的基于窗口的分组中自然地发生 - 结构化流可以长时间地保持部分聚合的中间状态,使得晚期数据可以正确地更新旧窗口的聚集,如下所示。
但是,要运行此查询的天数,系统必须绑定其累积的中间内存中状态的数量。这意味着系统需要知道何时可以从内存中状态删除旧聚合,因为应用程序将不再接收该聚合的延迟数据。为了实现这一点,在 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 分钟" 作为允许数据超时的阈值。如果此查询在 Append 输出模式(稍后在 "输出模式" 部分中讨论)中运行,则引擎将从列 "timestamp" 跟踪当前事件时间,并在最终确定窗口计数和添加之前等待事件时间的额外 "10 分钟" 他们到结果表。这是一个例证。
如图所示,由引擎跟踪的最大事件时间是蓝色虚线,并且在每个触发的开始处设置为(最大事件时间 - '10 分钟')的水印是红色线。例如,当引擎观察数据(12:14,狗),它将下一个触发器的水印设置为 12:04。对于窗口 12:00 - 12:10,部分计数保持为内部状态,而系统正在等待延迟数据。在系统发现数据(即(12:21,owl))使得水印超过 12:10 之后,部分计数被最终确定并附加到表。此计数将不会进一步更改,因为所有超过 12:10 的 "太晚" 数据将被忽略。
请注意,在追加输出模式下,系统必须等待 "延迟阈值" 时间才能输出窗口的聚合。如果数据可能很晚,(例如 1 天),并且您希望部分计数而不等待一天,这可能不是理想的。将来,我们将添加更新输出模式,这将允许每次更新聚合写入到每个触发器。
用于清除聚合状态的水印的条件重要的是要注意,水印应当满足以下条件以清除聚合查询中的状态(从 Spark 2.1 开始,将来会改变)。
流 DataFrames 可以与静态 DataFrames 连接以创建新的流 DataFrames。这里有几个例子。
- val staticDf = spark.read. ...
- val streamingDf = spark.readStream. ...
- streamingDf.join(staticDf, "type") // inner equi-join with a static DF
- streamingDf.join(staticDf, "type", "right_join") // right outer join with a static DF
但是,请注意,所有适用于静态 DataFrames / 数据集的操作在流式 DataFrames / 数据集中不受支持。虽然这些不支持的操作中的一些将在未来的 Spark 版本中得到支持,但还有一些基本上难以有效地在流数据上实现。例如,输入流数据集不支持排序,因为它需要跟踪流中接收的所有数据。因此,这在根本上难以有效地执行。从 Spark 2.0 开始,一些不受支持的操作如下:
此外,还有一些 Dataset 方法不能用于流数据集。它们是将立即运行查询并返回结果的操作,这对流数据集没有意义。相反,这些功能可以通过显式地启动流查询来完成(参见下一部分)。
如果您尝试任何这些操作,您将看到一个 AnalysisException 如 "操作 XYZ 不支持与流 DataFrames / 数据集"。
一旦定义了最终结果 DataFrame / Dataset,剩下的就是启动流计算。为此,您必须使用通过 Dataset.writeStream()返回的 DataStreamWriter(Scala / Java / Python 文档)。您必须在此界面中指定以下一个或多个。
有几种类型的输出模式:
不同类型的流查询支持不同的输出模式。这里是兼容性矩阵:
查询类型 | 支持的输出模式 | 注 | |
---|---|---|---|
无聚合的查询 | 支持完整模式 | 因为不可能保留结果表中的所有数据。 | |
带有聚合的聚合 | 聚合在带水印的事件时间聚合 | 附加,完全 | 附加模式使用水印来删除旧的聚合状态。但是窗口化聚合的输出被延迟了在 中指定的晚期阈值,如模式语义,在结束表之后,只有在结束表(在水印被交叉之后)才能将行添加一次。有关详细信息,请参阅延迟数据部分。完成模式不删除旧的聚合状态,因为从定义该模式保留结果表中的所有数据。 |
其他聚合 | 完全 | 不支持完全附加模式,因为聚合可以更新,因此违反了此模式的语义。完成模式不删除旧的聚合状态,因为从定义该模式保留结果表中的所有数据。 |
有几种类型的内置输出接收器:
下面是所有接收器的表格和相应的设置:
接收器 | 支持的输出模式 | 用法 | 容错 | 备注 |
---|---|---|---|---|
文件接收器 | Append |
|
Yes | 支持对分区表的写入。按时间分区可能有用。 |
Foreach 接收器 | 所有模式 |
|
取决于 ForeachWriter 实现 | 更多细节在下一节 |
控制台接收器 | Append, Complete |
|
No | |
内存接收器 | Append, Complete |
|
No | 将输出数据保存为表,用于交互式查询。表名是查询名称。 |
最后,你必须调用 start()才能真正开始执行查询。这返回一个 StreamingQuery 对象,它是连续运行的执行的句柄。您可以使用此对象来管理查询,我们将在下一小节中讨论。现在,让我们通过几个例子来理解这一切。
- // ========== DF with no aggregations ==========
- Dataset<Row> 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()
- .parquet("path/to/destination/directory")
- .start();
- // ========== DF with aggregation ==========
- Dataset<Row> 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
foreach 操作允许对输出数据计算任意操作。从 Spark 2.1 开始,这只适用于 Scala 和 Java。要使用这个,你必须实现接口 ForeachWriter(Scala / Java docs),它有一个方法,当触发后产生一系列行作为输出时被调用。请注意以下要点。
编写器必须是可序列化的,因 - 为它将被序列化并发送到执行器以供执行。
启动查询时创建的 StreamingQuery 对象可用于监视和管理查询。
- StreamingQuery query = df.writeStream().format("console").start(); // get the query object
- query.id(); // get the unique identifier of the running query
- query.name(); // get the name of the auto-generated or user-specified name
- query.explain(); // print detailed explanations of the query
- query.stop(); // stop the query
- query.awaitTermination(); // block until query is terminated, with stop() or with error
- query.exception(); // the exception if the query has been terminated with error
- query.sourceStatus(); // progress information about data has been read from the input sources
- query.sinkStatus(); // progress information about data written to the output sink
您可以在单个 SparkSession 中启动任意数量的查询。他们将同时运行共享集群资源。您可以使用 sparkSession.streams()获取可用于管理当前活动查询的 StreamingQueryManager(Scala / Java / Python 文档)。
- SparkSession spark = ...
- spark.streams().active(); // get the list of currently active streaming queries
- spark.streams().get(id); // get a query object by its unique id
- spark.streams().awaitAnyTermination(); // block until any one of them terminates
有两个 API 用于以交互式和异步方式监视和调试活动的查询。
您可以使用 streamingQuery.lastProgress()和 streamingQuery.status()直接获取活动查询的当前状态和指标。 lastProgress()在 Scala 和 Java 中返回一个 StreamingQueryProgress 对象,在 Python 中返回一个具有相同字段的字典。它具有关于在流的最后触发中所进行的进展的所有信息 - 什么数据被处理,什么是处理速率,等待时间等。还有 streamingQuery.recentProgress,它返回最后几个进度的数组。
此外,streamingQuery.status()在 Scala 和 Java 中返回 StreamingQueryStatus 对象,在 Python 中返回具有相同字段的字典。它提供有关查询立即执行的操作的信息 - 是触发器活动,正在处理数据等。这里有几个例子。
- StreamingQuery query = ...
- System.out.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"
- }
- }
- */
- System.out.println(query.status());
- /* Will print something like the following.
- {
- "message" : "Waiting for data to arrive",
- "isDataAvailable" : false,
- "isTriggerActive" : false
- }
- */
您还可以通过附加 StreamingQueryListener(Scala / Java docs)异步监视与 SparkSession 相关联的所有查询。使用 sparkSession.streams.attachListener()附加自定义 StreamingQueryListener 对象后,当查询启动和停止以及活动查询中有进度时,您将获得回调。这里是一个例子
- SparkSession spark = ...
- spark.streams.addListener(new StreamingQueryListener() {
- @Overrides void onQueryStarted(QueryStartedEvent queryStarted) {
- System.out.println("Query started: " + queryStarted.id());
- }
- @Overrides void onQueryTerminated(QueryTerminatedEvent queryTerminated) {
- System.out.println("Query terminated: " + queryTerminated.id());
- }
- @Overrides void onQueryProgress(QueryProgressEvent queryProgress) {
- System.out.println("Query made progress: " + queryProgress.progress());
- }
- });
在故障或故意关闭的情况下,您可以恢复先前查询的先前进度和状态,并继续在其停止的地方。这是通过使用检查点和预写日志来完成的。您可以配置具有检查点位置的查询,并且查询将保存所有进度信息(即每个触发器中处理的偏移范围)和正在运行的聚合(例如快速示例中的字计数)到检查点位置。此检查点位置必须是 HDFS 兼容文件系统中的路径,并且可以在启动查询时在 DataStreamWriter 中设置为选项。
- aggDF
- .writeStream()
- .outputMode("complete")
- .option("checkpointLocation", "path/to/HDFS/dir")
- .format("memory")
- .start();
来源: http://www.cnblogs.com/cutd/p/6590354.html