Flink原理实战每日一篇04

淡淡的烟草味﹌ 2022-01-10 03:05 226阅读 0赞

1，Window  分为两个大类，keyed 和 no-keyed 窗口

使用keyed window之后，相同的key的数据会分配在同一个分区。

使用no-keyed 比如windowAll()会将所有的数据路由到一个Task中计算（这样子想想会不会出现一个task卡死的情况？）

简单代码：

val data7: DataStream[Person] = env.fromElements(new Person("a", 1), new Person("c", 2))
    data7.keyBy(_.age).window(new MyWindow())
    data7.windowAll(new MyWindow()

2，重点来了 Window Assigner 分为两种类型的窗口：

1）基于时间的Time Window

滚动 **Tumbling** ，固定时间大小进行切分：

基于不同的时间策略

val data7: DataStream[Person] = env.fromElements(new Person("a", 1), new Person("c", 2))
    data7.keyBy(_.age).window(TumblingEventTimeWindows.of(Time.seconds(10)))
    data7.keyBy(_.age).window(TumblingProcessingTimeWindows.of(Time.seconds(10)))

滑动 窗口，不解释了，就是有重叠窗口数据需要指定窗口大小跟滑动的窗口大小：

data7.keyBy(_.age).window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
    data7.keyBy(_.age).window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))

会话窗口 主要将某段时间内活跃度较高的数据聚合成一个窗口进行计算，窗口的触发条件是 Session Gap ，使用场景主要是根据用户在线上某段时间内的活跃度对用户行为数据进行统计。

// withGap 指定的是不活跃时间间隔
    data7.keyBy(_.age).window(EventTimeSessionWindows.withGap(Time.milliseconds(10)))
    data7.keyBy(_.age).window(ProcessingTimeSessionWindows.withGap(Time.milliseconds(10)))

如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

动态调整Session Gap ：

data7.keyBy(_.age).window(EventTimeSessionWindows.withDynamicGap(
      new SessionWindowTimeGapExtractor[Person] {
        override def extract(element: Person): Long = {
           //动态指定Session Gap
          if (element.age==1 ) 1L else 2L
        }
      }
    ))
    data7.keyBy(_.age).window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor[Person] {
      override def extract(element: Person): Long = {
        1L
      }
    }))
    
    
    全局窗口 Global Windows 将所有相同的key的数据分配到单个窗口中计算结果，窗口没有起始和结束时间，需要借助于Triger来触发计算，不使用Triger，是触发不了计算的。使用这个窗口 需要非常的谨慎！！

data7.keyBy(_.age).window(GlobalWindows.create()).process(....)

2）基于数量的窗口  countWindows

说道这里哈，刚好有朋友有个业务场景，**计算最近5条数据的成功率**，不就是count Window的实际使用场景吗？ 使用Count Window 算子 加上滑动~~~~

3，Window Function   窗口函数，一般分为两大类，增量聚合和全量窗口函数

增量函数计算性能较高，占用空间小，主要基于中间状态的计算结果，维护中间结果状态值

全量窗口函数使用代价高，性能比较弱，主要是需要对窗口接入的数据进行缓存。

1） ReduceFunction算子

object WindowFunctionDemo {
      def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        val dataStream: DataStream[(Int, Int)] = env.fromCollection(List((1, 2), (3, 4), (1, 3)))
        dataStream.keyBy(_._1)
          .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
          .reduce { (v1, v2) => (v1._1, v1._2 + v2._2) }
    
        env.execute()
      }
    
    }

定义接口的方式：

// 定义接口的方式
    dataStream.keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
        .reduce(new ReduceFunction[(Int, Int)] {
          override def reduce(value1: (Int, Int), value2: (Int, Int)): (Int, Int) = {
             (value1._1, value2._2 + value2._2)
          }
        })

2） AggregateFunction ，它也是基于中间状态计算结果的增量函数，它比reduce函数更灵活，实现复杂度更高，上代码：

val dataStream2: DataStream[(String, Long)] = env.fromElements(("a",1L),("a",2L),("b",1L))
        val data = dataStream2.keyBy(_._1)
          .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
          //根据ID求第二个字段的平均值
    //        .aggregate(new MyAggregateFunction1)
            .aggregate(new AggregateFunction[(String, Long), (Long, Long), Double] {
          override def createAccumulator(): (Long, Long) = {
            (0L, 0L)
          }
    
          //定义输入的数据添加到累加器的逻辑
          override def add(input: (String, Long), acc: (Long, Long)): (Long, Long) = {
            (acc._1 + input._2, acc._2 + 1L)
          }
    
          //根据累加器得到结果
          override def getResult(acc: (Long, Long)): Double = {
            acc._1 / acc._2
          }
    
          //定义累加器合并的逻辑
          override def merge(acc1: (Long, Long), acc2: (Long, Long)): (Long, Long) = {
            (acc1._1 + acc2._1, acc1._2 + acc2._2)
          }
        })

3） FoldFunction 定义了如何将窗口中的输入元素与外部的元素合并的逻辑，这个就是动态添加前缀的业务场景吗？

上代码：  不过代码显示fold函数已经过时~ 我这是1.6.2版本............额 好吧。官方建议使用Aggregate函数来替代它

//将字符串加在第二位参数的后面
    val data2 = dataStream2.keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
      .fold("flink") {
        (acc, value) => acc + value._2
      }

4）ProcessWindowFunction 这个函数是全量函数，计算依赖窗口所有的数据元素。

代码：

val dataStream3: DataStream[(String, Long)] = env.fromElements(("a", 1L), ("a", 2L), ("b", 1L))
    val data3 = dataStream2.keyBy(_._1)
        .timeWindow(Time.seconds(1))
    
        .process(new ProcessWindowFunction[(String,Long),(String,Long,Long,Long,Long,Long),String,TimeWindow] {
          override def process(key: String, context: Context,
                               values: Iterable[(String, Long)],
                               out: Collector[(String, Long, Long, Long, Long,Long)]): Unit = {
            
            val sum = values.map(_._2).sum
            val min = values.map(_._2).min
            val max = values.map(_._2).max
            val avg = sum / values.size
            // 窗口时间逻辑
            val windowEnd = context.window.getEnd
            out.collect(key,sum,min,max,avg,windowEnd)
          }
        })

总结：ProcessWindowFunction也可以做增量统计，但是很消耗性能，一般跟增量函数结合起来去做增量统计，ProcessWindowFunction比其他函数要灵活，它也可以操作基于窗口之上的状态操作，主要是通过Context对象获取并操作Per-window state数据，window state 分为两个类型 globalStatre ：窗口的keyed state数据不限定在某个窗口中

windowStates 窗口的keyed state数据限定在固定的窗口中。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20220109/625c05427b1f45deadf778142c9189ea.png