Spark结构化流之水印

r囧r小猫 2023-02-15 15:59 148阅读 0赞

## 一.处理后期数据和加水印 ##

现在考虑如果存在事件迟到了应用程序会发生什么。例如，应用程序在12:11接收在12:04（即事件时间）生成的单词。应用程序应使用12:04而不是12:11来更新窗口的旧计数12:00 - 12:10。这在基于窗口的分组中很自然地发生，结构化流可以长时间保持部分聚合的中间状态，以便后期数据可以正确更新旧窗口的聚合，如下所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]  
但是，要连续几天运行此查询，系统必须限制其累积的中间内存状态量。这意味着系统需要知道何时可以从内存中删除旧状态聚合，因为应用程序将不再接收该聚合的最新数据。为此，我们在Spark 2.1中引入了**水印**功能，该功能使引擎自动跟踪数据中的当前事件时间，并尝试相应地清除旧状态。可以通过指定事件时间列和有关事件时间期望数据延迟的阈值来定义查询的水印。对于在时间结束的特定窗口T，引擎将维持状态并允许后期数据更新状态，直到(max event time seen by the engine - late threshold > 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()

在此示例中，我们将在“时间戳”列的值上定义查询的水印，还将“ 10分钟”定义为允许数据晚到的阈值。如果此查询以“update output”模式运行，则引擎将在“结果表”中保持窗口的更新计数，直到该窗口晚于水印为止，该时间滞后于“列”中的当前事件时间。“时间戳”的时间缩短10分钟。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]  
如图所示，引擎跟踪的最大事件时间是 蓝色虚线，(max event time - ‘10 mins’) 在每次触发开始时设置的水印是红线。例如，当引擎观察到数据时 (12:14, dog)，它将下一个触发器的水印设置为12:04。此水印可让引擎再保持10分钟的中间状态，以便对较晚的数据进行计数。例如，数据(12:09, cat)不正确且延迟，并且落在窗口12:00 - 12:10和12:05 - 12:15。由于它仍在12:04触发器中的水印之前，因此引擎仍将中间计数保持为状态，并正确更新相关窗口的计数。但是，当水印更新为12:11，(12:00 - 12:10)清除了窗口的中间状态，并且所有后续数据（例如(12:04, donkey)）都被认为“为时已晚”，因此被忽略。请注意，在每次触发之后，更新的计数（即紫色行）将写入到接收器中，作为更新输出指示的触发输出。

某些接收器（例如文件）可能不支持更新模式所需的细粒度更新。为了与他们一起工作，我们还支持追加模式，其中仅将最终计数写入接收器。

请注意，withWatermark在非流数据集上使用是no-op。由于水印不应以任何方式影响任何批量查询，因此我们将直接忽略它。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 2]  
与之前的更新模式类似，引擎为每个窗口维护中间计数。但是，部分计数不会更新到结果表，也不会写入接收器。引擎等待“ 10分钟”来计算延迟日期，然后删除窗口<水印的中间状态，并将最终计数附加到结果表/接收器。例如，12:00 - 12:10仅在将水印更新为之后，窗口的最终计数才添加到结果表中12:11。

## 二.加水印以清除聚合状态的条件 ##

重要的是要注意，必须满足以下条件才能使用水印清除聚合查询中的状态（从Spark 2.1.1开始，将来可能会更改）。

*  输出模式必须为“追加”或“更新”。完整模式要求保留所有聚合数据，因此不能使用水印删除中间状态。
 *  聚合必须在“事件时间”列上或window“事件时间”列上有一个。
 *  withWatermark必须在与汇总中使用的时间戳列相同的列上调用。例如， df.withWatermark(“time”, “1 min”).groupBy(“time2”).count()在附加输出模式下无效，因为水印是在与聚合列不同的列上定义的。
 *  withWatermark必须在使用水印详细信息的聚合之前调用。例如，df.groupBy(“time”).count().withWatermark(“time”, “1 min”)在追加输出模式下无效。

## 三.带水印聚合的语义保证 ##

*  水印延迟（设置为withWatermark）为“ 2小时”，确保引擎永远不会丢弃任何少于2小时的数据。换句话说，任何在此之前处理过的最新数据比事件时间少2小时（以事件时间计）的数据都可以保证得到汇总。
 *  保证仅在一个方向上严格。延迟超过2小时的数据不能保证被删除；它可能会或可能不会聚合。数据延迟更多，引擎处理数据的可能性越小。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200606154041301.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/2020060615552039.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/20200606160321384.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70