第一章.Flink的window机制

1.窗口概述及分类

  1. # 窗口概述
  2.     在流处理应用中,数据是连续不断的,因此我们不可能等到所有数据都到了才开始处理,当然我们可以每来一个消息就处理一次,但是有时我们需要做一些聚合类的运算,例如:在过去的1分钟内有多少用户点击了我们的网页,在这种情况下,我们必须定义一个窗口,用来收集最近一分钟内的数据,并对这个窗口中的数据进行计算
  3.     流式计算是一种被设计用于处理无限数据集的数据处理引擎,而无限数据集是指一种不断增长的本质上的无限的数据集,而window窗口是一种切割无限数据为有限块进行处理的手段
  4.     在Flink中,窗口(window)是处理无界流的核心,窗口把流切割成有限大小的多个存储桶(bucket),我们在这些桶上进行计算

窗口分为两类

  • 基于时间的窗口(时间驱动)
  • 基于元素个数的窗口(数据驱动)

2.基于时间的窗口

  • 时间窗口包含一个开始时间戳(包含)和结束时间戳(不包含)—->[start,end),这两个时间戳一起限制了窗口的尺寸,在Flink中,TimeWindow这个类表示基于时间的窗口
  • 时间窗口又分三种:滚动窗口,滑动窗口,会话窗口

一.滚动窗口(Tumbling Windows)

  • 滚动窗口有固定的大小,窗口与窗口之间不会重叠也没有缝隙,比如,如果指定一个长度为5分钟的窗口,当前窗口开始计算,每5分钟启动一个新的窗口
  • 滚动窗口能将数据流切分为不重叠的窗口,每一个事件只能属于一个窗口

day04[Flink流处理高阶编程(上)] - 图1

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.TimeCharacteristic;
  6. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  7. import org.apache.flink.streaming.api.datastream.WindowedStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  10. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  11. import org.apache.flink.streaming.api.windowing.time.Time;
  12. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  13. import org.apache.flink.util.Collector;
  14. import org.junit.After;
  15. import org.junit.Before;
  16. import org.junit.Test;
  18. public class $01_TimeWindow {
  19.      SingleOutputStreamOperator<WaterSensor> sensorDS = null;
  20.      WindowedStream<WaterSensor, String, TimeWindow> sensorWS = null;
  21.      StreamExecutionEnvironment env = null;
  22.      SingleOutputStreamOperator<String> resultDS = null;
  23.     @Before
  24.     public  void before(){
  25.         env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         //老版本需要该设置
  28.         //env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
  29.         sensorDS = env.socketTextStream("hadoop102", 9999)
  30.                 .map(new MapFunction<String, WaterSensor>() {
  31.                     @Override
  32.                     public WaterSensor map(String value) throws Exception {
  33.                         String[] line = value.split(",");
  34.                         return new WaterSensor(
  35.                                 line[0],
  36.                                 Long.parseLong(line[1]),
  37.                                 Integer.parseInt(line[2])
  38.                         );
  40.                     }
  41.                 });
  42.     }
  44.     /**
  45.      * 基于时间的滚动窗口
  46.      */
  47.     @Test
  48.     public void test01(){
  49.          sensorWS = sensorDS.keyBy(sensor -> sensor.getId())
  50.                  /**
  51.                   * flink version <= 1.11的写法
  52.                   */
  53.                  //.timeWindow(Time.seconds(3));
  54.                  /**
  55.                   * 新写法
  56.                  */
  57.                  .window(TumblingProcessingTimeWindows.of(Time.seconds(3)));
  59.     }
  61.     @After
  62.     public void after() throws Exception {
  63.          resultDS =  sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  64.             @Override
  65.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  66.                 out.collect("key" + s + "\n"
  67.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  68.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  69.             }
  70.         });
  72.         resultDS.print();
  73.         env.execute();
  75.     }
  76. }

二.滑动窗口(Sliding Windows)

  • 与滚动窗口一样,滑动窗口也有固定的长度,另外一个参数我们叫滑动步长,用来控制窗口启动的频率
  • 所以如果滑动步长小于窗口长度,滑动窗口会重叠,这种情况下,一个元素可能会被分配到多个窗口中

day04[Flink流处理高阶编程(上)] - 图2

示例代码

  1. /**
  2.      * 基于时间的滑动窗口
  3.      */
  4.     @Test
  5.     public void test02(){
  6.         sensorWS = sensorDS.keyBy(sensor -> sensor.getId())
  7.                 /**
  8.                  * <=1.11的写法
  9.                  */
  10.                 //.timeWindow(Time.seconds(5),Time.seconds(3));
  11.         /**
  12.          * 新写法
  13.          */
  14.         .window(TumblingProcessingTimeWindows.of(Time.seconds(3)));
  16.     }

三.会话窗口(Session Windows)

  • 会话窗口分配器会根据活动的元素进行分组,会话窗口不会有重叠,与滚动窗口和滑动窗口相比,会话窗口也没有固定的开启和关闭时间
  • 如果会话窗口有一段时间没有收到数据,会话窗口会自动关闭,这段没有收到数据的时间就是会话窗口的gap(间隔)
  • 我们可以配置静态的gap,也可以通过一个gap extractor 函数来定义gap的长度,当时间超过了这个gap,当前的会话窗口就会关闭,后续的元素会被分配到一个新的会话窗口

day04[Flink流处理高阶编程(上)] - 图3

示例代码

  1. /**
  2.      * 基于时间的会话窗口-静态Gap
  3.      */
  4.     @Test
  5.     public void test03(){
  6.         sensorWS = sensorDS.keyBy(sensor -> sensor.getId())
  8.                 .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));
  10.     }
  11.     /**
  12.      * 基于时间的会话窗口-动态Gap
  13.      */
  14.     @Test
  15.     public void test04(){
  16.         sensorWS = sensorDS.keyBy(sensor -> sensor.getId())
  18.                 .window(ProcessingTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<WaterSensor>() {
  19.                     @Override
  20.                     public long extract(WaterSensor element) {
  21.                         return element.getTs() * 1000L;
  22.                     }
  23.                 }));
  25.     }

3.全局窗口(Global Windows)

全局窗口分配器会分配相同key的所有元素,进入同一个Global window,这种窗口机制只有指定自定义的触发器时才有用,否则,不会做任何运算,因为这种窗口没有能够处理聚集在一起元素的结束点

day04[Flink流处理高阶编程(上)] - 图4

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  11. import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
  12. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  13. import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
  14. import org.apache.flink.streaming.api.windowing.time.Time;
  15. import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
  16. import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
  17. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  18. import org.apache.flink.util.Collector;
  20. import java.util.Arrays;
  22. public class $02_GlobalWindow {
  23.     public static void main(String[] args) throws Exception {
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         /**
  28.          * 老版本需要这么指定,默认是处理时间
  29.          * 新版本不需要指定,默认是事件时间,而且开窗的时候,自己就要明确指定时间语义
  30.          */
  31.         //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  32.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  33.                 .map(new MapFunction<String, WaterSensor>() {
  34.                     @Override
  35.                     public WaterSensor map(String value) throws Exception {
  36.                         String[] line = value.split(",");
  37.                         return new WaterSensor(
  38.                                 line[0],
  39.                                 Long.parseLong(line[1]),
  40.                                 Integer.parseInt(line[2])
  41.                         );
  43.                     }
  44.                 })
  45.                 .assignTimestampsAndWatermarks(
  46.                         WatermarkStrategy
  47.                         .<WaterSensor>forMonotonousTimestamps()
  48.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  49.                             @Override
  50.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  51.                                 return element.getTs() * 1000L;
  52.                             }
  53.                         })
  54.                 );
  56.         //TODO 全局窗口,需要自定义触发器
  57.         WindowedStream<WaterSensor, String, GlobalWindow> sensorWS = sensorDS.
  58.                 keyBy(sensor -> sensor.getId())
  59.                 .window(GlobalWindows.create())
  60.                 .trigger(CountTrigger.of(3))
  61.                 .evictor(TimeEvictor.of(Time.seconds(5)));
  64.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
  65.             @Override
  66.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  67.                 out.collect("key=" + s + "\n"
  68.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  69.                         + "窗口划分:[" + context.window().toString()+ ")" + "\n\n"
  70.                 );
  71.             }
  72.         });
  74.         resultDS.print();
  75.         env.execute();
  77.     }
  78. }

4.基于元素个数的窗口

一.滚动窗口

  • 默认的CountWindow是一个滚动窗口,只需要指定窗口大小即可,当元素数量达到窗口大小时,就会触发窗口的执行
  • 哪个窗口先达到3个元素,哪个窗口就关闭,不影响其他的窗口

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.datastream.WindowedStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  9. import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
  10. import org.apache.flink.util.Collector;
  12. public class $03_CountWindow {
  13.     public static void main(String[] args) throws Exception {
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         env.setParallelism(1);
  17.         /**
  18.          * 老版本需要这么指定,默认是处理时间
  19.          * 新版本不需要指定,默认是事件时间,而且开窗的时候,自己就要明确指定时间语义
  20.          */
  21.         //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  22.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  23.                 .map(new MapFunction<String, WaterSensor>() {
  24.                     @Override
  25.                     public WaterSensor map(String value) throws Exception {
  26.                         String[] line = value.split(",");
  27.                         return new WaterSensor(
  28.                                 line[0],
  29.                                 Long.parseLong(line[1]),
  30.                                 Integer.parseInt(line[2])
  31.                         );
  33.                     }
  34.                 });
  36.         //TODO 基于条数的滚动窗口
  37.         WindowedStream<WaterSensor, String, GlobalWindow> sensorWS = sensorDS.
  38.                 keyBy(sensor -> sensor.getId())
  39.                 .countWindow(3);
  42.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
  43.             @Override
  44.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  45.                 out.collect("key=" + s + "\n"
  46.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  47.                         + "窗口划分:[" + context.window().toString()+ ")" + "\n\n"
  48.                 );
  49.             }
  50.         });
  52.         resultDS.print();
  53.         env.execute();
  55.     }
  56. }

二.滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的,只是在传参数时需要传入两个参数,一个是window_size,一个是sliding_size

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.datastream.WindowedStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  9. import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
  10. import org.apache.flink.util.Collector;
  12. /**
  13.  * 窗口的划分并不是以第一条数据为基准的
  14.  *  第一个输出的窗口 = 步长
  15.  *  每经过一个步长,都有一个窗口输出,关闭
  16.  */
  18. public class $04_CountWindow2 {
  19.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(1);
  23.         /**
  24.          * 老版本需要这么指定,默认是处理时间
  25.          * 新版本不需要指定,默认是事件时间,而且开窗的时候,自己就要明确指定时间语义
  26.          */
  27.         //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  29.                 .map(new MapFunction<String, WaterSensor>() {
  30.                     @Override
  31.                     public WaterSensor map(String value) throws Exception {
  32.                         String[] line = value.split(",");
  33.                         return new WaterSensor(
  34.                                 line[0],
  35.                                 Long.parseLong(line[1]),
  36.                                 Integer.parseInt(line[2])
  37.                         );
  39.                     }
  40.                 });
  42.         //TODO 基于条数的滑动窗口
  43.         WindowedStream<WaterSensor, String, GlobalWindow> sensorWS = sensorDS.
  44.                 keyBy(sensor -> sensor.getId())
  45.                 .countWindow(5,2);
  48.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
  49.             @Override
  50.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  51.                 out.collect("key=" + s + "\n"
  52.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  53.                         + "窗口划分:[" + context.window().toString()+ ")" + "\n\n"
  54.                 );
  55.             }
  56.         });
  58.         resultDS.print();
  59.         env.execute();
  61.     }
  62. }

5.窗口函数

  • 指定了窗口的分配器后,需要指定如何计算,这事由窗口函数来负责,一旦窗口关闭,窗口函数去计算处理窗口中的每个元素
  • 窗口函数可以是ReduceFunction,AggregateFunction,ProcessWindowFunction中的任意一种
  • ReduceFunction,AggregateFunction更加高效,原因是Flink可以对到来的元素进行增量聚合,ProcessWindowFunction可以得到一个包含这个窗口的所有元素的迭代器,以及这些元素所属窗口的元数据信息
  • ProcessWindowFunction不能被高效执行的原因是Flink在执行这个函数之前,需要在内存缓存这个窗口上的所有元素

一.增量函数

增量函数:

  • 来一条处理一条
  • 窗口的输出次数:窗口触发关闭的时候,也就是说,每个窗口只会向下游传递一次结果
  • aggregate相比reduce:都是增量聚合
  • reduce第一条不调用方法
  • aggregate比reduce更灵活,reduce要求输入类型,中间状态,输出类型要一致,aggregate输入类型,中间状态(累加器),输出类型可以不一致

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.AggregateFunction;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.api.common.functions.ReduceFunction;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.datastream.WindowedStream;
  9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  10. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  11. import org.apache.flink.streaming.api.windowing.time.Time;
  12. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  13. import org.junit.After;
  14. import org.junit.Before;
  15. import org.junit.Test;
  17. public class $05_IncreFunction {
  18.     SingleOutputStreamOperator<WaterSensor> sensorDS = null;
  19.     WindowedStream<WaterSensor, String, TimeWindow> sensorWS = null;
  20.     StreamExecutionEnvironment env = null;
  21.     SingleOutputStreamOperator<WaterSensor> resultDS = null;
  22.     SingleOutputStreamOperator<String> resultDS1 = null;
  23.     @Before
  24.     public  void before(){
  25.         env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         sensorDS = env.socketTextStream("hadoop102", 9999)
  28.                 .map(new MapFunction<String, WaterSensor>() {
  29.                     @Override
  30.                     public WaterSensor map(String value) throws Exception {
  31.                         String[] line = value.split(",");
  32.                         return new WaterSensor(
  33.                                 line[0],
  34.                                 Long.parseLong(line[1]),
  35.                                 Integer.parseInt(line[2])
  36.                         );
  38.                     }
  39.                 });
  40.         sensorWS = sensorDS.keyBy(sensor -> sensor.getId())
  41.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));//窗口分配器
  42.     }
  44.     /**
  45.      * 窗口函数-增量函数 - reduce
  46.      */
  47.     @Test
  48.     public void test01(){
  49.         resultDS = sensorWS.reduce(new ReduceFunction<WaterSensor>() {
  50.             @Override
  51.             public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
  52.                 System.out.println(value1 + "<=======>" + value2);
  53.                 return new WaterSensor(value1.getId(), System.currentTimeMillis(), value1.getVc()+value2.getVc());
  54.             }
  55.         });
  58.     }
  59.     /**
  60.      * 窗口函数-增量函数 -aggregate
  61.      */
  62.     @Test
  63.     public void test02(){
  64.         resultDS1 = sensorWS.aggregate(new AggregateFunction<WaterSensor, Integer, String>() {
  65.             /**
  66.              * 初始化累加器
  67.              * @return
  68.              */
  69.             @Override
  70.             public Integer createAccumulator() {
  71.                 System.out.println("create.....");
  72.                 return 0;
  73.             }
  75.             /**
  76.              * 累加的逻辑
  77.              * @param value
  78.              * @param accumulator
  79.              * @return
  80.              */
  81.             @Override
  82.             public Integer add(WaterSensor value, Integer accumulator) {
  83.                 System.out.println("add....");
  84.                 return value.getVc() + accumulator;
  85.             }
  87.             /**
  88.              * 获取结果
  89.              * @param accumulator
  90.              * @return
  91.              */
  92.             @Override
  93.             public String getResult(Integer accumulator) {
  94.                 System.out.println("getResult....");
  95.                 return accumulator.toString();
  96.             }
  98.             /**
  99.              * 只有会话窗口才会调用
  100.              * @param a
  101.              * @param b
  102.              * @return
  103.              */
  104.             @Override
  105.             public Integer merge(Integer a, Integer b) {
  106.                 System.out.println("merge.....");
  107.                 return a + b;
  108.             }
  109.         });
  112.     }
  117.     @After
  118.     public void after() throws Exception {
  119.         if(resultDS != null){
  120.             resultDS.print();
  121.         }
  122.         if(resultDS1 != null){
  123.             resultDS1.print();
  124.         }
  125.         env.execute();
  126.     }
  131. }

二.全量函数

全量函数:数据全部存起来,触发的时候一次性计算和输出

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.datastream.WindowedStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  9. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  10. import org.apache.flink.streaming.api.windowing.time.Time;
  11. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  12. import org.apache.flink.util.Collector;
  14. import java.util.Arrays;
  16. /**
  17.  * 窗口函数-->全量函数
  18.  */
  20. public class $06_AllFunctionWindow {
  21.     public static void main(String[] args) throws Exception {
  23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  24.         env.setParallelism(1);
  25.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  26.                 .map(new MapFunction<String, WaterSensor>() {
  27.                     @Override
  28.                     public WaterSensor map(String value) throws Exception {
  29.                         String[] line = value.split(",");
  30.                         return new WaterSensor(
  31.                                 line[0],
  32.                                 Long.parseLong(line[1]),
  33.                                 Integer.parseInt(line[2])
  34.                         );
  36.                     }
  37.                 });
  39.         //TODO 全局窗口,需要自定义触发器
  40.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  41.                 keyBy(sensor -> sensor.getId())
  42.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
  45.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  46.             @Override
  47.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  48.                 out.collect("key=" + s + "\n"
  49.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  50.                         + "数据为：" + Arrays.asList(elements).toString() + "\n"
  51.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n"
  52.                 );
  53.             }
  54.         });
  56.         resultDS.print();
  57.         env.execute();
  59.     }
  60. }

6.Non-Keyed Windows

  • 在用窗口前首先需要确认应该是在keyBy之后的流上用,还是在没有keyBy的流上用
  • 在keyed streams上使用窗口,窗口计算被并行的运用在多个task上,可以认为每个task都有自己单独窗口
  • 在非non-keyed stream上使用窗口,流的并行度只能是1,所有的窗口逻辑只能在一个单独的task上执行
  • 需要注意的是:非key分区的流上使用window,如果把并行度强行设置为>1,则会抛出异常

示例代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.functions.MapFunction;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
  8. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  9. import org.apache.flink.streaming.api.windowing.time.Time;
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  11. import org.apache.flink.util.Collector;
  13. import java.util.Arrays;
  15. /**
  16.  * NoKeyedWindow
  17.  */
  19. public class $07_NoKeyedWindow {
  20.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(3);
  24.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  25.                 .map(new MapFunction<String, WaterSensor>() {
  26.                     @Override
  27.                     public WaterSensor map(String value) throws Exception {
  28.                         String[] line = value.split(",");
  29.                         return new WaterSensor(
  30.                                 line[0],
  31.                                 Long.parseLong(line[1]),
  32.                                 Integer.parseInt(line[2])
  33.                         );
  35.                     }
  36.                 });
  38.         sensorDS.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
  39.                 .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
  40.                     @Override
  41.                     public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  42.                         out.collect(Arrays.asList(elements).toString());
  43.                     }
  44.                 }).print();
  45.         env.execute();
  47.     }
  48. }

7.窗口源码分析

以滚动窗口为例

  1. package com.atguigu.flink.day04;
  4. /**
  5.  * 窗口相关源码分析(以滚动窗口为例)
  6.  */
  8. public class $09_SourceCodeAnalysis {
  9.     public static void main(String[] args) throws Exception {
  10.         /**
  11.          * 
  12.          * 窗口是怎么划分的
  13.          *    long start =TimeWindow.getWindowStartWithOffset(
  14.          *      now, (globalOffset + staggerOffset) % size, size);
  15.          *    return Collections.singletonList(new TimeWindow(start, start + size));
  16.          *    public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
  17.          *         return timestamp - (timestamp - offset + windowSize) % windowSize;
  18.          *     }
  19.          *
  20.          */
  24.         /**
  25.          *
  26.          * 窗口为什么左闭右开
  27.          * Gets the largest timestamp that still belongs to this window.
  28.          *
  29.          * <p>This timestamp is identical to {@code getEnd() - 1}.
  30.          *
  31.          * @return The largest timestamp that still belongs to this window.
  32.          * @see #getEnd()
  33.          *
  34.          *
  35.          *         @Override
  36.          *         public long maxTimestamp() {
  37.          *             return end - 1;
  38.          *         }
  39.          */
  42.         /**
  43.          * 窗口什么时候触发
  44.          *  @Override
  45.          *     public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
  46.          *        return TriggerResult.FIRE;
  47.          *     }
  48.          * 时间 >= maxTimestamp
  49.          */
  51.         /**
  52.          * 窗口什么时候创建？窗口什么时候销毁？（窗口的生命周期？）
  53.          *             创建： 属于本窗口的第一条数据来的时候，new的，放到一个 单例集合
  54.          *             销毁： cleanupTime = window.maxTimestamp() + allowedLateness;
  55.          *                 清空状态的时间 = 时间进展到 窗口最大时间戳 + 允许迟到的时间
  56.          */
  59.     }
  60. }

第二章.Flink的时间语义与WaterMark

1.时间语义

一.事件时间

  1.     事件时间是指这个事件发生的时间
  2.     在event进入flink之前,通常被嵌入到了event中,一般作为这个event的时间戳存在
  3.     在事件时间体系中,时间的进度依赖于事件本身,和任何设备的时间无关,事件时间必须制定如何产生Event Time Watermarks(水印),在事件时间体系中,水印是代表时间进度的标志(作用相当于现实时间的时钟)
  4.     在理想条件下,不管事件时间何时到达或者他们到达的顺序如何,事件时间处理将产生完全一致且确定的结果,事件时间会在等待无序时间(迟到时间)时产生一定的延迟,由于只能等待有限的时间,因此这限制了确定性事件时间应用程序的可使用性
  5.     假设所有数据都已到达,事件时间操作将按照预期方式进行,即使在处理无序或迟到的事件或重新处理历史数据时,也会产生正确且一致的效果,例如,每小时事件时间窗口将包含带有事件时间戳的所有记录,将记录计入该小时,无论他们到达的顺序或处理时间
  6.     在使用窗口的时候,如果使用事件时间,就指定时间分配器为事件时间分配器

注意: 在1.12之前默认的时间语义是处理时间,从1.12开始,Flink内部已经把默认的语义改成了事件时间

二.处理时间

  1.     处理时间指的是执行操作的各个设备的时间
  2.     对于运行在处理时间上的流程序,所有的基于时间的操作(比如时间窗口)都是使用的设备时钟,比如一个长度为1个小时的窗口将会包含设备时钟表示的1个小时内所有的数据,假设应用程序在9:15am启动,第一个小时窗口将会包含9:15am到10:00am所有的数据,然后下个窗口是10:00am-11.00am等
  3.     处理时间是最简单时间语义,数据流与设备之间不需要做任何的协调,它提供了最好的性能和最低的延迟,但是在分布式和异步的环境之下,处理时间没有方法保证确定性,容易受到数据传递速度的影响:事件的延迟和乱序
  4.     在使用窗口的时候,如果使用处理时间,就指定时间分配器为处理时间分配器

三.案例

day04[Flink流处理高阶编程(上)] - 图5

四.代码

  1. package com.atguigu.flink.day04;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.streaming.api.TimeCharacteristic;
  8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  9. import org.apache.flink.streaming.api.datastream.WindowedStream;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
  12. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
  13. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  14. import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
  15. import org.apache.flink.streaming.api.windowing.time.Time;
  16. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  17. import org.apache.flink.util.Collector;
  19. import java.util.Arrays;
  21. public class $08_TimeCharacteristic_EventTime {
  22.     public static void main(String[] args) throws Exception {
  24.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  25.         env.setParallelism(1);
  26.         /**
  27.          * 老版本需要这么指定,默认是处理时间
  28.          * 新版本不需要指定,默认是事件时间,而且开窗的时候,自己就要明确指定时间语义
  29.          */
  30.         //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  31.         SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 9999)
  32.                 .map(new MapFunction<String, WaterSensor>() {
  33.                     @Override
  34.                     public WaterSensor map(String value) throws Exception {
  35.                         String[] line = value.split(",");
  36.                         return new WaterSensor(
  37.                                 line[0],
  38.                                 Long.parseLong(line[1]),
  39.                                 Integer.parseInt(line[2])
  40.                         );
  42.                     }
  43.                 })
  44.                 .assignTimestampsAndWatermarks(
  45.                         WatermarkStrategy
  46.                         .<WaterSensor>forMonotonousTimestamps()
  47.                         .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  48.                             @Override
  49.                             public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  50.                                 return element.getTs() * 1000L;
  51.                             }
  52.                         })
  53.                 );
  56.         WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS.
  57.                 keyBy(sensor -> sensor.getId())
  58.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)));
  61.         SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  62.             @Override
  63.             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  64.                 out.collect("key" + s + "\n"
  65.                         + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
  66.                         + "数据为:" + Arrays.asList(elements).toString() + "\n"
  67.                         + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n");
  68.             }
  69.         });
  71.         resultDS.print();
  72.         env.execute();
  74.     }
  75. }

2.WaterMark初理解

  • WaterMark是用来衡量事件时间的进展
  • 用来解决乱序的问题
  • 是一个特殊的时间戳,从指定生成的位置插入到流里
  • 单调不减的
  • 用来触发窗口的
  • Flink认为,事件时间小于WaterMark的数据应该都已经处理过了,如果后续还有事件时间小于watermark的数据来,称为迟到数据
  1.     支持event time的流式处理框架需要一种能够测量 event time 进度的方式,比如一个窗口算子创建了一个长度为1小时的窗口,那么这个算子需要知道事件时间已经到达了这个窗口的关闭时间,从而在程序中去关闭这个窗口
  2.     事件时间可以不依赖处理时间来表示时间的进度,例如在程序中,即使处理时间和事件时间有相同的速度,事件时间可能会轻微的落后处理时间,另外一方面,使用事件时间可以在几秒内处理已经缓存在kafka中的数据,这些数据可以照样被正确处理,就像实时发生的一样能够进入正确的窗口
  3.     这种在flink中去测量事件时间的进度的机制就是waterMark(水印),waterMark作为数据流的一部分在流动,并且携带一个时间戳t
  4.     一个WaterMark(t)表示在这个流里面事件时间已经到了时间t,意味着此时,流中不应该存在这样的数据:它的时间戳t2<=t(时间比较旧或者等于时间戳)

WaterMark原理图

day04[Flink流处理高阶编程(上)] - 图6

day04[Flink流处理高阶编程(上)] - 图7