Flink 窗口 window

一、基本概念

1.窗口分类

　　TimeWindow：按照时间生成 Window。对于 TimeWindow，可以根据窗口实现原理的不同分成三类：滚动窗口（TumblingWindow）、滑动窗口（Sliding Window）和会话窗口（Session Window）。

　　CountWindow：按照指定的数据条数生成一个 Window，与时间无关。

2.时间分类

　　Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink 通过时间戳分配器访问事件时间戳。

　　Ingestion Time：是数据进入 Flink 的时间。

　　Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是 Processing Time。

3.水位Watermark

　　由于事件产生的时间，和到达Flink的时间并不是完全有序的，可能先发生的时间却后达到Flink。因此需要需要设定一个延迟时间t，窗口不是到达长度之后就触发计算，而是到达长度+延迟t之后才触发计算。而水位watermark就是窗口当前数据的时间戳减去延迟时间，表示小于watermark的数据都已经到达了（不含watermark）。

二、案例演示

案例1：按Processing Time划分滚动时间窗口

　　TimeWindow 是将指定时间范围内的所有数据组成一个 window，一次对一个window 里面的所有数据进行计算。Flink 默认的时间窗口根据 Processing Time 进行窗口的划分，将 Flink 获取到的数据根据进入 Flink 的时间划分到不同的窗口中。时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。

　　在下面的代码中，因为没有指定时间类型，所以采用了默认的Processing Time，即Flink实际计算数据的时间， 通过.timeWindow(Time.seconds(10))设定窗口的大小为10秒，当一条数据进来后开始计时，10秒之后输出这个窗口中所有数据的计算结果。

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object WindowTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val socketStream = env.socketTextStream("hadoop102",7777)

    val dataStream: DataStream[SensorReading] = socketStream.map(d => {
      val arr = d.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).toDouble)
    })

    //统计10秒内的最小温度
    val minTemperatureStream = dataStream.map(data=>(data.id,data.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(10)) //10秒滚动窗口，不指定时间特性，默认为ProcessingTime
      .reduce((data1, data2)=>(data1._1,data1._2.min(data2._2)))

    //打印原始的dataStream
    dataStream.print("data stream")

    //打印窗口数据流
    minTemperatureStream.print("min temperature")

    env.execute("window test")

  }

}

　　测试：　　

　　连续输入两条数据

[atguigu@hadoop102 ~]$ nc -lk 7777
sensor_1, 1547718200, 30.8
sensor_1, 1547718201, 40.8

　　在一个10秒的滚动窗口内，窗口流minTemperatureStream 只输出了一条数据。此时触发TimeWindow去计算的时机就是第一条数据来的10秒过后。

data stream> SensorReading(sensor_1,1547718200,30.8)
data stream> SensorReading(sensor_1,1547718201,40.8)
min temperature> (sensor_1,30.8)

案例2：按EventTime划分带水位的滚动时间窗口

　　知识点：

　　①通过env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)设定窗口的时间特性为事件时间。

　　②在assignTimestampsAndWatermarks()方法中，传递一个BoundedOutOfOrdernessTimestampExtractor类实现对象，构造器参数就是容忍的延迟时间，实现方法，指明时间戳用哪个字段。

　　③env.getConfig.setAutoWatermarkInterval(300) //周期性的生成watermark：系统会周期性的将watermark插入到流中(水位线也是一种特殊的事件!)。默认周期是200毫秒。

　　④如果在keyBy之前设置水位线，则所有分区共用一个watermark。但是输出还是根据keyBy的结果分开输出。

object WindowTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val socketStream = env.socketTextStream("hadoop102",7777)

    val dataStream: DataStream[SensorReading] = socketStream
      .map(d => {
        val arr = d.split(",")
        SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).toDouble)
      })
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(2)) {
        override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
      })

      //.assignAscendingTimestamps(_.timestamp) //升序数据添指定时间戳

    //统计5秒内的最小温度
    val minTemperatureStream = dataStream.map(data=>(data.id,data.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5)) //5秒滚动窗口
      .reduce((data1, data2)=>(data1._1,data1._2.min(data2._2)))

    //打印原始的dataStream
    dataStream.print("data stream")

    //打印窗口数据流
    minTemperatureStream.print("min temperature")

    env.execute("window test")

  }

}

　　测试：

　　sockt输入数据如下

[atguigu@hadoop102 ~]$ nc -lk 7777
sensor_1, 1547718199, 29
sensor_1, 1547718200, 30
sensor_1, 1547718201, 31
sensor_1, 1547718202, 32
sensor_1, 1547718203, 33

　　控制台打印如下：

data stream> SensorReading(sensor_1,1547718199,29.0)
data stream> SensorReading(sensor_1,1547718200,30.0)
data stream> SensorReading(sensor_1,1547718201,31.0)
data stream> SensorReading(sensor_1,1547718202,32.0)
min temperature> (sensor_1,29.0)
data stream> SensorReading(sensor_1,1547718203,33.0)

　　滚动窗口的第一个窗口的起始时间如何确定？

//滚动窗口，添加初始窗口的源码
public Collection<TimeWindow> assignWindows(Object element, long timestamp, WindowAssignerContext context) {
    if (timestamp > -9223372036854775808L) {
        long start = TimeWindow.getWindowStartWithOffset(timestamp, this.offset, this.size);
        return Collections.singletonList(new TimeWindow(start, start + this.size));
    } else {
        throw new RuntimeException("Record has Long.MIN_VALUE timestamp (= no timestamp marker). Is the time characteristic set to 'ProcessingTime', or did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?");
    }
}

//计算窗口起始点，时区偏移offset默认为0。windowSize实际上是滑动长度。加一下再取模也没有意义。

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) { return timestamp - (timestamp - offset + windowSize) % windowSize; }

　　对于我们的测试数据，窗口起始点start = 1547718199 - (1547718199 - 0 + 5) % 5 = 1547718195，所以第一个窗口是1547718195到1547718200，如果没有水位线设置，会在接收到数据时间戳为1547718200时关闭窗口，但是由于设定了延迟2秒，所以当接收到时间戳为1547718202的数据时才打印输出。

　　也可以自定义watermark从事件数据中抽取时间戳。方式一，周期式，继承AssignerWithPeriodicWatermarks，默认200毫秒抽取一次。方式二，间断式，继承AssignerWithPunctuatedWatermarks，根据需要自定义筛选条件。周期式举例：

class MyAssigner extends AssignerWithPeriodicWatermarks[SensorReading]{

  // 1分钟延迟
  val bound = 60 * 1000

  // 记录数据的最大时间戳
  var maxTs = Long.MinValue

  // 水位线等于最大时间戳减延迟
  override def getCurrentWatermark: Watermark = new Watermark(maxTs - bound)

  override def extractTimestamp(t: SensorReading, l: Long): Long = {

    //更新最大的时间戳
    maxTs = maxTs.max(t.timestamp * 1000)

    //时间戳单位毫秒
    t.timestamp * 1000
  }
}

　　

案例3：滑动时间窗口

　　滑动窗口和滚动窗口特性类似，滚动窗口可以看作一种特殊的滑动窗口，其窗口长度与滑动长度一样。在.timeWindow(Time.seconds(10),Time.seconds(5)) 方法中，设定了窗口的长度为10，滑动长度为5。窗口长度决定了窗口计算的数据的范围有多大，而滑动长度决定了窗口计算并关闭的时机。

def main(args: Array[String]): Unit = {
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  env.setParallelism(1)
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

  val socketStream = env.socketTextStream("hadoop102",7777)

  val dataStream: DataStream[SensorReading] = socketStream
    .map(d => {
      val arr = d.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).toDouble)
    })
    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(0)) {//简单起见，去掉水位设置
      override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
    })

  //统计15秒内的最小温度，5秒输出一次
  val minTemperatureStream = dataStream.map(data=>(data.id,data.temperature))
    .keyBy(_._1)
    .timeWindow(Time.seconds(15), Time.seconds(5)) //滑动窗口
    .reduce((data1, data2)=>(data1._1,data1._2.min(data2._2)))

  //打印原始的dataStream
  dataStream.print("data stream")

  //打印窗口数据流
  minTemperatureStream.print("min temperature")

  env.execute("window test")

}

　　socket数据

[atguigu@hadoop102 ~]$ nc -lk 7777
sensor_1, 1547718199, 29      
sensor_1, 1547718200, 30
sensor_1, 1547718201, 31

　　控制台打印

data stream> SensorReading(sensor_1,1547718199,29.0)
data stream> SensorReading(sensor_1,1547718200,30.0)
min temperature> (sensor_1,29.0)
data stream> SensorReading(sensor_1,1547718201,31.0)

　　滑动窗口的第一个窗口如何确定？与滚动窗口不同，由于滑动长度与窗口长度不一样，所以会设置多个初始窗口。

public Collection<TimeWindow> assignWindows(Object element, long timestamp, WindowAssignerContext context) {
    if (timestamp <= -9223372036854775808L) {
        throw new RuntimeException("Record has Long.MIN_VALUE timestamp (= no timestamp marker). Is the time characteristic set to 'ProcessingTime', or did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?");
    } else {
        List<TimeWindow> windows = new ArrayList((int)(this.size / this.slide));
        long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, this.offset, this.slide);

        for(long start = lastStart; start > timestamp - this.size; start -= this.slide) {
            windows.add(new TimeWindow(start, start + this.size));
        }

        return windows;
    }
}

　　对于这个案例，lastStart = 1547718199 - (1547718199 - 0 + 5) % 5 = 1547718195，只要start > 1547718199 -15 = 1547718184就会一直循环添加窗口，在第一次for循环中，添加第一个窗口1547718195到1547718210，第二次添加1547718190到1547718205,第三次添加1547718185到1547718200。生成的窗口数为Math.ceil(15.0 / 5.0)，所以在输入第二条数据，时间戳为1547718200，就触发了第三个窗口关闭。