是时候放弃 spark streaming，转向 structured streaming 了

1 .是 Spark Streaming, Structured Streaming 正如在之前的那篇⽂文章中 Spark Streaming 设计原理理 中说到 Spark 团队之后对 Spark Streaming 的 维护可能越来越少，Spark 2.4 版本的 Release Note ⾥里里⾯面果然⼀一个 Spark Streaming 相关的 ticket 都 没有。相⽐比之下，Structured Streaming 有将近⼗十个 ticket 说明。所以各位同学，是时候舍弃 Spark Streaming 转向 Structured Streaming 了了，当然理理由并不不⽌止于此。我们这篇⽂文章就来分析⼀一下 Spark Streaming 的不不⾜足，以及Structured Streaming 的设计初衷和思想是怎么样的。⽂文章主要参考今年年 （2018 年年）sigmod 上⾯面的这篇论⽂文：Structured Streaming: A Declarative API for Real-Time Applications in Apache Spark 。 ⾸首先可以注意到的了了论⽂文标题中的 Declarative API，中⽂文⼀一般叫做声明式编程 API。⼀一般直接看到这 个词可能不不知道什什么意思，但是当我们列列出他的对⽴立单词：Imperative API，中⽂文⼀一般叫命令式编程 API，仿佛⼀一切都明了了了了。是的，没错，Declarative 只是表达出我们想要什什么，⽽而 Imperative 则是 说为了了得到什什么我们需要做哪些东⻄西⼀一个个说明。举个例例⼦子，我们要⼀一个糕点，去糕点店直接去定做 告诉店员我们要什什么样式的糕点，然后店员去给我们做出来，这就是 Declarative。⽽而 Imperative 对 应的就是⾯面粉店了了。 0. Spark Streaming 不不 在开始正式介绍 Structured Streaming 之前有⼀一个问题还需要说清楚，就是 Spark Streaming 存在 哪些不不⾜足？总结⼀一下主要有下⾯面⼏几点： Processing Time Event Time。⾸首先解释⼀一下，Processing Time 是数据到达 Spark 被 处理理的时间，⽽而 Event Time 是数据⾃自带的属性，⼀一般表示数据产⽣生于数据源的时间。⽐比如 IoT 中， 传感器器在 12:00:00 产⽣生⼀一条数据，然后在 12:00:05 数据传送到 Spark，那么 Event Time 就是 12:00:00，⽽而 Processing Time 就是 12:00:05。我们知道 Spark Streaming 是基于 DStream 模型的 micro-batch 模式，简单来说就是将⼀一个微⼩小时间段，⽐比如说 1s，的流数据当前批数据来处理理。如果 我们要统计某个时间段的⼀一些数据统计，毫⽆无疑问应该使⽤用 Event Time，但是因为 Spark Streaming 的数据切割是基于 Processing Time，这样就导致使⽤用 Event Time 特别的困难。 Complex, low-level api。这点⽐比较好理理解，DStream （Spark Streaming 的数据模型）提供的 API 类似 RDD 的 API 的，⾮非常的 low level。当我们编写 Spark Streaming 程序的时候，本质上就是要去 构造 RDD 的 DAG 执⾏行行图，然后通过 Spark Engine 运⾏行行。这样导致⼀一个问题是，DAG 可能会因为开 发者的⽔水平参差不不⻬齐⽽而导致执⾏行行效率上的天壤之别。这样导致开发者的体验⾮非常不不好，也是任何⼀一个 基础框架不不想看到的（基础框架的⼝口号⼀一般都是：你们专注于⾃自⼰己的业务逻辑就好，其他的交给 我）。这也是很多基础系统强调 Declarative 的⼀一个原因。 reason about end-to-end application。这⾥里里的 end-to-end 指的是直接 input 到 out，⽐比如 Kafka 接⼊入 Spark Streaming 然后再导出到 HDFS 中。DStream 只能保证⾃自⼰己的⼀一致性语义是 exactly- once 的，⽽而 input 接⼊入 Spark Streaming 和 Spark Straming 输出到外部存储的语义往往需要⽤用户⾃自 ⼰己来保证。⽽而这个语义保证写起来也是⾮非常有挑战性，⽐比如为了了保证 output 的语义是 exactly-once 语义需要 output 的存储系统具有幂等的特性，或者⽀支持事务性写⼊入，这个对于开发者来说都不不是⼀一 件容易易的事情。 

2 . 。尽管批流本是两套系统，但是这两套系统统⼀一起来确实很有必要，我们有时候确实 需要将我们的流处理理逻辑运⾏行行到批数据上⾯面。关于这⼀一点，最早在 2014 年年 Google 提出 Dataflow 计 算服务的时候就批判了了 streaming/batch 这种叫法，⽽而是提出了了 unbounded/bounded data 的说 法。DStream 尽管是对 RDD 的封装，但是我们要将 DStream 代码完全转换成 RDD 还是有⼀一点⼯工作 量量的，更更何况现在 Spark 的批处理理都⽤用 DataSet/DataFrame API 了了。 1. Structured Streaming Structured Streaming 在 Spark 2.0 版本于 2016 年年引⼊入，设计思想参考很多其他系统的思想，⽐比如 区分 processing time 和 event time，使⽤用 relational 执⾏行行引擎提⾼高性能等。同时也考虑了了和 Spark 其他组件更更好的集成。Structured Streaming 和其他系统的显著区别主要如下： Incremental query model: Structured Streaming 将会在新增的流式数据上不不断执⾏行行增量量查 询，同时代码的写法和批处理理 API （基于 Dataframe 和 Dataset API）完全⼀一样，⽽而且这些 API ⾮非常的简单。 Support for end-to-end application: Structured Streaming 和内置的 connector 使的 end- to-end 程序写起来⾮非常的简单，⽽而且 "correct by default"。数据源和 sink 满⾜足 "exactly-once" 语义，这样我们就可以在此基础上更更好地和外部系统集成。 Spark SQL ：我们知道 Spark SQL 执⾏行行引擎做了了⾮非常多的优化⼯工作，⽐比如执⾏行行计 划优化、codegen、内存管理理等。这也是 Structured Streaming 取得⾼高性能和⾼高吞吐的⼀一个原 因。 2. Structured Streaming 下⾯面我们看⼀一下 Structured Streaming 的核⼼心设计。 Input and Output: Structured Streaming 内置了了很多 connector 来保证 input 数据源和 output sink 保证 exactly-once 语义。⽽而实现 exactly-once 语义的前提是： 

3 . Input 数据源必须是可以 replay 的，⽐比如 Kafka，这样节点 crash 的时候就可以重新读取 input 数据。常⻅见的数据源包括 Amazon Kinesis, Apache Kafka 和⽂文件系统。 Output sink 必须要⽀支持写⼊入是幂等的。这个很好理理解，如果 output 不不⽀支持幂等写⼊入，那 么⼀一致性语义就是 at-least-once 了了。另外对于某些 sink, Structured Streaming 还提供了了 原⼦子写⼊入来保证 exactly-once 语义。 API: Structured Streaming 代码编写完全复⽤用 Spark SQL 的 batch API，也就是对⼀一个或者多 个 stream 或者 table 进⾏行行 query。query 的结果是 result table，可以以多种不不同的模式 （append, update, complete）输出到外部存储中。另外，Structured Streaming 还提供了了⼀一 些 Streaming 处理理特有的 API：Trigger, watermark, stateful operator。 Execution: 复⽤用 Spark SQL 的执⾏行行引擎。Structured Streaming 默认使⽤用类似 Spark Streaming 的 micro-batch 模式，有很多好处，⽐比如动态负载均衡、再扩展、错误恢复以及 straggler （straggler 指的是哪些执⾏行行明显慢于其他 task 的 task）重试。除了了 micro-batch 模 式，Structured Streaming 还提供了了基于传统的 long-running operator 的 continuous 处理理模 式。 Operational Features: 利利⽤用 wal 和状态存储，开发者可以做到集中形式的 rollback 和错误恢 复。还有⼀一些其他 Operational 上的 feature，这⾥里里就不不细说了了。 3. Structured Streaming 可能是受到 Google Dataflow 的批流统⼀一的思想的影响，Structured Streaming 将流式数据当成⼀一个 不不断增⻓长的 table，然后使⽤用和批处理理同⼀一套 API，都是基于 DataSet/DataFrame 的。如下图所示， 通过将流式数据理理解成⼀一张不不断增⻓长的表，从⽽而就可以像操作批的静态数据⼀一样来操作流数据了了。 在这个模型中，主要存在下⾯面⼏几个组成部分： Input Unbounded Table: 流式数据的抽象表示 Query: 对 input table 的增量量式查询 Result Table: Query 产⽣生的结果表 Output: Result Table 的输出 

4 .下⾯面举⼀一个具体的例例⼦子，NetworkWordCount，代码如下: 1 // Create DataFrame representing the stream of input lines from connection to localhost:9999 2 val lines = spark.readStream 3 .format("socket") 4 .option("host", "localhost") 5 .option("port", 9999) 6 .load() 7 8 // Split the lines into words 9 val words = lines.as[String].flatMap(_.split(" ")) 10 11 // Generate running word count 12 val wordCounts = words.groupBy("value").count() 13 14 // Start running the query that prints the running counts to the console 15 val query = wordCounts.writeStream 16 .outputMode("complete") 17 .format("console") 18 .start() 代码实际执⾏行行流程可以⽤用下图来表示。把流式数据当成⼀一张不不断增⻓长的 table，也就是图中的 Unbounded table of all input。然后每秒 trigger ⼀一次，在 trigger 的时候将 query 应⽤用到 input table 中新增的数据上，有时候还需要和之前的静态数据⼀一起组合成结果。query 产⽣生的结果成为 Result Table，我们可以选择将 Result Table 输出到外部存储。输出模式有三种： Complete mode: Result Table 全量量输出 

5 . Append mode (default): 只有 Result Table 中新增的⾏行行才会被输出，所谓新增是指⾃自上⼀一次 trigger 的时候。因为只是输出新增的⾏行行，所以如果⽼老老数据有改动就不不适合使⽤用这种模式。 Update mode: 只要更更新的 Row 都会被输出，相当于 Append mode 的加强版。 和 batch 模式相⽐比，streaming 模式还提供了了⼀一些特有的算⼦子操作，⽐比如 window, watermark, statefaul oprator 等。 window，下图是⼀一个基于 event-time 统计 window 内事件的例例⼦子。 1 import spark.implicits._ 2 3 val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String } 4 5 // Group the data by window and word and compute the count of each group 6 val windowedCounts = words.groupBy( 7 window("eventTime", "10 minutes", "5 minutes"), 8 $"word" 9 ).count() 如下图所示，窗⼝口⼤大⼩小为 10 分钟，每 5 分钟 trigger ⼀一次。在 12:11 时候收到了了⼀一条 12:04 的数据， 也就是 late data （什什么叫 late data 呢？就是 Processing Time ⽐比 Event Time 要晚），然后去更更新 其对应的 Result Table 的记录。 

6 .watermark，是也为了了处理理 ，很多情况下对于这种 late data 的时效数据并没有必要⼀一直保留留太久。 ⽐比如说，数据晚了了 10 分钟或者还有点有，但是晚了了 1 个⼩小时就没有⽤用了了，另外这样设计还有⼀一个好 处就是中间状态没有必要维护那么多。watermark 的形式化定义为 max(eventTime) - threshold，早 于 watermark 的数据直接丢弃。 1 import spark.implicits._ 2 3 val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String } 4 5 // Group the data by window and word and compute the count of each group 6 val windowedCounts = words 7 .withWatermark("eventTime", "10 minutes") 8 .groupBy( 9 window("eventTime", "10 minutes", "5 minutes"), 10 $"word") 11 .count() ⽤用下图表示更更加形象。在 12:15 trigger 时 watermark 为 12:14 - 10m = 12:04，所以 late date (12:08, dog; 12:13, owl) 都被接收了了。在 12:20 trigger 时 watermark 为 12:21 - 10m = 12:11，所以 late data (12:04, donkey) 都丢弃了了。 

7 .除此之后 Structured Streaming 还提供了了⽤用户可以⾃自定义状态计算逻辑的算⼦子： mapGroupsWithState flatMapGroupsWithState 看名字⼤大概也能看出来 mapGroupsWithState 是 one -> one，flatMapGroupsWithState 是 one -> multi。这两个算⼦子的底层都是基于 Spark Streaming 的 updateStateByKey。 4. Continuous Processing Mode 好，终于要介绍到“真正”的流处理理了了，我之所以说“真正”是因为 continuous mode 是传统的流处理理模 式，通过运⾏行行⼀一个 long-running 的 operator ⽤用来处理理数据。之前 Spark 是基于 micro-batch 模式 的，就被很多⼈人诟病不不是“真正的”流式处理理。continuous mode 这种处理理模式只要⼀一有数据可⽤用就会 进⾏行行处理理，如下图所示。epoch 是 input 中数据被发送给 operator 处理理的最⼩小单位，在处理理过程 中，epoch 的 offset 会被记录到 wal 中。另外 continuous 模式下的 snapshot 存储使⽤用的⼀一致性算 法是 Chandy-Lamport 算法。 

8 .这种模式相⽐比与 micro-batch 模式缺点和优点都很明显。 缺点是不不容易易做扩展 优点是延迟更更低 关于为什什么延迟更更低，下⾯面两幅图可以做到⼀一⽬目了了然。 5. ⼀一 

9 .5. ⼀一 对于 Structured Streaming 来说，因为有两种模式，所以我们分开讨论。 micro-batch 模式可以提供 end-to-end 的 exactly-once 语义。原因是因为在 input 端和 output 端 都做了了很多⼯工作来进⾏行行保证，⽐比如 input 端 replayable + wal，output 端写⼊入幂等。 continuous mode 只能提供 at-least-once 语义。关于 continuous mode 的官⽅方讨论的实在太少， 甚⾄至只是提了了⼀一下。在和 @李李呈祥 讨论之后觉得应该还是 continuous mode 由于要尽可能保证低延 迟，所以在 sink 端没有做⼀一致性保证。 6. Benchmark Structured Streming 的官⽅方论⽂文⾥里里⾯面给出了了 Yahoo! Streaming Benchmark 的结果，Structured Streaming 的 throughput ⼤大概是 Flink 的 2 倍和 Kafka Streaming 的 90 多倍。 7. 总结⼀一下，Structured Streaming 通过提供⼀一套 high-level 的 declarative api 使得流式计算的编写相 ⽐比 Spark Streaming 简单容易易不不少，同时通过提供 end-to-end 的 exactly-once 语义 8. 最后，闲扯⼀一点别的。Spark 在 5 年年推出基于 micro-batch 模式的 Spark Streaming 必然是基于当时 Spark Engine 最快的⽅方式，尽管不不是真正的流处理理，但是在吞吐量量更更重要的年年代，还是尝尽了了甜头。 ⽽而 Spark 的真正基于 continuous 处理理模式的 Structured Streaming 直到 Spark 2.3 版本才真正推 出，从⽽而导致近两年年让 Flink 尝尽了了甜头（当然和 Flink 的优秀的语义模型存在很⼤大的关系）。在实时 计算领域，由 Spark 的卓越核⼼心 SQL Engine 助⼒力力的 Structured Streaming，还是⻛风头正劲的 Flink， 亦或是其他流处理理引擎，究竟谁将占领统治地位，还是值得期待⼀一下的。 Ps: 本⼈人本周⼆二在由阿⾥里里巴巴 EMR 团队主导的 Apache Spark 社群钉钉群做了了⼀一场 《从 Spark Streaming 到 Structured Streaming》的直播，直播 ppt 可以在 slidetalk ⽹网站上⾯面查看，地址：htt ps://www.slidestalk.com/s/FromSparkStreamingtoStructuredStreaming58639 ，Apache Spark 钉钉社群定期会有技术分享并可以观看直播回放，感兴趣的可以⼀一起来学习，钉钉群⼆二维码如下，使 ⽤用钉钉扫码即可加⼊入。 

10 .9. Reference 1. Zaharia M, Das T, Li H, et al. Discretized streams: Fault-tolerant streaming computation at scale[C]//Proceedings of the Twenty-Fourth ACM Symposium on Operating Systems Principles. ACM, 2013: 423-438. 2. Akidau T, Bradshaw R, Chambers C, et al. The dataflow model: a practical approach to balancing correctness, latency, and cost in massive-scale, unbounded, out-of-order data 

11 . processing[J]. Proceedings of the VLDB Endowment, 2015, 8(12): 1792-1803. 3. Armbrust M, Das T, Torres J, et al. Structured Streaming: A Declarative API for Real-Time Applications in Apache Spark[C]//Proceedings of the 2018 International Conference on Management of Data. ACM, 2018: 601-613. 4. The world beyond batch: Streaming 101 5. The world beyond batch: Streaming 102 6. Streaming Systems 7. https://databricks.com/blog/2018/03/20/low-latency-continuous-processing-mode-in-struc tured-streaming-in-apache-spark-2-3-0.html 8. https://en.wikipedia.org/wiki/Chandy-Lamport_algorithm 9. A Deep Dive Into Structured Streaming: https://databricks.com/session/a-deep-dive-into-st ructured-streaming 10. Continuous Applications: Evolving Streaming in Apache Spark 2.0: https://databricks.com/b log/2016/07/28/continuous-applications-evolving-streaming-in-apache-spark-2-0.html 11. Spark Structured Streaming:A new high-level API for streaming: https://databricks.com/blo g/2016/07/28/structured-streaming-in-apache-spark.html 12. Event-time Aggregation and Watermarking in Apache Spark’s Structured Streaming: https:/ /databricks.com/blog/2017/05/08/event-time-aggregation-watermarking-apache-sparks-str uctured-streaming.html 13. http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html 14. Benchmarking Structured Streaming on Databricks Runtime Against State-of-the-Art Streaming Systems: https://databricks.com/blog/2017/10/11/benchmarking-structured-str eaming-on-databricks-runtime-against-state-of-the-art-streaming-systems.html