单词在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的计数。

基于事件时间窗口统计有两个参数索引：分组键（如单词）和窗口（事件时间字段）。

event-time 窗口生成

Structured Streaming中如何依据EventTime事件时间生成窗口的呢？查看类TimeWindowing源码中生成窗口规则：

org.apache.spark.sql.catalyst.analysis.TimeWindowing



// 窗口个数

/* 最大的窗口数 = 向上取整(窗口长度/滑动步长) */

maxNumOverlapping <- ceil(windowDuration / slideDuration)

for (i <- 0 until maxNumOverlapping)

   /**

      timestamp是event-time 传进的时间戳

      startTime是window窗口参数，默认是0 second 从时间的0s

      含义：event-time从1970年...有多少个滑动步长，如果说浮点数会向上取整

   */

   windowId <- ceil((timestamp - startTime) / slideDuration)

   /**

      windowId * slideDuration  向上取能整除滑动步长的时间

      (i - maxNumOverlapping) * slideDuration 每一个窗口开始时间相差一个步长

    */

   windowStart <- windowId * slideDuration + (i - maxNumOverlapping) * slideDuration + startTime

   windowEnd <- windowStart + windowDuration

   return windowStart, windowEnd

将【(event-time向上取 能整除 滑动步长的时间) - (最大窗口数×滑动步长)】作为"初始窗口"的开始时间，然后按照窗口滑动宽度逐渐向时间轴前方推进，直到某个窗口不再包含该event-time 为止，最终以"初始窗口"与"结束窗口"之间的若干个窗口作为最终生成的 event-time 的时间窗口。

每个窗口的起始时间start与结束时间end都是前闭后开（左闭右开）的区间，因此初始窗口和结束窗口都不会包含 event-time，最终不会被使用。假设数据为【2019-08-14 10:50:00, dog】，按照上述规则计算窗口示意图如下：

得到窗口如下:

延迟数据处理

Structed Streaming与Spark Streaming相比一大特性就是支持基于数据中的时间戳的数据处理。也就是在处理数据时，可以对记录中的eventTime事件时间字段进行考虑。因为eventTime更好的代表数据本身的信息，且可以借助eventTime处理比预期晚到达的数据，但是需要有一个限度(阈值)，不能一直等，应该要设定最多等多久。

延迟数据

在很多流计算系统中，数据延迟到达（the events arrives late to the application）的情况很常见，并且很多时候是不可控的，因为很多时候是外围系统自身问题造成的。Structured Streaming可以保证一条旧的数据进入到流上时，依然可以基于这些“迟到”的数据重新计算并更新计算结果。

上图中在12:04（即事件时间）生成的单词可能在12:11被应用程序接收，此时，应用程序应使用时间12:04而不是12:11更新窗口12:00-12:10的旧计数。但是会出现如下两个问题：

问题一：延迟数据计算是否有价值

如果某些数据，延迟很长时间（如30分钟）才到达流式处理系统，数据还需要再次计算吗？计算的结果还有价值吗？原因在于流式处理系统处理数据关键核心在于实时性；
实践表明，流计算关注的是近期数据，更新一个很早之前的状态往往已经不再具有很大的业务价值；

问题二：以前状态保存浪费资源

实时统计来说，如果保存很久以前的数据状态，很多时候没有作用的，反而浪费大量资源；
Spark 2.1引入的watermarking允许用户指定延迟数据的阈值，也允许引擎清除掉旧的状态。即根据watermark机制来设置和判断消息的有效性，如可以获取消息本身的时间戳，然后根据该时间戳来判断消息的到达是否延迟（乱序）以及延迟的时间是否在容忍的范围内（延迟的数据是否处理）。

Watermarking 水位

水位watermarking官方定义：


lets the engine automatically track the current event time in the data and attempt to clean up old state accordingly.

通过指定event-time列（上一批次数据中EventTime最大值）和预估事件的延迟时间上限（Threshold）来定义一个查询的水位线watermark。翻译：让Spark SQL引擎自动追踪数据中当前事件时间EventTime，依据规则清除旧的状态数据。

Watermark = MaxEventTime - Threshod

1：执行第一批次数据时，Watermarker为0，所以此批次中所有数据都参与计算；

2：Watermarker值只能逐渐增加，不能减少；

3：Watermark机制主要解决处理聚合延迟数据和减少内存中维护的聚合状态；

4：设置Watermark以后，输出模式OutputMode只能是Append和Update；

如下方式设置阈值Threshold，计算每批次数据执行时的水位Watermark：

看一下官方案例：词频统计WordCount，设置阈值Threshold为10分钟，每5分钟触发执行一次。

延迟到达但没有超过watermark：(12:08, dog)

在12:20触发执行窗口（12:10-12:20）数据中，(12:08, dog) 数据是延迟数据，阈值Threshold设定为10分钟，此时水位线【Watermark = 12:14 - 10m = 12:04】，因为12:14是上个窗口（12:05-12:15）中接收到的最大的事件时间，代表目标系统最后时刻的状态，由于12:08在12:04之后，因此被视为“虽然迟到但尚且可以接收”的数据而被更新到了结果表中，也就是(12:00 - 12:10, dog, 2)和(12:05 - 12:11, dog, 3)。

超出watermark：(12:04, donkey)

在12:25触发执行窗口（12:15-12:25）数据中，(12:04, donkey)数据是延迟数据，上个窗口中接收到最大的事件时间为12:21，此时水位线【Watermark = 12:21 - 10m = 12:11】，而(12:04, donkey)比这个值还要早，说明它”太旧了”，所以不会被更新到结果表中了。

设置水位线Watermark以后，不同输出模式OutputMode，结果输出不一样：

Update模式：总是倾向于“尽可能早”的将处理结果更新到sink，当出现迟到数据时，早期的某个计算结果将会被更新；
Append模式：推迟计算结果的输出到一个相对较晚的时刻，确保结果是稳定的，不会再被更新，比如：12:00 - 12:10窗口的处理结果会等到watermark更新到12：11之后才会写入到sink。

如果用于接收处理结果的sink不支持更新操作，则只能选择Append模式。

官方案例演示

编写代码，演示官方案例，如下几点注意：

1、该outputMode为update模式，即只会输出那些有更新的数据！！

2、官网案例该开窗窗口长度为10 min，步长5 min，水印为eventtime-10 min，但是测试的时候用秒

3、官网案例trigger(Trigger.ProcessingTime("5 minutes"))，但是测试的时候用秒

测试数据：

2019-10-10 12:00:07,dog

2019-10-10 12:00:08,owl



2019-10-10 12:00:14,dog

2019-10-10 12:00:09,cat



2019-10-10 12:00:15,cat

2019-10-10 12:00:08,dog

2019-10-10 12:00:13,owl

2019-10-10 12:00:21,owl



2019-10-10 12:00:04,donkey  --丢失

2019-10-10 12:00:17,owl     --不丢失

具体案例代码如下：

package cn.itcast.structedstreaming

import java.sql.Timestamp

import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkContext
import org.apache.spark.sql.streaming.OutputMode, StreamingQuery, Trigger
import org.apache.spark.sql.DataFrame, SparkSession

/**
 * 基于Structured Streaming 读取TCP Socket读取数据，事件时间窗口统计词频，将结果打印到控制台
 * 每5秒钟统计最近10秒内的数据（词频：WordCount)，设置水位Watermark时间为10秒
 * 2019-10-10 12:00:07,dog
 * 2019-10-10 12:00:08,owl
 *
 * 2019-10-10 12:00:14,dog
 * 2019-10-10 12:00:09,cat
 *
 * 2019-10-10 12:00:15,cat
 * 2019-10-10 12:00:08,dog
 * 2019-10-10 12:00:13,owl
 * 2019-10-10 12:00:21,owl
 *
 * 2019-10-10 12:00:04,donkey  --丢失
 * 2019-10-10 12:00:17,owl     --不丢失
 */
object StructuredWindow 
  def main(args: Array[String]): Unit = 
    // 1. 构建SparkSession实例对象，传递sparkConf参数
    val spark: SparkSession = SparkSession.builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[*]")
      .config("spark.sql.shuffle.partitions", "3")
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    import org.apache.spark.sql.functions._
    import spark.implicits._

    // 2. 使用SparkSession从TCP Socket读取流式数据
    val inputStreamDF: DataFrame = spark.readStream
      .format("socket")
      .option("host", "node1")
      .option("port", 9999)
      .load()

    // 3. 针对获取流式DStream进行词频统计
    val resultStreamDF = inputStreamDF
      .as[String]
      .filter(StringUtils.isNotBlank(_))
      // 将每行数据进行分割单词: 2019-10-12 09:00:02,cat dog
      .flatMap(line => 
        val arr = line.trim.split(",")
        val timestampStr: String = arr(0)
        val wordsStr: String = arr(1)
        wordsStr
          .split("\\\\s+")
          //(时间戳,单词)
          .map((Timestamp.valueOf(timestampStr), _))
      )
      // 设置列的名称
      .toDF("timestamp", "word")
      // TODO：设置水位Watermark
      .withWatermark("timestamp", "10 seconds")
      // TODO：设置基于事件时间（event time）窗口 -> time, 每5秒统计最近10秒内数据
      .groupBy(
        window($"timestamp", "10 seconds", "5 seconds"),
        $"word"
      ).count()
      // 按照窗口字段降序排序
      //.orderBy($"window")

    /*
        root
         |-- window: struct (nullable = true)
         |    |-- start: timestamp (nullable = true)
         |    |-- end: timestamp (nullable = true)
         |-- word: string (nullable = true)
         |-- count: long (nullable = false)
     */
    //resultStreamDF.printSchema()

    // 4. 将计算的结果输出，打印到控制台
    val query: StreamingQuery = resultStreamDF.writeStream
      .outputMode(OutputMode.Update())
      .format("console")
      .option("numRows", "100")
      .option("truncate", "false")
      .trigger(Trigger.ProcessingTime("5 seconds"))
      .start()
    query.awaitTermination()
    query.stop()

以上是关于2021年大数据Spark（五十二）：Structured Streaming 事件时间窗口分析的主要内容，如果未能解决你的问题，请参考以下文章

2021年大数据常用语言Scala（二十二）：函数式编程映射 map

2021年大数据常用语言Scala（三十二）：scala高级用法样例类

Spark（五十二）：Spark Scheduler模块之DAGScheduler流程

2021年大数据Flink（四十二）：BroadcastState

2021年大数据Flink（三十二）：Table与SQL案例准备 API

2021年大数据ELK（二十二）：采集Apache Web服务器日志