Flink_03_Window(个人总结)

    声明: 1. 本文为我的个人复习总结, 并非那种从零基础开始普及知识 内容详细全面, 言辞官方的文章
              2. 由于是个人总结, 所以用最精简的话语来写文章
              3. 若有错误不当之处, 请指出
keyBy不仅是为了分组, 同时还是为了能把数据分布到不同分区进行并行计算

所以开窗前最好先.keyBy

如果没keyBy, 那么调用的就是windowAll

TimeWindow:

1. 滚动窗口

   窗口大小 = 步长

   .timeWindow(Time.seconds(15))
       
   // 或是 在时间语义为事件时间时
   .window(TumblingEventTimeWindows.of(Time.seconds(15)))
   

2. 滑动窗口

   窗口大小 ≥ 步长, 会有窗口重叠

   .timeWindow(Time.seconds(15),Time.seconds(5))
       
   // 或是 在时间语义为事件时间时
   .window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(5)))
   

3. 会话窗口

   超过一段时间(session时间范围内)没有接收到新数据就会生成新的窗口

   // 或是 在时间语义为事件时间时  
   .window(EventTimeSessionWindows.withGap(Time.seconds(30)))
   

并不是以最小数据的到来时间作为窗口的起点, 而是由一个计算公式:
TimeWindow类中:

 public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
     // offset默认为0, windowSize%windowSize=0, 
     // 故等价为timetamp-timestamp%windowSize, 即以是windowSize的整数倍 & 最接近而且小于等于 最小数据的时间 作为窗口起点
	// offset一般是用来调时区的, 可由SlidingEventTimeWindows of(Time size, Time slide, Time offset)来设置
     return timestamp - (timestamp - offset + windowSize) % windowSize;
 }

CountWindow:

1. 滚动窗口

   .countWindow(15)
   

2. 滑动窗口

   .countWindow(15,5)
   

增量聚合函数:

每来一条数据, 就进行计算(提前计算, 预聚合)

1. .reduce(new ReduceFunction( )…), 只需要实现reduce方法即可

2. .aggregate(AggregateFunction<T, ACC, R> function), AggregateFunction麻烦些, 要自己实现好多方法

单独用的时候, 延时低, 但是计算次数太多伤性能

对于调用窗口函数进行聚合时, 最好先调用.aggregate 进行预聚合, 如:

• .aggregate(AggregateFunction<IN, ACC, OUT>aggFunction,ProcessWindowFunction<IN, OUT, KEY, W> windowFunction)

• .aggregate(AggregateFunction<IN, ACC, OUT>aggFunction,WindowFunction<IN, OUT, KEY, W>windowFunction)

  第一个参数的输出, 是第二个函数的输入

ProcessWindowFunction的 process方法中Iterable<IN> elements参数, 迭代器里只有一个元素

全窗口函数:

数据都到齐了后, 再进行计算

1. .apply(new WindowFunction( ){ }) // 方法参数里有当前窗口
2. .process.(new ProcessWindowFunction( ){ }) // 方法参数里有ctx上下文, 更全些

其他函数:

1. .trigger( ) 触发器: 定义 window 什么时候关闭, 关闭后触发计算并输出结果

2. .evitor( ) 移除器: 定义移除某些数据的逻辑

3. .allowedLateness( ) 允许处理迟到的数据

4. .sideOutputLateData( ) 将迟到的数据放入侧输出流

5. .getSideOutput( ) 获取侧输出流