devbox
代码探险家
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink学习笔记(二)_tumblingprocessingtimewindows

作者：代码探险家 | 2024-08-23 03:17:01

赞

踩

tumblingprocessingtimewindows

以下笔记基于对尚硅谷Java版Flink(2020版)的学习,Flink版本1.10

目录

Flink中的Window

  Window概述

  Window类型

    滚动窗口(Tumbling Windows)

    滑动窗口(Sliding Windows)

    会话窗口(Session Windows)

    全局窗口(global window)

  Window API

时间语义与Wartermark

  Flink中的时间语义

  EventTime的引入

  Watermark基本概念

  Watermark的引入

  窗口迟到数据的三种处理方法

Flink中的Window


  Window概述


    streaming流式计算是一种被设计用于处理无限数据集的数据处理引擎，无限数据集是指一种不断增长的本质上无限的数据集，而window是一种切割无限数据为有限块进行处理的手段。
    Window是无限数据流处理的核心，Window将一个无限的stream拆分成有限大小的“buckets”桶，我们可以在这些桶上做计算操作。


  Window类型


    时间窗口(TimeWindow)：滚动时间窗口，滑动时间窗口，会话窗口
    计数窗口(CountWindow)：滚动计数窗口，滑动计数窗口

    滚动窗口(Tumbling Windows)


        将数据依据固定的窗口长度对数据进行切片
        特点：时间对齐，窗口长度固定，没有重叠
        滚动窗口分配器将每个元素分配到一个指定窗口大小的窗口中，滚动窗口有一个固定的大小，并且不会出现重叠。
        适用场景：适合做BI统计等（做每个时间段的聚合计算）


    滑动窗口(Sliding Windows)


        滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成
        特点：时间对齐，窗口长度固定，可以有重叠
        滑动窗口分配器将元素分配到固定长度的窗口中，与滚动窗口类似，窗口的大小由窗口大小参数来配置，另一个窗口滑动参数控制滑动窗口开始的频率。因此，滑动窗口如果滑动参数小于窗口大小的话，窗口是可以重叠的，在这种情况下元素会被分配到多个窗口中
        适用场景：对最近一个时间段内的统计（求某接口最近5min的失败率来决定是否要报警）


    会话窗口(Session Windows)


        由一系列事件组合一个指定时间长度的timeout间隙组成，类似于web应用的session，也就是一段时间没有接收到新数据就会生成新的窗口
        特点：时间无对齐
        session窗口分配器通过session活动来对元素进行分组，session窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况，相反，当它在一个固定的时间周期内不再收到元素，即非活动间隔产生，那个这个窗口就会关闭
        一个session窗口通过一个session间隔来配置，这个session间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去


    全局窗口(global window)


  Window API

  1. // 对全数据开窗，会造成并行度变成1，不建议使用
  2. dataStream.windowAll(WindowAssigner<? super T,W> assigner)
  3.  
  4. // 分组流开窗，返回WindowedStream
  5. keyedStream.window(WindowAssigner<? super T,W> assigner)
  6.   // 基于事件时间的滚动时间窗口：
  7.   //   TumblingEventTimeWindows.of(Time size)
  8.   //   TumblingEventTimeWindows.of(Time size, Time offset)
  9.   // 基于处理时间的滚动时间窗口：
  10.   //   TumblingProcessingTimeWindows.of(Time size)
  11.   //   TumblingProcessingTimeWindows.of(Time size, Time offset)
  12.   // 基于事件时间的滑动时间窗口：
  13.   //   SlidingEventTimeWindows.of(Time size, Time slide)
  14.   //   SlidingEventTimeWindows.of(Time size, Time slide, Time offset)
  15.   // 基于处理时间的滑动时间窗口：
  16.   //   SlidingProcessingTimeWindows.of(Time size, Time slide)
  17.   //   SlidingProcessingTimeWindows.of(Time size, Time slide, Time offset)
  18.   // 基于事件时间的会话窗口：
  19.   //   EventTimeSessionWindows.withGap(Time size)
  20.   // 基于处理时间的会话窗口：
  21.   //   ProcessingTimeSessionWindows.withGap(Time size)
  22. // 分组流开滚动时间窗口，返回WindowedStream
  23. keyedStream.timeWindow(Time size)
  24. // 分组流开滑动时间窗口，返回WindowedStream
  25. keyedStream.timeWindow(Time size, Time slide)
  26. // 分组流开滚动计数窗口，返回WindowedStream
  27. keyedStream.countWindow(long size)
  28. // 分组流开滑动计数窗口，返回WindowedStream
  29. keyedStream.countWindow(long size, long slide)
  30.  
  31. // 滚动计算
  32. windowedStream.sum(int positionToSum)
  33. windowedStream.sum(String field)
  34. windowedStream.max(int positionToMax)
  35. windowedStream.max(String field)
  36. windowedStream.min(int positionToMin)
  37. windowedStream.min(String field)
  38.   // 指定字段以外的部分，取第一条记录的信息
  39. windowedStream.maxBy(int positionToMax)
  40. windowedStream.maxBy(String field)
  41. windowedStream.minBy(int positionToMin)
  42. windowedStream.minBy(String field)
  43.   // 指定字段以外的部分，取最大/最小记录的信息

    增量聚合函数
      每条数据到来就进行计算，保持一个简单的状态。等到窗口中的所有数据聚合完成的时候，再输出结果。效率高，延时低。

  1. // ReduceFunction
  2. windowedStream.reduce(ReduceFunction<T> function)
  3.   // 实现ReduceFunction接口,在reduce方法里参数1为状态行,参数2为新数据,返回值为输出
  4. // AggregateFunction
  5. windowedStream.aggregate(AggregateFunction<T,ACC,R> function)
  6.   // 实现AggregateFunction接口，实现下面的方法
  7.   //   createAccumulator()             创建累加器(状态存储用)
  8.   //   add(IN value, ACC accumulator)  处理一条数据，并返回累加器
  9.   //   getResult(ACC accumulator)      通过累加器获取输出内容
  10.   //   merge(ACC a, ACC b)             合并两个累加器，返回具有合并状态的累加器

    全窗口函数
      先把窗口所有数据收集起来（到桶里），等到计算的时候会遍历所有数据，并输出结果。

  1. // WindowFunction
  2. windowedStream.apply(WindowFunction<T,R,K,W> function)
  3.   // 实现WindowFunction接口,在apply方法里通过当前窗口的所有数据计算出需要输出的数据
  4. // ProcessWindowFunction
  5. process(ProcessWindowFunction<T,R,K,W> function)
  6.   // 实现ProcessWindowFunction接口,在process方法里通过当前窗口的所有数据计算出需要输出的数据

    其它可选API

  1. // 触发器,定义 window 什么时候关闭，触发计算并输出结果
  2. windowedStream.trigger(Trigger<? super T,? super W> trigger)
  3. // 移除器,定义移除某些数据不参与计算的逻辑
  4. windowedStream.evictor(Evictor<? super T,? super W> evictor)
  5. // 允许处理迟到的数据，叠加到已经计算的结果上。即窗口结果计算完成后，先输出一条，先不关闭，迟到数据每来一条再叠加到计算结果输出一次。
  6. windowedStream.allowedLateness(Time lateness)
  7. // 将迟到的数据放入侧输出流，时间窗口已经关闭，但是仍有迟到的数据，那么把迟到的数据和相应窗口的侧输出流数据做另外计算。
  8.   windowedStream.sideOutputLateData(OutputTag<T> outputTag)
  9. // 获取侧输出流
  10. singleOutputStreamOperator.getSideOutput(OutputTag<X> sideOutputTag)

时间语义与Wartermark


  Flink中的时间语义


    Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。
    Ingestion Time：是数据进入Flink的时间。
    Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是Processing Time。


  EventTime的引入

  1. // 从调用时刻开始给env创建的每一个stream追加时间特征
  2. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

  Watermark基本概念


    Watermark是一种衡量Event Time进展的机制。
    Watermark是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark机制结合window来实现。
    数据流中的Watermark用于表示timestamp小于Watermark的数据，都已经到达了，因此，window的执行也是由Watermark触发的。
    Watermark可以理解成一个延迟触发机制，我们可以设置Watermark的延时时长t，每次系统会校验已经到达的数据中最大的maxEventTime，然后认定eventTime小于maxEventTime - t的所有数据都已经到达，如果有窗口的停止时间等于maxEventTime – t，那么这个窗口被触发执行。
    当Flink接收到数据时，会按照一定的规则去生成Watermark，这条Watermark就等于当前所有到达数据中的maxEventTime - 延迟时长，也就是说，Watermark是基于数据携带的时间戳生成的，一旦Watermark比当前未触发的窗口的停止时间要晚，那么就会触发相应窗口的执行。
    由于event time是由数据携带的，因此，如果运行过程中无法获取新的数据，那么没有被触发的窗口将永远都不被触发。
    Watermark 就是触发前一窗口的“关窗时间”，一旦触发关门那么以当前时刻为准在窗口范围内的所有所有数据都会收入窗中。
    只要没有达到水位那么不管现实中的时间推进了多久都不会触发关窗。

    Watermark是一种特殊的数据，值只能增加不能减少。
    在并行处理时，如果上游有多个，则当前处理已其中最小的那个watermark为准，并且会把watermark向下游广播。

 

  Watermark的引入


    提取时间戳并周期性取得watermark

  1. // 设置提取watermark的周期间隔,默认为200毫秒
  2. env.getConfig().setAutoWatermarkInterval(long interval);
  3. // 设置周期性提取器
  4. dataStream.assignTimestampsAndWatermarks(AssignerWithPeriodicWatermarks<T> timestampAndWatermarkAssigner)
  5.   // 实现AssignerWithPeriodicWatermarks接口,在extractTimestamp方法里返回数据的时间戳,在getCurrentWatermark方法里返回当前的watermark
  6. // 设置周期性提取器：有界无序时间戳提取器,参数为watermark的延迟时长
  7. dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(Time.seconds(10)) {
  8.               @Override
  9.               public long extractTimestamp(UserBehavior element) {
  10.                   return element.timestamp * 1000;
  11.               }
  12.           } );
  13. // 设置周期性提取器：升序时间戳提取器,watermark的延迟时长为0
  14. dataStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
  15.               @Override
  16.               public long extractAscendingTimestamp(UserBehavior element) {
  17.                   return element.timestamp * 1000;
  18.               }
  19.           });

    提取时间戳并断点式取得watermark

  1. // 设置断点式提取器
  2. dataStream.assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks<T> timestampAndWatermarkAssigner)
  3.   // 实现AssignerWithPunctuatedWatermarks接口,在extractTimestamp方法里返回数据的时间戳,在checkAndGetNextWatermark方法里返回当前的watermark

  窗口迟到数据的三种处理方法


    1.使用watermark的延迟时长延迟窗口的关闭,watermark到达窗口结束时间时输出一个结果，如果没有设置allowedLateness就立即关闭窗口，如果设置了allowedLateness就不关闭。
    2.使用.allowedLateness(Time lateness)设置迟到数据延迟等待时间,在等待时间内每来一条数据，再叠加到计算结果输出一次。超过时间后，关闭窗口。
    3.使用.sideOutputLateData(OutputTag<T> outputTag)把窗口关闭后迟到的数据输出到侧输出流，后续使用侧输出流做处理。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/代码探险家/article/detail/1019224
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号