Flink 窗口算子 -6-8

窗口算子

• 窗口算子提供了一种基于有限大小的桶对事件进行分组，并对这些桶中有限大小的数据进行计算



定义窗口算子

• 用于键值分区窗口的算子可以并行计算

• 用于非键值分区窗口只能单线程计算



新建窗口需要2个窗口组件

• 窗口分配器：用于决定输入流中元素该如何划分窗口的分配器，对于键值分区窗口会产生WindowedStream,非键值分区则是AllWindowedStream

• 窗口函数：一个用于处理分配到窗口中元素的窗口函数



   val minTempPerWindow: DataStream[(String, Double)] = sensorData
      .map(r => (r.id, r.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))//窗口分配器
      .reduce((r1, r2) => (r1._1, r1._2.min(r2._2)))//窗口函数



内置窗口分配器

• 时间窗口和基于数量的窗口

• 基于数量的窗口会按照元素到达窗口的顺序以固定数量进行分组，不确定性，需要自定义触发器丢弃不完整或过期数据。



• 每个时间窗口都有一个开始时间戳和一个结束时间戳

• 所有内置窗口分配器，都提供一个默认触发器，一旦事件超过窗口结束时间，就会触发窗口计算。

• 窗口会随着系统首次为其分配元素而窗口

• Flink用于不会对空窗口执行计算



• 内置窗口类型为：TimeWindow

• 两个时间戳的时间区间左闭右开



滚动时间窗口

• TumblingEventTimeWindows

• TumblingProcessingTimeWindows

• 只接受一个参数：以时间单元表示窗口大小

• 使用of（Time size）方法指定时间大小

• timeWindow(Time size)简写



  val avgTempPerWindow: DataStream[(String, Double)] = sensorData
      .map(r => (r.id, r.temperature))
      .keyBy(_._1)
      .timeWindow(Time.seconds(15))
      .aggregate(new AvgTempFunction)

滑动时间窗口

• 滑动间隔小会出现元素重叠

• 滑动间隔大，会出现丢失元素

会话窗口

• EventTimeSessionWindows.withGap(Time size)

• ProcessingTimeSessionWindows.withGap(Time size)