八、Structured Streaming低延迟持续处理模式（Continuous Processing和micro-batch processing）

野性酷女 2024-04-18 16:20 42阅读 0赞

# 一、Continuous Processing基本介绍 #

## 1、介绍 ##

连续处理是Spark 2.3中引入的一种新的实验性流执行模式，**它支持低延迟(~1 ms)端到端，并保证at-least-once**。与默认的微批处理引擎相比，**默认的*micro-batch processing*可以保证exactly-once语义，**但最多只能实现约100ms的延迟。对于某些类型的查询(下面将讨论)，您可以选择在不修改应用程序逻辑(即不更改DataFrame/Dataset操作)的情况下以哪种模式执行它们。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

要在连续处理模式下运行受支持的查询，只需指定一个连续触发器，并将所需的检查点间隔作为参数。例如,

import org.apache.spark.sql.streaming.Trigger
    spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
      .option("subscribe", "topic1")
      .load()
      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
      .option("topic", "topic1")
      .trigger(Trigger.Continuous("1 second"))  // only change in query
      .start()

checkpoint 间隔为1秒意味着连续处理引擎将每秒记录查询的进度。 生成的checkpoint采用与微批处理引擎兼容的格式，因此可以使用任何触发器重新启动任何查询。 例如，假如查询支持微批处理和连续处理，那么实际上也可以用连续处理触发器去启动微批处理触发器，反之亦然。

请注意，无论何时切换到连续模式，都将获得至少一次的容错保证。

## 2、支持的查询 ##

（1）从Spark 2.4开始，连续处理模式仅支持以下类型的查询。

*  Operations：在连续模式下仅支持dataset/dataframe的类似于map的操作，即支持projection（select，map，flatMap，mapPartitions等）和selection（where，filter等）。
 *  除了聚合函数（因为尚不支持聚合），current\_timestamp（）和current\_date（）（使用时间的确定性计算具有挑战性）之外，支持所有SQL函数。

（2）Sources

*  Kafka Source：支持所有操作。
 *  Rate source：适合测试。只有连续模式支持的选项是numPartitions和rowsPerSecond。

（3）Sinks

*  Kafka sink：支持所有选项。
 *  Memory sink：适合调试。
 *  Console sink：适合调试。支持所有操作。请注意，控制台将打印你在连续触发器中指定的每个checkpoint间隔。

## 3、注意事项 ##

*  连续处理引擎启动多个长时间运行的任务，这些任务不断从源中读取数据，处理数据并连续写入接收器。 查询所需的任务数取决于查询可以并行从源读取的分区数。 因此，在开始连续处理查询之前，必须确保群集中有足够的核心并行执行所有任务。 例如，如果您正在读取具有10个分区的Kafka主题，则群集必须至少具有10个核心才能使查询正常执行。
 *  停止连续处理流可能会产生虚假的任务终止警告。 这些可以安全地忽略。
 *  目前没有自动重试失败的任务。 任何失败都将导致查询停止，并且需要从检查点手动重新启动。

# 二、低延迟场景 #

假设我们想建立一个实时流处理程序来标记信用卡欺诈交易。在理想的情况下，一旦犯罪分子刷了信用卡，我们就能立即对欺诈行为进行识别和阻止。但是，又不想让进行合法交易的用户感觉到延迟从而影响用户体验。所以流数据的端到端处理延迟需要有一个严格的上限。考虑到在数据传输过程中存在其他延误，流处理必须在10~20ms时间范围内实现对每笔交易的欺诈识别。

我们先尝试利用Structured Streaming构建这个流处理程序。假设我们有一个名为“isPaymentFlagged” 的函数可以用于判别欺诈交易。为了尽量减少延迟，我们将使用0秒的处理时间触发器，即对于每个微批数据Spark需要在获得时立即启动处理，从较高的层次来看，查询看起来像这样：

payments \
      .filter("isPaymentFlagged(paymentId)") \
      .writeStream \
       {...}
      .trigger(processingTime = "0 seconds") \
      .start()

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

可以看到Spark处理记录的时间超过100ms，虽然对于很多应用场景来说100ms已经足够好了，但是对于上述用例是无法满足需求的。那么，新出现的连续流处理能够解决问题吗？

payments \
      .filter("isPaymentFlagged(paymentId)") \
      .writeStream \
       {...}
      .trigger(continuous = "5 seconds") \
      .start

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2][]

# 三、基于微批（Micro-Batch）的流处理 #

Structured Streaming默认使用微批处理执行模型。 这意味着Spark流式计算引擎会定期检查流数据源，并对自上一批次结束后到达的新数据执行批量查询。 在高层次上，它看起来像这样。

![70][]

在这个体系结构中，Driver驱动程序通过将记录偏移量保存到预写日志中来对数据处理进度设置检查点，然后可以使用它来重新启动查询。 需要注意的是，为了获得确定性的重新执行（deterministic re-executions）和端到端语义，在处理下一个微批数据之前，要将该微批数据中的偏移范围保存到日志中。 所以，当前到达的数据需要等待当前的微批处理作业完成，且其中数据的偏移量范围被计入日志后，才能在下一个微批作业中得到处理。 在细粒度上，时间线看起来像这样。

![70 1][]

这会导致数据到达和得到处理并输出结果之间的延时超过100ms。

在之前，用户只能通过微批处理引擎构建结构化数据的流处理应用，从而可以从Spark SQL的优化工作中受益。在之前，我们能够在100ms的延时级别实现高吞吐量的流数据处理，在过去的几年里，在与数千开发人员和上百个使用案例合作的过程中，我们发现微批的数据延时对于大多数实际的流式工作负载（如ETL和监控）已经足够了。然而，一些场景确实需要更低的延时，所以我们设计并构建了连续处理模式。

# 四、连续处理（Continuous Processing） #

在连续处理模式中，Spark不再是启动周期性任务，而是启动一系列连续读取，处理和写入数据的长时间运行的任务。 在高层次上，设置和记录级时间线看起来像这些（与上述微量批处理执行图相对照）。

![70 2][]

![70 3][]

由于事件在到达时会被立即处理和写入结果，所以端到端延迟只有几毫秒。

此外，我们利用著名的Chandy-Lamport算法对查询进度设置检查点。 特殊标记的记录被注入到每个任务的输入数据流中; 我们将它们称为“时间代标记（epoch marker）”，并将它们之间的差距称为“时间代(epoch)”。当任务遇到标记时，任务异步报告处理后的最后偏移量给driver。 一旦driver程序接收到写入接收器的所有任务的偏移量，它就会将它们写入前述的预写日志。 由于检查点的设置是完全异步的，任务可以不间断地持续并提供一致的毫秒级延迟。

在Spark2.3.0中，流数据的连续处理模式还是一种实验性功能，在此模式下支持Structured Streaming所支持的所有流数据源以及DataFrame / Dataset / SQL操作的子集。

具体而言，你可以在满足以下条件的查询中设置可选的连续触发器：

1.从支持的数据源（例如Kafka）读取数据并写入支持的接收器（例如Kafka）、内存或者控制台（数据写入内存或控制台便于程序的debug）

2.仅使用map-like操作（例如：select, where, map, flatMap, filter等）；

3.除聚合函数外，还有任何SQL函数以及基于当前时间的函数，如current\_timestamp（）和current\_date（）。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/61c846573a304aa3af2ebeff781a1a97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/b0ecbcee576b4cc19451950c12647105.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/cd4d0383da2642d290aafd974d4c365b.png
[70]: https://img-blog.csdn.net/20180321185555387?watermark/2/text/Ly9ibG9nLmNzZG4ubmV0L3UwMTM0Njg5MTc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 1]: https://img-blog.csdn.net/20180321190537725?watermark/2/text/Ly9ibG9nLmNzZG4ubmV0L3UwMTM0Njg5MTc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 2]: https://img-blog.csdn.net/20180321191623274?watermark/2/text/Ly9ibG9nLmNzZG4ubmV0L3UwMTM0Njg5MTc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 3]: https://img-blog.csdn.net/20180321191634519?watermark/2/text/Ly9ibG9nLmNzZG4ubmV0L3UwMTM0Njg5MTc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70