2.4　Spark 2.4 Streaming

Spark 2.0为我们带来了一个新的流处理框架Structured Streaming，这是一个基于Spark SQL和Catalyst优化器构建的高级流API。它允许用户使用与操作静态数据的DataFrame/Dataset API对流数据进行编程，利用Catalyst优化器自动地增量化查询计划，并且它不但支持流数据的不断写入，还支持其他的静态数据的插入。

在Apache Spark 2.2.X版本的基础上，Apache Spark 2.3.X版本、Apache Spark 2.4.3版本中Structured Streaming的更新如下。

　SPARK-27494：当处理键值为空时，修复了Kafka source v2中的一个错误。

　SPARK-27550：修复“test dependencies.sh”对Scala 2.12不使用“kafka-0-8”配置文件。

　Continuous Processing：一种新的执行引擎，只更改一行用户代码，可以以亚毫秒的端到端延迟执行流式查询。

　Stream-Stream Joins：能够连接两个数据stream，缓冲行，直到匹配的元组到达另一个stream。谓词可用于事件时间列，以绑定需要保留的状态量。

　Streaming API V2：一种用于新数据源和接收器的实验API，适用于批量、微批量和连续执行。注意，这个API仍在进行积极的开发。

　SPARK-24565：使用ForeachBatch (Python,Scala,Java),将每个微批次的输出为DataFrame。

　SPARK-24396：为foreach和ForeachWriter添加了Python API。

　SPARK-25005：支持“kafka.isolation.level”以只读模式使用事务生成器，从Kafka主题中提交记录。

　SPARK-24662：支持追加或完成模式的LIMIT运算符。

　SPARK-24763：从流聚合的值中删除冗余密钥数据。

　SPARK-24156：当输入流中没有数据时，使用有状态操作（mapGroupsWithState, stream-stream join, streaming aggregation, streaming dropDuplicates）更快生成输出结果和状态清理。

　SPARK-24730：支持在查询中有多个输入流时选择最小或最大水印。

　SPARK-25399：修复了一个错误，在该错误中，将连续处理的执行线程重新用于微批量流可能导致正确性问题。

　SPARK-18057：将Kafka客户端版本从0.10.0.1升级到2.0.0。

Spark 2.3引入了新的低延迟处理模式，称为连续处理，它可以在至少一次的情况下实现低至1ms的端到端延迟。在查询中不更改数据集／数据帧操作的情况下，可以根据应用程序要求选择模式。

2.4.1　Structured Streaming

Structured Streaming会通过Checkpoint（检查点）机制和预写日志的方式来确保对流数据的语义一致性exactly-once处理，让整个处理过程更加可靠。

在Spark 2.2版本中增加了对kafka 0.10的支持，为记录添加了可见的基于事件时间水印的延迟，并且现在Structured Streaming已经支持对所有格式的文件操作。

到Spark 2.1版本为止，Structured Streaming还处于试验阶段，还需几个版本的迭代，才能稳定到可以在生产环境下使用的程度。

下面结合官方文档了解一下Structured Streaming。

首先写一个程序来监听网络端口发来的内容，然后进行WordCount。

第一步：创建程序入口SparkSession，并引入spark.implicits来允许Scalaobject隐式转换为DataFrame。

第二步：创建流。配置从socket读取流数据，地址和端口为localhost:9999。

第三步：进行单词统计。这里lines是DataFrame，使用as[String]给它定义类型转换为DataSet。之后在DataSet里进行单词统计。

第四步：创建查询句柄，定义打印结果方式并启动程序。这里使用writeStream方法，输出模式为全部输出到控制台。

接下来运行该程序，在Linux命令窗口运行nc–lk 9999开启9999端口。

然后在Spark的Home目录下提交程序，传入IP和端口号。

程序启动之后，在前面的命令窗口中输入单词apache spark，程序会运行并打印出结果，如图2-1所示。

最后再输入hello spark，程序会再次运行并打印出如图2-2所示的结果。

可见，融合了DataSetAPI的流处理框架的程序代码十分简捷，并且执行效率也会比原来的SparkStreaming更高。

Structured Streaming的关键思想如图2-3所示：把数据流视为一张数据不断增加的表，这样用户就可以基于这张表进行数据处理，就好像使用批处理来处理静态数据一样，但实际上Spark底层是把新数据不断地增量添加到这张无界的表的下一行中。

Structured Streaming共有3种输出模式，这3种模式都只适用于某些类型的查询。

（1）CompleteMode：完整模式。整个更新的结果表将被写入外部存储器。由存储连接器决定如何处理整张表的写入。聚合操作以及聚合之后的排序操作支持这种模式。

（2）AppendMode：附加模式。只有自上次触发执行后在结果表中附加的新行会被写入外部存储器。这仅适用于结果表中的现有行不会更改的查询，如select、where、map、flatMap、filter、join等操作支持这种模式。

（3）UpdateMode：更新模式（这个模式将在以后的版本中实现）。只有自上次触发执行后在结果表中更新的行将被写入外部存储器（不输出未更改的行）。

2.4.2　增量输出模式

上面例子中使用的是CompleteMode，程序中接收数据的输入表是lines，它是一个DataFrame，新来的数据会被添加进去。之后的wordCounts是结果表。当程序启动时，Spark会不断检测是否有新数据加入到lines中，如果有新数据，则运行一个增量的查询，与上一次查询的结果合并，并且更新结果表，如图2-4所示。

这个看似非常简单的设计，背后的实现逻辑却并不简单，因为其他的流处理框架用户是需要自己推理怎么解决容错和数据一致性的问题，如经典的at-most-once、at-least-once、exactly-once问题，而在上面的CompleteMode下，Spark因为只在有新数据进来的时候才会更新结果，所以帮用户解决了这些问题。我们可以通过StructuredStreaming在基于event-time的操作和延迟数据的处理这两个问题上的解决方式来简单地了解背后的实现机制。

event-time是嵌入事件本身的时间，记录了事件发生的时间。很多时候我们需要用这个时间来实现业务逻辑，例如，我们要获取IOT设备每分钟产生的事件数量，则可能需要使用生成数据的时间，而不是Spark接收它们的时间。在这个模式下，event-time作为每行数据中的一列，可以用于基于时间窗口的聚合，也正因如此，基于event-time的窗口函数可以同样被定义在静态数据集上（如日志文件等）。

而关于容错方面，提供端到端的exactly-once语义是Structured Streaming的主要设计目标之一，要实现exactly-once，就要考虑数据源（sources）、执行引擎（execution）和存储（sinks）3个方面。Structured Streaming是这样实现的：假定每个数据源都有偏移量（与kafka的offsets类似）用来追溯数据在数据流中的位置；在执行引擎中会通过checkpoint（检查点）和WAL（writeaheadlogs预写日志）记录包括被处理的数据的偏移量范围在内的程序运行进度信息；在存储层设计成多次处理结果幂等，即处理多次结果相同。这样确保了Structured Streaming端到端exactly-once的语义一致性。

上面提到的event-time列带来的另一个好处是，可以很自然地处理“迟到”的数据（如没有按照event-time的时间顺序被Spark接收），因为更新旧的结果表时，可以完全控制更新和清除旧的聚合结果来限制处于中间状态的数据（窗口函数中，这些数据可以是有状态的）的大小。Spark 2.1引入的watermarking允许用户指定延迟数据的阈值，也允许引擎清除掉旧的状态。

先来看一下基础的窗口函数在Structured Streaming中的应用。例如，要对10min的单词进行计数，每5min统计一次，我们要把滑动窗口的大小设置为10min，每5min滑动一次。具体代码如下：

如图2-5所示，从12:00开始，12:05启动第一次查询，数据是12:00—12:10时间段的，当然，此时只有前5min的数据被统计进来。在12:10分时统计12:05—12:15时间段的数据（此时也是只有前5min的数据），并把上一个窗口的后5min的数据（即12:05—12:10时间段的数据）的统计结果合并到上一个窗口的结果中去，之后每次启动查询，都会把上一个窗口的查询结果补全，并把本窗口的前5min数据的统计结果记下来。

但是，我们可能会遇到这种情况：如图2-6所示，12:04产生的数据，一直到12:10之后才被程序接收到，此时数据依然会被正确地合并到对应的窗口中去，但是这样会导致查询结果长时间处于中间状态，而如果要长时间运行程序，就必须限制累积到内存中的中间状态结果的大小，这就要求系统知道什么时候清理这些中间状态的数据。

在Spark 2.1中引入水印（watermarking），使得系统可以自动跟踪目前的event-time，并按照用户指定的时间清理旧的状态。使用方法如下：

这里设置时间为10min，在Append模式下，具体执行逻辑如图2-7所示。

图2-7中点线表示目前收到的数据的最大event-time，实线是水印（计算方法是运算截止到触发点时收到的数据最大的event-time减去latethreshold，也就是减去10），当水印时间小于窗口的结束时间时，计算的数据都被保留为中间数据，当水印时间大于窗口结束时间时，就把这个窗口的运算结果加入到结果表中去，之后即使再收到属于这个窗口的数据，也不再进行计算，而直接忽略掉。图中在12:15时刻最大的event-time为12:14，所以水印时间为12:04，小于第一个窗口的12:10，所以此时计算的数据处于中间状态，存放在内存中，在下一个触发点也就是12:20时，最大event-time为12:21，水印时间是12:11，大于12:10，此时认为12:00—12:10窗口的数据已经完全到达，把中间结果中属于这个窗口的数据写入到结果表中，并且之后不再对这个结果进行更新。

需要注意的是，在Append模式下，系统在输出某个窗口的运行结果之前一定会根据设置等待延迟数据，这就意味着如果用户把延迟阈值设置为1天，那么在这一天内就无法看到这个窗口的（中间）结果。不过，在以后的版本中，Spark会加入Update模式来解决这一问题。

Structured Streaming目前有4种输出数据的处理方式。

（1）writeStream.format("parquet").start()，输出到文件中，Append模式支持这种方式。这种方式自带容错机制。

（2）writeStream.format("console").start()，输出到控制台，用于调试。Append模式和Complete模式支持这种方式。

（3）writeStream.format("memory").queryName("table").start()，以table的形式输出到内存，可以在之后的程序中使用spark.sql(select * from table).show来对结果进行处理。这种方式同样用于调试。Append模式和Complete模式支持这种方式。

（4）writeStream.foreach(…).start()，对输出结果进行进一步处理。要使用它，用户必须传入一个实现ForeachWriter接口的writer类，这个类必须是可序列化的，因为稍后会把结果发送到不同的executor进行分布式计算。这个接口有3个方法要实现，即open、process、close。

　open方法：open方法有两个入参：version、partitionId，其中partitionId是分区ID，version是重复数据删除的唯一ID标识。open方法用于处理Executor分区的初始化（如打开连接启动事物等）。version用于数据失败时重复数据的删除，当从失败中恢复时，一些数据可能生成多次，但它们具有相同的版本。如果此方法发现使用的partitionId和version这个分区已经处理后，可以返回false，以跳过进一步的数据处理，但close仍然将被要求清理资源。

　process方法：调用此方法处理Executor中的数据。此方法只在open时调用，返回true。

　close方法：停止执行Executor一个分区中的新数据时调用，保证被调用open时返回true，或者返回false。然而，在下列情况下，close不被调用：JVM崩溃，没有抛出异常Throwable；open方法抛出异常Throwable。

最后要说明的是，为了程序在重启之后可以接着上次的执行结果继续执行，需要设置检查点，方式如下：