四、Structured Streaming的事件-时间（Event-time）和延迟数据（Late Data）

「爱情、让人受尽委屈。」 2024-04-18 15:09 38阅读 0赞

**目录**

1、基于事件-时间的窗口操作

2、延迟数据和水印操作

--------------------

Event-time是嵌入到数据本身的时间，所以首先 ***event-time 是一个基于事件的时间***。对于许多的应用来说，你可能希望操作这个事件-时间。例如，如果你想获得每分钟物联网设备产生的事件数量，然后想使用数据产生时的时间（也就是数据的event-time），而不是Spark接收他们的时间。每个设备中的事件是表中的一行，而事件-时间是行中的一个列值。这就允许将基于窗口的聚合（比如每分钟的事件数）看成是事件-时间列的分组和聚合的特殊类型——每个时间窗口是一个组，每行可以属于多个窗口/组。

而且，这个模型自然处理那些比期望延迟到达的事件-时间数据。当Spark正在更新结果表时，当有延迟数据，它就会完全控制更新旧的聚合，而且清理旧的聚合去限制中间状态数据的大小。从Spark 2.1开始，我们已经开始支持水印（watermarking ），它允许用户确定延迟的阈值，允许引擎相应地删除旧的状态。

## 1、基于事件-时间的窗口操作 ##

在滑动的事件-时间窗口上的聚合对于结构化流是简单的，非常类似于分组聚合。在分组聚合中，聚合的值对用户确定分组的列保持唯一的。在基于窗口的聚合中，聚合的值对每个窗口的事件-时间保持唯一的。

修改我们前面的单词计数的例子，现在当产生一行句子时，附件一个时间戳。我们想每5分钟统计一次10分钟内的单词数。例如，12:00 - 12:10, 12:05 - 12:15, 12:10 - 12:20等。注意到12:00 - 12:10是一个窗口，表示数据12:00之后12:10之前到达。比如12:07到达的单词，这个单词应该在12:00 - 12:10和12:05 - 12:15两个窗口中都要被统计。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

基于事件-时间窗口的单词统计代码：

object WindowOnEventTime {
      case class TimeWord(word:String,timestamp:Timestamp)
      def main(args: Array[String]): Unit = {
        val window_size = 10//args(0)    //窗口长度
        //窗口滑动距离应当小于或等于窗口长度
        val slide_size = 5//if (args.length == 2) args(1).toInt else window_size
        if (slide_size > window_size) throw  new Exception("===wrong paragrams===")
        val windowDuration = s"$window_size seconds"
        val slideDuration = s"$slide_size seconds"
    
        Logger.getLogger("org").setLevel(Level.ERROR)
        val sqLContext = SparkSession.builder().appName(" HelloApp").master("local[*]")
          .getOrCreate()
        import sqLContext.implicits._
        val df = sqLContext.readStream.format("socket")
          .option("host","192.168.199.135")
          .option("port",9999)
          .option("includeTimestamp",true)//添加时间戳
          .load()//nc -lk 9999
        val words = df.as[(String,Timestamp)]
          .flatMap(line=>line._1.split(" ").map(word=> TimeWord(word,line._2))).toDF()
        // 对 "value" 列做 count，得到多行二列的 Dataset/DataFrame；即 result table
        import org.apache.spark.sql.functions._
        val wordCounts = words
          .groupBy(window($"timestamp",windowDuration,slideDuration),$"word")
          .count().orderBy("window")
        val query = wordCounts.writeStream.outputMode("complete")
          .format("console")
          .option("truncate","false")
          .start()
        query.awaitTermination()
      }
    }

-------------------------------------------
    Batch: 1
    -------------------------------------------
    +------------------------------------------+-----+-----+
    |window                                    |word |count|
    +------------------------------------------+-----+-----+
    |[1970-01-19 11:32:15, 1970-01-19 11:32:25]|cat  |1    |
    |[1970-01-19 11:32:15, 1970-01-19 11:32:25]|love |1    |
    |[1970-01-19 11:32:15, 1970-01-19 11:32:25]|china|1    |
    |[1970-01-19 11:32:15, 1970-01-19 11:32:25]|a    |1    |
    |[1970-01-19 11:32:15, 1970-01-19 11:32:25]|have |1    |
    |[1970-01-19 11:32:15, 1970-01-19 11:32:25]|i    |2    |
    |[1970-01-19 11:32:20, 1970-01-19 11:32:30]|love |1    |
    |[1970-01-19 11:32:20, 1970-01-19 11:32:30]|i    |2    |
    |[1970-01-19 11:32:20, 1970-01-19 11:32:30]|have |1    |
    |[1970-01-19 11:32:20, 1970-01-19 11:32:30]|a    |1    |
    |[1970-01-19 11:32:20, 1970-01-19 11:32:30]|cat  |1    |
    |[1970-01-19 11:32:20, 1970-01-19 11:32:30]|china|1    |
    +------------------------------------------+-----+-----+

从结果中可以看到，有三个窗口：\[15， 25\]，\[20， 30\] 窗口长度10秒，滑动距离5s，也就是说每隔5秒统计最近10秒内的单词数。

## 2、延迟数据和水印操作 ##

**（1）基本介绍**

现在考虑如果一个事件延迟到达这个应用将会发生什么。例如，12:04产生了一个word，可能在12:11才能被应用接收到。应用应该使用12:04这个时间去更新窗口12:00-12:10中的单词计数，而不是12:11。这在基于窗口的分组中自然发生——结构化流可以维持部分聚合的中间状态很长一段时间以满足延迟数据来正确更新旧窗口中的聚合。如下所示

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

延迟数据dog，在12:11才被应用接受到，事实上，它在12:04已经产生。在前两次结果表中都未被统计，但是统计在了最后一次结果表中。这次在统计中，Spark引擎一致维持中间数据状态，直到延迟数据到达，并统计到结果表中。

对于一个运行好几天的查询来说，有必要绑定累计中间内存状态的数量。这意味着系统需要知道什么时候可以从内存状态中删除旧聚合，由于应用对那个聚合不再接收延迟数据。在Spark 2.1中，引入了水印，它使引擎自动跟踪当前数据中的事件时间，尝试一致地删除旧的状态。通过确定事件时间列和按照事件时间数据预期延迟的阈值可以定义查询的水印。对于一个以时间T开始的特定窗口，引擎将会维持状态，并且允许延迟数据更新状态直到(max event time seen by the engine - late threshold > T，换句话说，阈值内的延迟数据将会倍聚合，但是比阈值更延迟的数据将被删除。让我们来看个例子，late threshold=10分钟，输出模式为Update Mode。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2][]

**黄色实心圆：**准时到达应用的数据，例如第一个dog单词，12:07产生，12:07到达（产生和到达可能相差一些秒）。

**红色实心圆：**延迟到达应用的数据，例如12:09产生的cat单词，12:09+10分钟=12:19，实际到达时间小于12:15，在水印之内。

**红色空心圆：**延迟到达应用的数据，例如donkey这个词在12:04产生，但在12:04+10分钟=12:14之后到达应用，在水印之外

**（2）使用方法**

// 计数（添加水印 阈值是）
        val late =windowDuration
        val windowedCounts = words.withWatermark("timestamp", late)
          .groupBy(window($"timestamp", windowDuration, slideDuration), $"word")
          .count().orderBy("window")

**在非流数据集上使用watermark是无效的。**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/674e160fe81c40eea40029770b40a6e0.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/261492eab35e49d6a52d3cc545b2bfb7.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/538251d33cef49098d6613ff573c48ab.png