Flink-时间窗口

本文介绍: 在流数据处理应用中，一个很重要、也很常见的操作就是窗口计算。所谓的“窗口”，一般就是划定的一段时间范围，也就是“时间窗”；对在这范围内的数据进行处理，就是所谓的窗口计算。所以窗口和时间往往是分不开的。

在流数据处理应用中，一个很重要、也很常见的操作就是窗口计算。所谓的“窗口”，一般就是划定的一段时间范围，也就是“时间窗”；对在这范围内的数据进行处理，就是所谓的窗口计算。所以窗口和时间往往是分不开的。

时间 语义

1、处理时间（Proce ss in g Time）

处理时间的概念非常简单，就是指执行处理操作的机器的系统时间。在这种时间语义下处理窗口非常简单粗暴，不需要各个节点之间进行协调同步，也不需要考虑数据在流中的位置，简单来说就是“我的地盘听我的”。所以处理时间是最简单的时间语义。

2、事件时间（Event Time）

事件时间，是指每个事件在对应的设备上发生的时间，也就是数据生成的时间。数据一旦产生，这个时间自然就确定了，所以它可以作为一个属性嵌入到数据中。这其实就是这条数据记录的“时间戳”（Times t amp）。

水位线

在事件时间语义下，我们不依赖系统时间，而是基于数据自带的时间戳去定义了一个时钟，用来表示当前时间的进展。于是每个并行子任务都会有一个自己的逻辑时钟，它的前进是靠数据的时间戳来驱动的。

我们可以把时钟也以数据的形式传递出去，告诉下游任务当前时间的进展；而且这个时钟的传递不会因为窗口聚合之类的运算而停滞。一种简单的想法是，在数据流中加入一个时钟标记，记录当前的事件时间；这个标记可以直接广播到下游，当下游任务收到这个标记，就可以更新自己的时钟了。由于类似于水流中用来做标志的记号，在 Flink 中，这种用来衡量事件时间（Event Time）进展的标记，就被称作“水位线”（Waterm ar k）。

具体实现上，水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。

水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据
水位线主要的内容是一个时间戳，用来表示当前事件时间的进展
水位线是基于数据的时间戳生成的
水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进
水位线可以通过设置延迟，来保证正确处理乱序数据
一个水位线 Waterm ar k(t)，表示在当前流中事件时间已经达到了时间戳 t, 这代表 t 之前的所有数据都到齐了，之后流中不会出现时间戳 t’ ≤ t 的数据

水位线生成 策略

在Flink 的 DataStream API 中，有一个单独用于生成水位线的方法： as signTimestamp sAndWaterm ar ks()，它主要用来为流中的数据分配时间戳，并生成水位线来指示事件时间。

有序流的水位线生成策略

object f1 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度
    env.setParallelism(1)
    //获取当前的运行配置
      //setAutoWatermarkInterval（时间戳）自动生成水位线的时间间隔
    env.getConfig.setAutoWatermarkInterval(500L)
    //数据
    val stream = env.fromElements(
      Event(4, "aa", 1000L),
      Event(5, "bb", 2000L),
      Event(6, "cc", 2500L),
      Event(7, "dd", 4000L)
    )
    //设置水位线
      //1、有序流的水位线生成策略
    stream.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[Event]()
    .withTimestampAssigner(
      new SerializableTimestampAssigner[Event] {
        override def extractTimestamp(t: Event, l: Long): Long = t.time //指定字段中的time为时间戳
      }
    ))

    //执行
    env.execute()
  }
  case class Event(id: Int, name: String, time: Long )
}