Spark结构化流编程指南【基础信息】

Dear 丶 2023-02-12 02:25 163阅读 0赞

## 一.概述 ##

结构化流是基于Spark SQL引擎构建的可伸缩且容错的流处理引擎。可以像对静态数据进行批处理计算一样来表示流计算。当流数据继续到达时，Spark SQL引擎将负责逐步递增地运行它并更新最终结果。可以在Scala，Java，Python或R中使用Dataset / DataFrame API来表示流聚合，事件时间窗口，流到批处理联接等。计算在同一优化的Spark SQL引擎上执行。最后，系统通过检查点和预写日志来确保端到端的一次容错保证。简而言之，结构化流提供了快速，可扩展，容错，端到端的精确一次流处理，而用户无需推理流。

在内部，默认情况下，结构化流查询是使用微批量处理引擎处理的，该引擎将数据流作为一系列小批量作业处理，从而实现了低至100毫秒的端到端延迟以及一次精确的容错保证。但是，自Spark 2.3起，我们引入了一种称为“ 连续处理”的新低延迟处理模式，该模式可以实现一次最少保证的低至1毫秒的端到端延迟。在不更改查询中的Dataset / DataFrame操作的情况下，将能够根据应用程序需求选择模式。

## 二.快速案例 ##

1.获取SparkSession入口并导入转换

val spark = SparkSession
          .builder()
          .appName("从Socket获取数据进行结构化处理")
          .master("local[2]")
          .getOrCreate()
    
        // 导入转换
        import spark.implicits._

2.创建流处理框架

// split the lines into words
        val wordCounts = lines.as[String]
          .flatMap(_.split(" "))
          .groupBy("value")
          .count()

此linesDataFrame表示一个包含流文本数据的无界表。该表包含一列名为“值”的字符串，流文本数据中的每一行都成为表中的一行。请注意，由于我们正在设置转换，并且尚未开始转换，因此当前未接收到任何数据。接下来，我们使用将该DataFrame转换为String的Dataset .as\[String\]，以便我们可以应用该flatMap操作将每一行拆分为多个单词。最后，我们wordCounts通过对数据集中的唯一值进行分组并对其进行计数来定义DataFrame。请注意，这是一个流数据帧，它表示流的运行字数。  
3.开启执行  
设置outputMode(“complete”)为在每次更新计数时将完整的计数集（由指定）打印到控制台。然后使用开始流计算start()。

// start running program
        val result = wordCounts
          .writeStream
          .outputMode("complete")
          .format("console")
          .start()
        
        result.awaitTermination()

执行此代码后，流计算将在后台开始。result对象是该活动流查询的句柄，我们已决定使用来等待查询终止，awaitTermination()以防止在该查询处于活动状态时退出该过程。  
4.启动Socket  
启动linux系统，执行：nc -lk 9999，启动Socket  
![在这里插入图片描述][2020052412384860.png]  
5.输入数据  
![在这里插入图片描述][20200525180418188.png]  
6.执行结果

-------------------------------------------
    Batch: 0
    -------------------------------------------
    +-----+-----+
    |value|count|
    +-----+-----+
    +-----+-----+
    
    -------------------------------------------
    Batch: 1
    -------------------------------------------
    +-----+-----+
    |value|count|
    +-----+-----+
    |scala|    1|
    +-----+-----+
    
    -------------------------------------------
    Batch: 2
    -------------------------------------------
    +------+-----+
    | value|count|
    +------+-----+
    | scala|    1|
    | spark|    2|
    |  sale|    1|
    | flink|    1|
    |  scla|    1|
    |hadoop|    2|
    +------+-----+
    
    -------------------------------------------
    Batch: 3
    -------------------------------------------
    +------+-----+
    | value|count|
    +------+-----+
    | scala|    1|
    | spark|    3|
    |  sale|    1|
    | flink|    1|
    |  scla|    1|
    |hadoop|    2|
    +------+-----+

7.备注

*  若先启动程序则首次执行内容为空！
 *  程序执行的触发的时间差异较大，需耐心等待！

## 三.编程模型 ##

结构化流传输中的关键思想是将实时数据流视为被连续添加的表。这导致了一个新的流处理模型，该模型与批处理模型非常相似。将像在静态表上一样将流计算表示为类似于批处理的标准查询，Spark 在无界输入表上将其作为增量查询运行。  
1.基本概念  
将输入数据流视为“输入表”。流上到达的每个数据项都像是将新行附加到输入表中。  
图解如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]  
对输入的查询将生成“结果表”。在每个触发间隔（例如，每1秒钟），新行将附加到输入表中，并最终更新结果表。无论何时更新结果表，我们都希望将更改后的结果行写入外部接收器。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]  
“输出”定义为写到外部存储器的内容。可以在不同的模式下定义输出：

*  complete完整模式 -整个更新的结果表将被写入外部存储器。由存储连接器决定如何处理整个表的写入。
 *  append追加模式 -仅将自上次触发以来追加在结果表中的新行写入外部存储器。这仅适用于结果表中现有行预计不会更改的查询。
 *  update更新模式 -仅自上次触发以来在结果表中已更新的行将被写入外部存储（自Spark 2.1.1起可用）。请注意，这与完整模式的不同之处在于此模式仅输出自上次触发以来已更改的行。如果查询不包含聚合，它将等同于追加模式。

请注意，每种模式都适用于某些类型的查询。稍后将对此进行详细讨论。

为了说明此模型的用法，让我们在上面的快速示例的上下文中了解该模型。第一个linesDataFrame是输入表，最后一个wordCountsDataFrame是结果表。需要注意的是在流处理的查询lines数据帧生成wordCounts是完全一样的，因为它是一个静态的数据帧。但是，启动此查询后，Spark将不断检查套接字连接中是否有新数据。如果有新数据，Spark将运行一个“增量”查询，该查询将先前的运行计数与新数据结合起来以计算更新的计数，如下所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]  
请注意，结构化流不会实现整个表。它从流数据源读取最新的可用数据，对其进行增量处理以更新结果，然后丢弃该源数据。它仅保留更新结果所需的最小中间状态数据（例如，前面示例中的中间计数）。

此模型与许多其他流处理引擎明显不同。许多流系统要求用户自己维护运行中的聚合，因此必须考虑容错和数据一致性（至少一次，最多一次或恰好一次）。在此模型中，Spark负责在有新数据时更新结果表，从而使用户免于推理。  
2.处理事件时间和延迟数据  
事件时间是嵌入数据本身的时间。对于许多应用程序，可能需要在此事件时间进行操作。例如，如果要获取每分钟由IoT设备生成的事件数，则可能要使用生成数据的时间（即数据中的事件时间），而不是Spark收到的时间。此事件时间在此模型中非常自然地表示-设备中的每个事件都是表中的一行，而事件时间是该行中的列值。这允许基于窗口的聚合（例如，每分钟的事件数）只是事件时间列上的一种特殊的分组和聚合类型-每个时间窗口是一个组，每行可以属于多个窗口/组。

此外，此模型自然会根据事件时间处理比预期晚到达的数据。由于Spark正在更新结果表，因此它具有完全控制权，可以在有较晚数据时更新旧聚合，并可以清除旧聚合以限制中间状态数据的大小。从Spark 2.1开始支持水印功能，该功能允许用户指定最新数据的阈值，并允许引擎相应地清除旧状态。  
3.容错语义  
提供端到端的一次语义是结构化流设计背后的主要目标之一。为此，我们设计了结构化流源，接收器和执行引擎，以可靠地跟踪处理的确切进度，以便它可以通过重新启动和/或重新处理来处理任何类型的故障。假定每个流源都有偏移量（类似于Kafka偏移量或Kinesis序列号），以跟踪流中的读取位置。引擎使用检查点和预写日志来记录每个触发器中正在处理的数据的偏移范围。流接收器被设计为是幂等的，用于处理流程。结合使用可重播的源和幂等的接收器，结构化流可以确保在任何故障下端到端的一次精确语义 。

[2020052412384860.png]: https://img-blog.csdnimg.cn/2020052412384860.png
[20200525180418188.png]: https://img-blog.csdnimg.cn/20200525180418188.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200525182002757.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200525201646316.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200525203334164.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70