• 7、如何使用Flink中的窗口（Window算子）

    目录

    1、如何理解 Flink中的窗口（window）

    2、Flink中窗口的类型

    2.1 根据上游DataStream类型分类

    2.2 根据驱动类型分类

    2.3 根据进入到窗口数据的分发规则分类

    3、怎样使用 Flink中的 Window算子

    4、怎样使用 Flink中的 Window Assigners

    4.1、基于处理时间的滑动窗口

    4.2、基于处理时间的滚动窗口

    4.3、基于处理时间的会话窗口

    4.4、基于事件时间的滑动窗口

    4.5、基于事件时间的滚动窗口

    4.6、基于事件时间的会话窗口

    4.7、计数窗口

    5、怎样使用 Flink中的 Window Funcation

    5.1、ReduceFunction

    5.2、AggregateFunction

    5.3、ProcessWindowFunction

    5.4、增量聚合的 ProcessWindowFunction

    6、Flink中的 Window的生命周期

    6.1、触发器 - Triggers

    6.2、移除器 - Evictors

    7、对迟到数据的处理

    7.1、生成水位线时，设置最大乱序时间

    7.2、设置窗口延迟关闭 - allowedLateness

    7.3、使用侧输出流获取迟到的数据 - sideOutputLateData

    1、如何理解 Flink中的窗口（window）

            Flink中的窗口好像一个桶，可以根据不同的时间语义，将无界流中的数据分配到指定的桶中去，再通过 水位线或者处理时间 触发对桶中的数据进行计算

            其目的就是为了将无限的数据 根据指定的规则进行切分成有限的数据进行计算

    2、Flink中窗口的类型

    2.1 根据上游DataStream类型分类

    按键分区窗口(Keyed Windows) ：

             基于KeyedStream做窗口操作，窗口计算会在多个并行子任务上同时执行，相同的key的数据会进入到同一个窗口中去。

    非按键分区-Non-Keyed Windows ：

            基于DataStream做窗口操作，流上的数据会进入同一窗口中，只能有一个Task处理(不推荐这种方式)

    2.2 根据驱动类型分类

    时间窗口(Time Window)：

            通过指定的时间语义的时间点来定义窗口的开始和结束，流中的数据被分配到哪个窗口中，也由数据上的时间标识来决定，当接收到带有结束时间标识的数据时，将触发窗口计算，并销毁窗口

    计数窗口(Count Window):

           窗口的大小由数据个数来限制，当窗口内接收的数据个数到达窗口大小时，将触发窗口计算，并销毁窗口

     2.3 根据进入到窗口数据的分发规则分类

    滚动窗口（Tumbling Windows）：

            滚动窗口的大小是固定的，且各自范围之间不重叠。 比如说，如果你指定了滚动窗口的大小为 5 分钟，那么每 5 分钟就会有一个窗口被计算，且一个新的窗口被创建（如下图所示）

    滑动窗口（Sliding Windows）：

            滑动窗口的 assigner 分发元素到指定大小的窗口，窗口大小通过 window size 参数设置。 滑动窗口需要一个额外的滑动距离（window slide）参数来控制生成新窗口的频率。 因此，如果 slide 小于窗口大小，滑动窗口可以允许窗口重叠。这种情况下，一个元素可能会被分发到多个窗口。

    比如说，你设置了大小为 10 分钟，滑动距离 5 分钟的窗口，你会在每 5 分钟得到一个新的窗口， 里面包含之前 10 分钟到达的数据（如下图所示）

    会话窗口（Session Windows）： 

             与滚动窗口和滑动窗口不同，会话窗口不会相互重叠，且没有固定的开始或结束时间。 会话窗口在一段时间没有收到数据之后会关闭，即在一段不活跃的间隔之后。 会话窗口的 assigner 可以设置固定的会话间隔（session gap）或 用 session gap extractor 函数来动态地定义多长时间算作不活跃。 当超出了不活跃的时间段，当前的会话就会关闭，并且将接下来的数据分发到新的会话窗口。

    全局窗口（Global Windows）： 

    全局窗口的 assigner 将拥有相同 key 的所有数据分发到一个全局窗口。 这样的窗口模式仅在你指定了自定义的 trigger 时有用。 否则，计算不会发生，因为全局窗口没有天然的终点去触发其中积累的数据。

    3、怎样使用 Flink中的 Window算子

    Keyed Windows API :

    1. stream
    2.        .keyBy(...)               <-  仅 keyed 窗口需要
    3.        .window(...)              <-  必填项："assigner" 	 指定窗口类型
    4.       [.trigger(...)]            <-  可选项："trigger"  	 指定触发器
    5.       [.evictor(...)]            <-  可选项："evictor"  	 指定移除器
    6.       [.allowedLateness(...)]    <-  可选项："lateness" 	 指定窗口延迟关闭时间 
    7.       [.sideOutputLateData(...)] <-  可选项："output tag" 
    8.        .reduce/aggregate/apply() <-  必填项："function"   指定窗口聚合函数
    9.       [.getSideOutput(...)]      <-  可选项："output tag" 指定侧输出流(用来接收迟到数据)

    Non-Keyed Windows API :

    1. stream
    2.        .windowAll(...)           <-  必填项："assigner" 	 指定窗口类型
    3.       [.trigger(...)]            <-  可选项："trigger"  	 指定触发器
    4.       [.evictor(...)]            <-  可选项："evictor"  	 指定移除器
    5.       [.allowedLateness(...)]    <-  可选项："lateness" 	 指定窗口延迟关闭时间 
    6.       [.sideOutputLateData(...)] <-  可选项："output tag" 指定侧输出流(用来接收迟到数据)
    7.        .reduce/aggregate/apply() <-  必填项："function"   指定窗口聚合函数
    8.       [.getSideOutput(...)]      <-  可选项："output tag"

    4、怎样使用 Flink中的 Window Assigners

    功能说明：

            通过stream.window(WindowAssigner) 来指定 窗口的类型

    4.1、基于处理时间的滑动窗口

    使用场景：

            每隔x秒统计近y秒内的数据(y>x)

    代码示例：

    1. /*
    2.  * TODO 基于处理时间的滑动窗口
    3.  *     每2秒计算最近10秒内的数据
    4.  * */
    5. public class SlidingProcessingTimeWindow {
    6.     public static void main(String[] args) throws Exception {
    7.         // 1.获取执行环境
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         env.setParallelism(3);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         // TODO Window：KeyedStream → WindowedStream
    13.         Window(env);
    14.  
    15.         // TODO WindowAll：DataStream → AllWindowedStream
    16.         //WindowAll(env);
    17.  
    18.         // 3.触发程序执行
    19.         env.execute();
    20.     }
    21.  
    22.     // TODO Keyed Windows
    23.     private static void Window(StreamExecutionEnvironment env) {
    24.         env
    25.                 .socketTextStream("localhost", 9999)
    26.                 .map(new MapFunction>() {
    27.                          @Override
    28.                          public Tuple2 map(String value) throws Exception {
    29.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                          }
    31.                      }
    32.                 )
    33.                 .keyBy(value -> value.f0)
    34.                 // 滑动窗口，窗口长度10s，滑动步长2s
    35.                 .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(2)))
    36.                 .process(new ShowProcessWindowFunction())
    37.                 .print()
    38.         ;
    39.     }
    40.  
    41.     // TODO Non-Keyed Windows
    42.     private static void WindowAll(StreamExecutionEnvironment env) {
    43.         env
    44.                 .socketTextStream("localhost", 9999)
    45.                 .map(new MapFunction>() {
    46.                          @Override
    47.                          public Tuple2 map(String value) throws Exception {
    48.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    49.                          }
    50.                      }
    51.                 )
    52.                 // 滑动窗口，窗口长度5s，滑动步长2s
    53.                 .windowAll(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(2)))
    54.                 .process(new ShowProcessAllWindowFunction())
    55.                 .print()
    56.         ;
    57.     }
    58. }

    4.2、基于处理时间的滚动窗口

    使用场景：

            计算固定时间段内的数据

    代码示例：

    1. /*
    2.  * TODO 基于处理时间的滚动窗口
    3.  *   计算每小时内的用户访问数
    4.  * */
    5. public class TumblingProcessingTimeWindow {
    6.     public static void main(String[] args) throws Exception {
    7.         // 1.获取执行环境
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         env.setParallelism(3);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         // TODO Window：KeyedStream → WindowedStream
    13.         //Window(env);
    14.  
    15.         // TODO WindowAll：DataStream → AllWindowedStream
    16.         WindowAll(env);
    17.  
    18.         // 3.触发程序执行
    19.         env.execute();
    20.     }
    21.  
    22.     // TODO Keyed Windows
    23.     private static void Window(StreamExecutionEnvironment env) {
    24.         env
    25.                 .socketTextStream("localhost", 9999)
    26.                 .map(new MapFunction>() {
    27.                          @Override
    28.                          public Tuple2 map(String value) throws Exception {
    29.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                          }
    31.                      }
    32.                 )
    33.                 .keyBy(value -> value.f0)
    34.                 // 滚动窗口，窗口长度5s
    35.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    36.                 .process(new ShowProcessWindowFunction())
    37.                 .print()
    38.         ;
    39.     }
    40.  
    41.     // TODO Non-Keyed Windows
    42.     private static void WindowAll(StreamExecutionEnvironment env) {
    43.         env
    44.                 .socketTextStream("localhost", 9999)
    45.                 .map(new MapFunction>() {
    46.                          @Override
    47.                          public Tuple2 map(String value) throws Exception {
    48.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    49.                          }
    50.                      }
    51.                 )
    52.                 // 滚动窗口，窗口长度5s
    53.                 .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    54.                 .process(new ShowProcessAllWindowFunction())
    55.                 .print()
    56.         ;
    57.     }
    58. }

    4.3、基于处理时间的会话窗口

    使用场景：

            计算指定时间间隔内的数据

    代码示例：

    1. /*
    2.  * TODO 基于处理时间的会话窗口
    3.  *    相邻两个元素的处理时间间隔 大于指定会话周期 触发窗口计算
    4.  * */
    5. public class ProcessingTimeSessionWindow {
    6.     public static void main(String[] args) throws Exception {
    7.         // 1.获取执行环境
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         env.setParallelism(3);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         // TODO Window：KeyedStream → WindowedStream
    13.         //Window(env);
    14.  
    15.         // TODO WindowAll：DataStream → AllWindowedStream
    16.         WindowAll(env);
    17.  
    18.         // 3.触发程序执行
    19.         env.execute();
    20.     }
    21.  
    22.     // TODO Keyed Windows
    23.     private static void Window(StreamExecutionEnvironment env) {
    24.         env
    25.                 .socketTextStream("localhost", 9999)
    26.                 .map(new MapFunction>() {
    27.                          @Override
    28.                          public Tuple2 map(String value) throws Exception {
    29.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                          }
    31.                      }
    32.                 )
    33.                 .keyBy(value -> value.f0)
    34.                 // 会话窗口，超时间隔5s
    35.                 .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
    36.                 .process(new ShowProcessWindowFunction())
    37.                 .print()
    38.         ;
    39.     }
    40.  
    41.     // TODO Non-Keyed Windows
    42.     private static void WindowAll(StreamExecutionEnvironment env) {
    43.         env
    44.                 .socketTextStream("localhost", 9999)
    45.                 .map(new MapFunction>() {
    46.                          @Override
    47.                          public Tuple2 map(String value) throws Exception {
    48.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    49.                          }
    50.                      }
    51.                 )
    52.                 // 会话窗口，超时间隔5s
    53.                 .windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
    54.                 .process(new ShowProcessAllWindowFunction())
    55.                 .print()
    56.         ;
    57.     }
    58. }

    4.4、基于事件时间的滑动窗口

    代码示例：

    1. /*
    2.  * TODO 基于事件时间的滑动窗口
    3.  *   每2秒计算一次最近10秒内的数据
    4.  * */
    5. public class SlidingEventTimeWindow {
    6.     public static void main(String[] args) throws Exception {
    7.         // 1.获取执行环境
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         env.setParallelism(3);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         // TODO Window：KeyedStream → WindowedStream
    13.         Window(env);
    14.  
    15.         // TODO WindowAll：DataStream → AllWindowedStream
    16.         //WindowAll(env);
    17.  
    18.         // 3.触发程序执行
    19.         env.execute();
    20.     }
    21.  
    22.     // TODO Keyed Windows
    23.     private static void Window(StreamExecutionEnvironment env) {
    24.         env
    25.                 .socketTextStream("localhost", 9999)
    26.                 .map(new MapFunction>() {
    27.                          @Override
    28.                          public Tuple2 map(String value) throws Exception {
    29.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                          }
    31.                      }
    32.                 )
    33.                 .assignTimestampsAndWatermarks(
    34.                         WatermarkStrategy
    35.                                 .>forMonotonousTimestamps()
    36.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    37.                 )
    38.                 .keyBy(value -> value.f0)
    39.                 // 滚动窗口，窗口长度10s，滑动步长2s
    40.                 .window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
    41.                 .process(new ShowProcessWindowFunction())
    42.                 .print()
    43.         ;
    44.     }
    45.  
    46.     // TODO Non-Keyed Windows
    47.     private static void WindowAll(StreamExecutionEnvironment env) {
    48.         env
    49.                 .socketTextStream("localhost", 9999)
    50.                 .map(new MapFunction>() {
    51.                          @Override
    52.                          public Tuple2 map(String value) throws Exception {
    53.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    54.                          }
    55.                      }
    56.                 )
    57.                 .assignTimestampsAndWatermarks(
    58.                         WatermarkStrategy
    59.                                 //.>forMonotonousTimestamps()
    60.                                 .>forGenerator(new PeriodWatermarkStrategy())
    61.                                 .withTimestampAssigner(
    62.                                         (Tuple2 element, long recordTimestamp) -> {
    63.                                             //System.out.println("Step1：extractTimestamp-从事件数据中提取时间戳");
    64.                                             return element.f1;
    65.                                         }
    66.                                 )
    67.                         //.withIdleness(Duration.ofSeconds(5))  //空闲等待5s
    68.                 )
    69.                 .keyBy(value -> value.f0)
    70.                 // 滑动窗口，窗口长度10s，滑动步长2s
    71.                 .windowAll(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(2)))
    72.                 .process(new ShowProcessAllWindowFunction())
    73.                 .print()
    74.         ;
    75.     }
    76.  
    77. }

    4.5、基于事件时间的滚动窗口

    代码示例：

    1. /*
    2.  * TODO 基于事件时间的滚动窗口
    3.  *   计算每个窗口周期内的数据（计算每小时内的用户访问数）
    4.  * */
    5. public class TumblingEventTimeWindow {
    6.     public static void main(String[] args) throws Exception {
    7.         // 1.获取执行环境
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         env.setParallelism(3);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         // TODO Window：KeyedStream → WindowedStream
    13.         //Window(env);
    14.  
    15.         // TODO WindowAll：DataStream → AllWindowedStream
    16.         WindowAll(env);
    17.  
    18.         // 3.触发程序执行
    19.         env.execute();
    20.     }
    21.  
    22.     // TODO Keyed Windows
    23.     private static void Window(StreamExecutionEnvironment env) {
    24.         env
    25.                 .socketTextStream("localhost", 9999)
    26.                 .map(new MapFunction>() {
    27.                          @Override
    28.                          public Tuple2 map(String value) throws Exception {
    29.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                          }
    31.                      }
    32.                 )
    33.                 .assignTimestampsAndWatermarks(
    34.                         WatermarkStrategy
    35.                                 .>forMonotonousTimestamps()
    36.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    37.                 )
    38.                 .keyBy(value -> value.f0)
    39.                 // 滚动窗口，窗口长度5s
    40.                 .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    41.                 .process(new ShowProcessWindowFunction())
    42.                 .print()
    43.         ;
    44.     }
    45.  
    46.     // TODO Non-Keyed Windows
    47.     private static void WindowAll(StreamExecutionEnvironment env) {
    48.         env
    49.                 .socketTextStream("localhost", 9999)
    50.                 .map(new MapFunction>() {
    51.                          @Override
    52.                          public Tuple2 map(String value) throws Exception {
    53.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    54.                          }
    55.                      }
    56.                 )
    57.                 .assignTimestampsAndWatermarks(
    58.                         WatermarkStrategy
    59.                                 .>forMonotonousTimestamps()
    60.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    61.                 )
    62.                 .keyBy(value -> value.f0)
    63.                 // 滚动窗口，窗口长度5s
    64.                 .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
    65.                 .process(new ShowProcessAllWindowFunction())
    66.                 .print()
    67.         ;
    68.     }
    69. }

    4.6、基于事件时间的会话窗口

    代码示例：

    1. /*
    2.  * TODO 基于会话时间的会话窗口
    3.  *    相邻两个元素的处理时间间隔 大于指定会话周期 触发窗口计算
    4.  * */
    5. public class EventTimeSessionWindow {
    6.     public static void main(String[] args) throws Exception {
    7.         // 1.获取执行环境
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         env.setParallelism(1);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         // TODO Window：KeyedStream → WindowedStream
    13.         // Window(env);
    14.  
    15.         // TODO WindowAll：DataStream → AllWindowedStream
    16.         WindowAll(env);
    17.  
    18.         // 3.触发程序执行
    19.         env.execute();
    20.     }
    21.  
    22.     // TODO Keyed Windows
    23.     private static void Window(StreamExecutionEnvironment env) {
    24.         env
    25.                 .socketTextStream("localhost", 9999)
    26.                 .map(new MapFunction>() {
    27.                          @Override
    28.                          public Tuple2 map(String value) throws Exception {
    29.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                          }
    31.                      }
    32.                 )
    33.                 .assignTimestampsAndWatermarks(
    34.                         WatermarkStrategy
    35.                                 .>forMonotonousTimestamps()
    36.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    37.                 )
    38.                 .keyBy(value -> value.f0)
    39.                 // 会话窗口，超时间隔5s
    40.                 .window(EventTimeSessionWindows.withGap(Time.seconds(5)))
    41.                 .process(new ShowProcessWindowFunction())
    42.                 .print()
    43.         ;
    44.     }
    45.  
    46.     // TODO Non-Keyed Windows
    47.     private static void WindowAll(StreamExecutionEnvironment env) {
    48.         env
    49.                 .socketTextStream("localhost", 9999)
    50.                 .map(new MapFunction>() {
    51.                          @Override
    52.                          public Tuple2 map(String value) throws Exception {
    53.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    54.                          }
    55.                      }
    56.                 )
    57.                 .assignTimestampsAndWatermarks(
    58.                         WatermarkStrategy
    59.                                 .>forMonotonousTimestamps()
    60.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    61.                 )
    62.                 // 会话窗口，超时间隔5s
    63.                 .windowAll(EventTimeSessionWindows.withGap(Time.seconds(5)))
    64.                 .process(new ShowProcessAllWindowFunction())
    65.                 .print()
    66.         ;
    67.     }
    68. }

    4.7、计数窗口

    代码示例：

    1. // TODO 计数窗口
    2. public class countWindow {
    3.     public static void main(String[] args) throws Exception {
    4.         // 1.获取执行环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(3);
    7.  
    8.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    9.         // TODO Window：KeyedStream → WindowedStream
    10.         // Window(env);
    11.  
    12.         // TODO countWindowAll：DataStream → AllWindowedStream
    13.          countWindowAll(env);
    14.  
    15.         // 3.触发程序执行
    16.         env.execute();
    17.     }
    18.  
    19.     // TODO Keyed Windows
    20.     private static void Window(StreamExecutionEnvironment env) {
    21.         env
    22.                 .socketTextStream("localhost", 9999)
    23.                 .map(new MapFunction>() {
    24.                          @Override
    25.                          public Tuple2 map(String value) throws Exception {
    26.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    27.                          }
    28.                      }
    29.                 )
    30.                 .keyBy(value -> value.f0)
    31.                 // TODO 滚动窗口，窗口长度=5个元素
    32.                 //.countWindow(5)
    33.                 // TODO 滑动窗口，窗口长度=5个元素，滑动步长=2个元素 (每当进入两个数据，就统计下窗口内最五个数据)
    34.                 .countWindow(5,2)
    35.                 .process(
    36.                         new ProcessWindowFunction, String, String, GlobalWindow>() {
    37.                             @Override
    38.                             public void process(String s, ProcessWindowFunction, String, String, GlobalWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    39.                                 // 当前水位线
    40.                                 long watermark = context.currentWatermark();
    41.                                 // 当前处理时间
    42.                                 long processingTime = context.currentProcessingTime();
    43.                                 // 窗口开始时间
    44.  
    45.                                 // 窗口结束时间
    46.  
    47.                                 // 计算窗口内数据数量
    48.                                 long count = elements.spliterator().estimateSize();
    49.  
    50.                                 String record = "key=" + s
    51.                                         + " 包含" + count + "条数据===>" + elements.toString()
    52.                                         + " 当前Watermark：" + watermark
    53.                                         + " 当前processingTime：" + processingTime;
    54.  
    55.                                 out.collect(record);
    56.                             }
    57.                         }
    58.                 )
    59.                 .print()
    60.         ;
    61.     }
    62.  
    63.     // TODO Non-Keyed Windows
    64.     private static void countWindowAll(StreamExecutionEnvironment env) {
    65.         env
    66.                 .socketTextStream("localhost", 9999)
    67.                 .map(new MapFunction>() {
    68.                          @Override
    69.                          public Tuple2 map(String value) throws Exception {
    70.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    71.                          }
    72.                      }
    73.                 )
    74.                 // TODO 滚动窗口，窗口长度=5个元素
    75.                 .countWindowAll(5)
    76.                 // TODO 滑动窗口，窗口长度=5个元素，滑动步长=2个元素 (每当进入两个数据，就统计下窗口内最近的五个数据)
    77.                 //.countWindowAll(5,2)
    78.                 .process(
    79.                         new ProcessAllWindowFunction, String, GlobalWindow>() {
    80.                             @Override
    81.                             public void process(ProcessAllWindowFunction, String, GlobalWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    82.                                 // 当前水位线
    83.  
    84.                                 // 当前处理时间
    85.  
    86.                                 // 窗口开始时间
    87.  
    88.                                 // 窗口结束时间
    89.  
    90.                                 // 计算窗口内数据数量
    91.                                 long count = elements.spliterator().estimateSize();
    92.  
    93.                                 String record = "窗口包含" + count + "条数据===>" + elements.toString();
    94.  
    95.                                 out.collect(record);
    96.                             }
    97.                         }
    98.                 )
    99.                 .print()
    100.         ;
    101.     }
    102.  
    103. }

    5、怎样使用 Flink中的 Window Funcation

    窗口函数的作用：

            窗口函数定义了当窗口触发后，对窗口内数据的计算逻辑

    窗口函数的分类：

    5.1、ReduceFunction

    函数功能：

            将两条数据合并成一条数据，输入和输出的数据类型必须相同

    代码示例：

    1. /*
    2.  * TODO 增量聚合函数：ReduceFunction
    3.  *  特点：
    4.  *      1.增量聚合：进入窗口一条数据，就会被计算一次(调用reduce方法)，但是不会立刻输出(只会更新状态)
    5.  *      2.第一条数据进入窗口后，不会调用 reduce 方法
    6.  *      3.数据类型不能改变：输入数据类型 = 输出数据类型 = `中间累加器数据类型`
    7.  *      4.窗口触发计算时，才会输出窗口的计算结果
    8.  * */
    9. public class ReduceFunctions {
    10.     public static void main(String[] args) throws Exception {
    11.         // 1.获取执行环境
    12.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    13.         env.setParallelism(3);
    14.  
    15.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    16.         env
    17.                 .socketTextStream("localhost", 9999)
    18.                 .map(new MapFunction>() {
    19.                          @Override
    20.                          public Tuple2 map(String value) throws Exception {
    21.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    22.                          }
    23.                      }
    24.                 )
    25.                 .keyBy(value -> value.f0)
    26.                 // TODO 指定窗口类型：滚动计数窗口，窗口长度=5个元素
    27.                 .countWindow(5)
    28.                 // TODO 对窗口内的元素求和
    29.                 .reduce(
    30.                         new ReduceFunction>() {
    31.  
    32.                             /**
    33.                              * @param value1 参与聚合的第一个值，也就是累加器的值
    34.                              * @param value2 参与聚合的第二个值，也就是新进入窗口的时间数据
    35.                              * @return
    36.                              * @throws Exception
    37.                              */
    38.                             @Override
    39.                             public Tuple2 reduce(Tuple2 value1, Tuple2 value2) throws Exception {
    40.                                 System.out.println("触发计算：" + value1 + " ，" + value2);
    41.                                 long sum = value1.f1 + value2.f1;
    42.                                 return new Tuple2<>(value1.f0, sum);
    43.                             }
    44.                         }
    45.                 )
    46.                 .print()
    47.         ;
    48.  
    49.         // 3.触发程序执行
    50.         env.execute();
    51.     }
    52.  
    53. }

    5.2、AggregateFunction

    函数功能：

            将两条数据合并成一条数据，输入和输出的数据类型可以不同

    代码示例：

    1. /*
    2.  * TODO 增量聚合函数：AggregateFunction
    3.  *  特点：
    4.  *      1.增量聚合：进入窗口一条数据，就会被计算一次(调用add方法)，但是不会立刻输出(只会更累加器的状态)
    5.  *      2.窗口内的第一条数据进入后，会创建窗口，创建累加器并初始化
    6.  *      3.窗口触发计算时，才会输出窗口的计算结果(调用getResult方法)
    7.  *      4.数据类型可以不同：输入数据类型、输出数据类型、累加器数据类型
    8.  *  泛型参数：
    9.  *      AggregateFunction
    10.  *      @IN    ：  输入数据类型
    11.  *      @ACC   ：  累加器数据类型
    12.  *      @OUT   ：  输出数据类型
    13.  *
    14.  * */
    15. public class AggregateFunctions {
    16.     public static void main(String[] args) throws Exception {
    17.         // 1.获取执行环境
    18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    19.         env.setParallelism(3);
    20.  
    21.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    22.         env
    23.                 .socketTextStream("localhost", 9999)
    24.                 .map(new MapFunction>() {
    25.                          @Override
    26.                          public Tuple2 map(String value) throws Exception {
    27.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    28.                          }
    29.                      }
    30.                 )
    31.                 .keyBy(value -> value.f0)
    32.                 // TODO 指定窗口类型：滚动计数窗口，窗口长度=5个元素
    33.                 .countWindow(5)
    34.                 // TODO 对窗口内的元素求和
    35.                 .aggregate(
    36.                         new AggregateFunction, Integer, String>() {
    37.                             /**
    38.                              * 创建累加器，并对累加器做初始化操作
    39.                              * @return
    40.                              */
    41.                             @Override
    42.                             public Integer createAccumulator() {
    43.                                 System.out.println("创建累加器");
    44.                                 return 0;
    45.                             }
    46.  
    47.                             /**
    48.                              * 事件数据与累加器的Merge逻辑，用来更新累加器状态
    49.                              * 进入一条数据，调用一次
    50.                              * @param value       The value to add
    51.                              * @param accumulator The accumulator to add the value to
    52.                              * @return
    53.                              */
    54.                             @Override
    55.                             public Integer add(Tuple2 value, Integer accumulator) {
    56.                                 System.out.println("调用add方法，value=" + value + " 当前累加器：" + accumulator);
    57.                                 return accumulator + 1;
    58.                             }
    59.  
    60.                             /**
    61.                              * 获取累加器的状态值，窗口触发时调用
    62.                              * @param accumulator The accumulator of the aggregation
    63.                              * @return
    64.                              */
    65.                             @Override
    66.                             public String getResult(Integer accumulator) {
    67.                                 System.out.println("调用getResult 方法");
    68.                                 return accumulator.toString();
    69.                             }
    70.  
    71.                             /**
    72.                              * 合并窗口逻辑(只有sessionWindow才会用的)
    73.                              * @param a An accumulator to merge
    74.                              * @param b Another accumulator to merge
    75.                              * @return
    76.                              */
    77.                             @Override
    78.                             public Integer merge(Integer a, Integer b) {
    79.                                 System.out.println("调用merge方法");
    80.                                 return null;
    81.                             }
    82.                         }
    83.                 )
    84.                 .print()
    85.         ;
    86.  
    87.         // 3.触发程序执行
    88.         env.execute();
    89.     }
    90.  
    91. }

    5.3、ProcessWindowFunction

    函数功能：

            将窗口内所有的数据缓存到Iterable，在窗口触发后对所有数据进行计算

    代码示例：

    1. /*
    2.  * TODO 全窗口函数：ProcessWindowFunction
    3.  *  特点：
    4.  *      1.窗口触发时才会调用一次process方法，对窗口内的数据统一计算
    5.  *  泛型参数说明：
    6.  *      ProcessWindowFunction
    7.  *      @IN     ：  输入数据类型
    8.  *      @OUT    ：  输出数据类型
    9.  *      @KEY    ：  key的数据类型
    10.  *      @W      ：  window类型(时间窗口、计数窗口)
    11.  *  上下文对象说明：
    12.  *      时间信息：currentProcessingTime()、currentWatermark()
    13.  *      状态信息：windowState()、globalState()
    14.  *      侧输出流：
    15.  *      窗口信息：window()
    16.  *  重点说明：
    17.  *      由于WindowFunction会存储窗口内的所有数据，当窗口内数量特别大时，慎用
    18.  *  思考：
    19.  *      1.什么时候需要使用全窗口函数呢？
    20.  *          计算平均数、计算中位数
    21.  * */
    22. public class ProcessWindowFunctions {
    23.     public static void main(String[] args) throws Exception {
    24.         // 1.获取执行环境
    25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    26.         env.setParallelism(3);
    27.  
    28.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    29.         // TODO 基于处理时间的 滚动时间窗口，窗口长度10秒
    30.         timewindow(env);
    31.  
    32.         // TODO 计数窗口
    33.         //countwindow(env);
    34.  
    35.         // 3.触发程序执行
    36.         env.execute();
    37.     }
    38.  
    39.     private static void timewindow(StreamExecutionEnvironment env) {
    40.         env
    41.                 .socketTextStream("localhost", 9999)
    42.                 .map(new MapFunction>() {
    43.                          @Override
    44.                          public Tuple2 map(String value) throws Exception {
    45.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    46.                          }
    47.                      }
    48.                 )
    49.                 .keyBy(value -> value.f0)
    50.                 // TODO 指定窗口类型：基于处理时间的 滚动时间窗口，窗口长度5秒
    51.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    52.                 // TODO 对窗口内的元素求和
    53.                 .process(
    54.                         new ProcessWindowFunction, String, String, TimeWindow>() {
    55.  
    56.                             /**
    57.                              * @param s        窗口中所属的key
    58.                              * @param context  上下文对象
    59.                              * @param elements 窗口中存储的数据
    60.                              * @param out      采集器，用来向下游发送数据
    61.                              * @throws Exception
    62.                              */
    63.                             @Override
    64.                             public void process(String s, ProcessWindowFunction, String, String, TimeWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    65.                                 // 通过窗口对象，获取窗口的范围信息
    66.                                 long startTs = context.window().getStart();
    67.                                 long endTs = context.window().getEnd();
    68.                                 String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
    69.                                 String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
    70.  
    71.                                 long count = elements.spliterator().estimateSize();
    72.  
    73.                                 // 当前处理数据
    74.                                 long currentProcessingTime = context.currentProcessingTime();
    75.                                 String currentProcessingTimeS = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
    76.  
    77.                                 // 当前水位线
    78.                                 long currentWatermark = context.currentWatermark();
    79.  
    80.                                 out.collect("key=" + s + "，的窗口[" + windowStart + "," + windowEnd + ") 包含" + count + "条数据 ===> " + elements.toString() + " 当前处理时间：" + currentProcessingTimeS);
    81.  
    82.                             }
    83.                         }
    84.                 )
    85.                 .print()
    86.         ;
    87.  
    88.     }
    89.  
    90.     private static void countwindow(StreamExecutionEnvironment env) {
    91.         env
    92.                 .socketTextStream("localhost", 9999)
    93.                 .map(new MapFunction>() {
    94.                          @Override
    95.                          public Tuple2 map(String value) throws Exception {
    96.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    97.                          }
    98.                      }
    99.                 )
    100.                 .keyBy(value -> value.f0)
    101.                 // TODO 指定窗口类型：计数窗口，窗口长度为5
    102.                 .countWindow(5)
    103.                 // TODO 对窗口内的元素求和
    104.                 .process(
    105.                         new ProcessWindowFunction, String, String, GlobalWindow>() {
    106.  
    107.  
    108.                             /**
    109.                              * @param s        窗口中所属的key
    110.                              * @param context  上下文对象
    111.                              * @param elements 窗口中存储的数据
    112.                              * @param out      采集器，用来向下游发送数据
    113.                              * @throws Exception
    114.                              */
    115.                             @Override
    116.                             public void process(String s, ProcessWindowFunction, String, String, GlobalWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    117.                                 // 通过窗口对象，获取窗口的范围信息
    118.  
    119.                                 long count = elements.spliterator().estimateSize();
    120.  
    121.                                 // 当前处理数据
    122.                                 long currentProcessingTime = context.currentProcessingTime();
    123.                                 String currentProcessingTimeS = DateFormatUtils.format(currentProcessingTime, "yyyy-MM-dd HH:mm:ss.SSS");
    124.  
    125.                                 // 当前水位线
    126.                                 long currentWatermark = context.currentWatermark();
    127.  
    128.                                 out.collect("key=" + s + "，的窗口 包含" + count + "条数据 ===> " + elements.toString() + " 当前处理时间：" + currentProcessingTimeS);
    129.  
    130.                             }
    131.                         }
    132.                 )
    133.                 .print()
    134.         ;
    135.  
    136.     }
    137. }

    5.4、增量聚合的 ProcessWindowFunction

    函数功能：

            既可以使用 ReduceFunction 或 AggregateFunction 增量聚合功能

            又可以使用 ProcessWindowFunction 中的元数据信息(窗口信息、水位线信息)

    代码示例：

    1. /*
    2.  * TODO 全窗口函数：ProcessWindowFunction
    3.  *  特点：
    4.  *      1.窗口触发时才会调用一次process方法，对窗口内的数据统一计算
    5.  *  泛型参数说明：
    6.  *      ProcessWindowFunction
    7.  *      @IN     ：  输入数据类型
    8.  *      @OUT    ：  输出数据类型
    9.  *      @KEY    ：  key的数据类型
    10.  *      @W      ：  window类型(时间窗口、计数窗口)
    11.  *  上下文对象说明：
    12.  *      时间信息：currentProcessingTime()、currentWatermark()
    13.  *      状态信息：windowState()、globalState()
    14.  *      侧输出流：
    15.  *      窗口信息：window()
    16.  *  重点说明：
    17.  *      由于WindowFunction会存储窗口内的所有数据，当窗口内数量特别大时，慎用
    18.  *  思考：
    19.  *      1.什么时候需要使用全窗口函数呢？
    20.  *      计算平均数、计算中位数
    21.  * */
    22. public class ReduceAggredateProcessWindowFunctions {
    23.     public static void main(String[] args) throws Exception {
    24.         // 1.获取执行环境
    25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    26.         env.setParallelism(3);
    27.  
    28.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    29.         // TODO 基于处理时间的 滚动时间窗口，窗口长度10秒
    30.         // TODO 使用 ReduceFunction + ProcessWindowFunction 增量聚合
    31.         //reduceAndProcessWindowFunction(env);
    32.  
    33.         // TODO 使用 AggregateFunction + ProcessWindowFunction 增量聚合
    34.         aggreateAndProcessWindowFunction(env);
    35.  
    36.         // 3.触发程序执行
    37.         env.execute();
    38.     }
    39.  
    40.     private static void reduceAndProcessWindowFunction(StreamExecutionEnvironment env) {
    41.         env
    42.                 .socketTextStream("localhost", 9999)
    43.                 .map(new MapFunction>() {
    44.                          @Override
    45.                          public Tuple2 map(String value) throws Exception {
    46.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    47.                          }
    48.                      }
    49.                 )
    50.                 .keyBy(value -> value.f0)
    51.                 // TODO 指定窗口类型：基于处理时间的 滚动时间窗口，窗口长度5秒
    52.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    53.                 // TODO 获取窗口中的最小元素和窗口的开始时间、结束时间
    54.                 .reduce(
    55.                         new ReduceFunction>() {
    56.  
    57.                             /**
    58.                              * @param value1 参与聚合的第一个值，也就是累加器的值
    59.                              * @param value2 参与聚合的第二个值，也就是新进入窗口的时间数据
    60.                              * @return
    61.                              * @throws Exception
    62.                              */
    63.                             @Override
    64.                             public Tuple2 reduce(Tuple2 value1, Tuple2 value2) throws Exception {
    65.                                 System.out.println("触发计算：" + value1 + " ，" + value2);
    66.                                 long min = Math.min(value1.f1, value2.f1);
    67.  
    68.                                 return new Tuple2<>(value1.f0, min);
    69.                             }
    70.                         }
    71.                         , new ProcessWindowFunction, String, String, TimeWindow>() {
    72.  
    73.                             /**
    74.                              * @param s        窗口中所属的key
    75.                              * @param context  上下文对象
    76.                              * @param elements 窗口中存储的数据
    77.                              * @param out      采集器，用来向下游发送数据
    78.                              * @throws Exception
    79.                              */
    80.                             @Override
    81.                             public void process(String s, ProcessWindowFunction, String, String, TimeWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    82.                                 // 通过窗口对象，获取窗口的范围信息
    83.                                 long startTs = context.window().getStart();
    84.                                 long endTs = context.window().getEnd();
    85.                                 String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
    86.                                 String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
    87.  
    88.                                 long count = elements.spliterator().estimateSize();
    89.  
    90.                                 // 当前处理数据
    91.                                 long currentProcessingTime = context.currentProcessingTime();
    92.                                 String currentProcessingTimeS = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
    93.  
    94.                                 // 当前水位线
    95.                                 long currentWatermark = context.currentWatermark();
    96.  
    97.                                 out.collect("key=" + s + "，的窗口[" + windowStart + "," + windowEnd + ") 包含" + count + "条数据 ===> " + elements.toString() + " 当前处理时间：" + currentProcessingTimeS);
    98.  
    99.                             }
    100.                         }
    101.                 )
    102.                 .print()
    103.         ;
    104.  
    105.     }
    106.  
    107.     private static void aggreateAndProcessWindowFunction(StreamExecutionEnvironment env) {
    108.         env
    109.                 .socketTextStream("localhost", 9999)
    110.                 .map(new MapFunction>() {
    111.                          @Override
    112.                          public Tuple2 map(String value) throws Exception {
    113.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    114.                          }
    115.                      }
    116.                 )
    117.                 .keyBy(value -> value.f0)
    118.                 // TODO 指定窗口类型：基于处理时间的 滚动时间窗口，窗口长度5秒
    119.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    120.                 // TODO 计算窗口内数据平均值并与窗口对应的key一同输出
    121.                 .aggregate(
    122.                         new AggregateFunction, Tuple2, Double>() {
    123.  
    124.                             /**
    125.                              * @return
    126.                              */
    127.                             @Override
    128.                             public Tuple2 createAccumulator() {
    129.                                 return new Tuple2(0L, 0L);
    130.                             }
    131.  
    132.                             /**
    133.                              * @param value       The value to add
    134.                              * @param accumulator The accumulator to add the value to
    135.                              * @return
    136.                              */
    137.                             @Override
    138.                             public Tuple2 add(Tuple2 value, Tuple2 accumulator) {
    139.                                 return new Tuple2(accumulator.f0 + value.f1, accumulator.f1 + 1L);
    140.                             }
    141.  
    142.                             /**
    143.                              * @param accumulator The accumulator of the aggregation
    144.                              * @return
    145.                              */
    146.                             @Override
    147.                             public Double getResult(Tuple2 accumulator) {
    148.                                 return (double) (accumulator.f0 * 1.0000 / accumulator.f1);
    149.                             }
    150.  
    151.                             /**
    152.                              * @param a An accumulator to merge
    153.                              * @param b Another accumulator to merge
    154.                              * @return
    155.                              */
    156.                             @Override
    157.                             public Tuple2 merge(Tuple2 a, Tuple2 b) {
    158.                                 return null;
    159.                             }
    160.                         }
    161.                         ,new ProcessWindowFunction(){
    162.                             /**
    163.                              * @param s        The key for which this window is evaluated.
    164.                              * @param context  The context in which the window is being evaluated.
    165.                              * @param elements The elements in the window being evaluated.
    166.                              * @param out      A collector for emitting elements.
    167.                              * @throws Exception
    168.                              */
    169.                             @Override
    170.                             public void process(String s, ProcessWindowFunction.Context context, Iterable elements, Collector out) throws Exception {
    171.                                 // 通过窗口对象，获取窗口的范围信息
    172.                                 long startTs = context.window().getStart();
    173.                                 long endTs = context.window().getEnd();
    174.                                 String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
    175.                                 String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
    176.  
    177.                                 long count = elements.spliterator().estimateSize();
    178.  
    179.                                 // 当前处理数据
    180.                                 long currentProcessingTime = context.currentProcessingTime();
    181.                                 String currentProcessingTimeS = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
    182.  
    183.                                 // 当前水位线
    184.                                 long currentWatermark = context.currentWatermark();
    185.  
    186.                                 out.collect("key=" + s + "，的窗口[" + windowStart + "," + windowEnd + ") 包含" + count + "条数据 ===> " + elements.toString() + " 当前处理时间：" + currentProcessingTimeS);
    187.                             }
    188.                         }
    189.                 )
    190.                 .print()
    191.         ;
    192.  
    193.     }
    194. }

    6、Flink中的 Window的生命周期

    窗口什么时候创建？

            当窗口所属的第一条数据到达时，窗口会被创建

    窗口什么时候删除？

            当 水位线 或者 processing time 超过窗口的结束时间戳 + allowedLateness时，窗口会被删除

    窗口什么时候被触发计算？

            当定义的触发器触发时， window function 会处理窗口内的数据

            默认触发器：

                    当 水位线 或者 processing time 超过窗口的结束时间戳时，窗口会被计算

    6.1、触发器 - Triggers

    触发器的作用：

            Trigger 决定了一个窗口（由 window assigner 定义）何时可以被 window function 处理

    内置 Trigger：

    自定义 Triggers：

    1. Trigger 接口：
    2.  
    3.     // 每个元素被加入窗口时调用
    4.     @Override
    5.     public TriggerResult onElement(Tuple2 element, long timestamp, GlobalWindow window, TriggerContext ctx) 
    6.  
    7.     // 在注册的 event-time timer 触发时调用
    8.     @Override
    9.     public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx)
    10.  
    11.     // 在注册的 processing-time timer 触发时调用
    12.     @Override
    13.     public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) 
    14.  
    15.     // 对应窗口被移除时所需的逻辑
    16.     @Override
    17.     public void clear(GlobalWindow window, TriggerContext ctx) 
    18.  
    19.     TriggerResult:
    20.         CONTINUE: 什么也不做
    21.         FIRE: 触发计算
    22.         PURGE: 清空窗口内的元素
    23.         FIRE_AND_PURGE: 触发计算，计算结束后清空窗口内的元素

    代码示例：

    1. // TODO 自定义触发器
    2. public class Triggers {
    3.     public static void main(String[] args) throws Exception {
    4.         // 1.获取执行环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(3);
    7.  
    8.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    9.         // TODO Window：KeyedStream → WindowedStream
    10.         // countWindow(env);
    11.  
    12.         // TODO countWindowAll：DataStream → AllWindowedStream
    13.         timeWindow(env);
    14.  
    15.         // 3.触发程序执行
    16.         env.execute();
    17.     }
    18.  
    19.     // TODO 基于 计数窗口 的触发器
    20.     private static void countWindow(StreamExecutionEnvironment env) {
    21.         env
    22.                 .socketTextStream("localhost", 9999)
    23.                 .map(new MapFunction>() {
    24.                          @Override
    25.                          public Tuple2 map(String value) throws Exception {
    26.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    27.                          }
    28.                      }
    29.                 )
    30.                 .keyBy(value -> value.f0)
    31.                 // TODO 滚动窗口，窗口长度=5个元素
    32.                 //.countWindow(5)
    33.                 // TODO 滑动窗口，窗口长度=5个元素，滑动步长=2个元素 (每当进入两个数据，就统计下窗口内最五个数据)
    34.                 .countWindow(5, 2)
    35.                 .trigger(
    36.                         new Trigger, GlobalWindow>() {
    37.                             @Override
    38.                             public TriggerResult onElement(Tuple2 element, long timestamp, GlobalWindow window, TriggerContext ctx) throws Exception {
    39.                                 System.out.println("调用onElement方法");
    40.                                 if (element.f1 == 999) {
    41.                                     return TriggerResult.FIRE;
    42.                                 } else {
    43.                                     return TriggerResult.CONTINUE;
    44.                                 }
    45.                             }
    46.  
    47.                             @Override
    48.                             public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    49.                                 System.out.println("调用onProcessingTime方法");
    50.                                 return null;
    51.                             }
    52.  
    53.                             @Override
    54.                             public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) throws Exception {
    55.                                 System.out.println("调用onEventTime方法");
    56.                                 return null;
    57.                             }
    58.  
    59.                             @Override
    60.                             public void clear(GlobalWindow window, TriggerContext ctx) throws Exception {
    61.                                 System.out.println("调用clear方法");
    62.                             }
    63.                         }
    64.                 )
    65.                 .process(
    66.                         new ProcessWindowFunction, String, String, GlobalWindow>() {
    67.                             @Override
    68.                             public void process(String s, ProcessWindowFunction, String, String, GlobalWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    69.                                 // 当前水位线
    70.                                 long watermark = context.currentWatermark();
    71.                                 // 当前处理时间
    72.                                 long processingTime = context.currentProcessingTime();
    73.                                 // 窗口开始时间
    74.  
    75.                                 // 窗口结束时间
    76.  
    77.                                 // 计算窗口内数据数量
    78.                                 long count = elements.spliterator().estimateSize();
    79.  
    80.                                 String record = "key=" + s
    81.                                         + " 包含" + count + "条数据===>" + elements.toString()
    82.                                         + " 当前Watermark：" + watermark
    83.                                         + " 当前processingTime：" + processingTime;
    84.  
    85.                                 out.collect(record);
    86.                             }
    87.                         }
    88.                 )
    89.                 .print()
    90.         ;
    91.     }
    92.  
    93.     // TODO 基于 处理时间窗口 的触发器
    94.     private static void timeWindow(StreamExecutionEnvironment env) {
    95.         env
    96.                 .socketTextStream("localhost", 9999)
    97.                 .map(new MapFunction>() {
    98.                          @Override
    99.                          public Tuple2 map(String value) throws Exception {
    100.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    101.                          }
    102.                      }
    103.                 )
    104.                 .keyBy(value -> value.f0)
    105.                 // 滚动窗口，窗口长度5s
    106.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    107.                 .trigger(
    108.                         new Trigger, TimeWindow>() {
    109.  
    110.                             @Override
    111.                             public TriggerResult onElement(Tuple2 element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
    112.                                 System.out.println("调用onElement方法");
    113.                                 return TriggerResult.CONTINUE;
    114.                             }
    115.  
    116.                             @Override
    117.                             public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
    118.                                 System.out.println("调用onProcessingTime方法");
    119.                                 return TriggerResult.FIRE;
    120.                             }
    121.  
    122.                             @Override
    123.                             public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
    124.                                 System.out.println("调用onEventTime方法");
    125.                                 return TriggerResult.CONTINUE;
    126.                             }
    127.                             
    128.                             @Override
    129.                             public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
    130.                                 System.out.println("调用clear方法");
    131.                             }
    132.                         }
    133.                 )
    134.                 .process(new ShowProcessWindowFunction())
    135.                 .print()
    136.         ;
    137.     }
    138. }

    6.2、移除器 - Evictors

    移除器的作用：

            Evictor 可以在 trigger 触发后、调用窗口函数之前或之后从窗口中删除元素

    内置 Evictors ：

            CountEvictor(元素个数) ：一旦窗口中的元素超过这个数量，多余的元素会从窗口缓存的开头移除

            DeltaEvictor ：接收 DeltaFunction 和 threshold 参数，计算最后一个元素与窗口缓存中所有元素的差值， 并移除差值大于或等于 threshold 的元素

            TimeEvictor ：接收 interval 参数，以毫秒表示。 它会找到窗口中元素的最大 timestamp max_ts 并移除比 max_ts - interval 小的所有元素

    代码示例：

    1. public class Evictors {
    2.     public static void main(String[] args) throws Exception {
    3.         // 1.获取执行环境
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(3);
    6.  
    7.         env
    8.                 .socketTextStream("localhost", 9999)
    9.                 .map(new MapFunction>() {
    10.                          @Override
    11.                          public Tuple2 map(String value) throws Exception {
    12.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    13.                          }
    14.                      }
    15.                 )
    16.                 .keyBy(value -> value.f0)
    17.                 // TODO 滚动窗口，窗口长度=5个元素
    18.                 .countWindow(5)
    19.                 .evictor(CountEvictor.of(2L))
    20.                 .process(
    21.                         new ProcessWindowFunction, String, String, GlobalWindow>() {
    22.                             @Override
    23.                             public void process(String s, ProcessWindowFunction, String, String, GlobalWindow>.Context context, Iterable> elements, Collector out) throws Exception {
    24.                                 // 当前水位线
    25.                                 long watermark = context.currentWatermark();
    26.                                 // 当前处理时间
    27.                                 long processingTime = context.currentProcessingTime();
    28.                                 // 窗口开始时间
    29.  
    30.                                 // 窗口结束时间
    31.  
    32.                                 // 计算窗口内数据数量
    33.                                 long count = elements.spliterator().estimateSize();
    34.  
    35.                                 String record = "key=" + s
    36.                                         + " 包含" + count + "条数据===>" + elements.toString()
    37.                                         + " 当前Watermark：" + watermark
    38.                                         + " 当前processingTime：" + processingTime;
    39.  
    40.                                 out.collect(record);
    41.                             }
    42.                         }
    43.                 )
    44.                 .print()
    45.         ;
    46.  
    47.         // 3.触发程序执行
    48.         env.execute();
    49.     }
    50. }

    7、对迟到数据的处理

            在使用 event-time 窗口时，数据可能会迟到，默认情况下，watermark 一旦越过窗口结束的 timestamp，迟到的数据就会被直接丢弃，Flink中提供多种多处理迟到数据的策略。

    7.1、生成水位线时，设置最大乱序时间

            在生成 watermark 时，设置一个可容忍的最大乱序时间，保证窗口计算被延迟执行，保证更多的迟到的数据能够进入窗口

            注意：迟到的数据超过了设置的最大乱序时间，将会被丢弃

    1. WatermarkStrategy
    2.                 .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) // 设置最大乱序时间为1s
    3.                 .withTimestampAssigner((element,recordTimestamp) -> element.f1);

    7.2、设置窗口延迟关闭 - allowedLateness

    可以通过 window.allowedLateness 来设置窗口延迟关闭的时间

    allowedLateness 默认为0

            表示 当watermark 或 processing time 超过 窗口结束的timestamp时，触发计算 并销毁窗口

    allowedLateness 大于0时

            表示  当watermark 或 processing time 超过 窗口结束的timestamp时，会触发计算 但是并不会销毁窗口
            而是当 窗口结束的timestamp + allowedLateness 的数据到来后，才会销毁窗口
            并且 每次接收到迟到的数据后 都会触发窗口计算

    代码示例：

    1. public class AllowedLateness {
    2.     public static void main(String[] args) throws Exception {
    3.         // 1.获取执行环境
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(1);
    6.  
    7.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    8.         env
    9.                 .socketTextStream("localhost", 9999)
    10.                 .map(new MapFunction>() {
    11.                          @Override
    12.                          public Tuple2 map(String value) throws Exception {
    13.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    14.                          }
    15.                      }
    16.                 )
    17.                 .assignTimestampsAndWatermarks(
    18.                         WatermarkStrategy
    19.                                 .>forMonotonousTimestamps()
    20.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    21.                 )
    22.                 .keyBy(value -> value.f0)
    23.                 // 滚动窗口，窗口长度5s
    24.                 .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    25.                 // 窗口延迟10s关闭
    26.                 .allowedLateness(Time.seconds(10))
    27.                 .process(new ShowProcessWindowFunction())
    28.                 .print()
    29.         ;
    30.  
    31.         // 3.触发程序执行
    32.         env.execute();
    33.     }
    34. }

    运行结果：

    7.3、使用侧输出流获取迟到的数据 - sideOutputLateData

    在 Flink中可以使用 侧输出流-OutputTag 来获取窗口中迟到的数据

    代码示例：

    1. // TODO 使用侧流接收迟到的数据
    2. public class UseSideOutputReceiveLatedata {
    3.     public static void main(String[] args) throws Exception {
    4.         // 1.获取执行环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(1);
    7.  
    8.         // 声明 OutputTag 对象,用来接收迟到的数据
    9.         OutputTag> outputTag = new OutputTag>("side-output"){};
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         SingleOutputStreamOperator processDataStream = env
    13.                 .socketTextStream("localhost", 9999)
    14.                 .map(new MapFunction>() {
    15.                          @Override
    16.                          public Tuple2 map(String value) throws Exception {
    17.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    18.                          }
    19.                      }
    20.                 )
    21.                 .assignTimestampsAndWatermarks(
    22.                         WatermarkStrategy
    23.                                 .>forMonotonousTimestamps()
    24.                                 .withTimestampAssigner((element, recordTimestamp) -> element.f1)
    25.                 )
    26.                 .keyBy(value -> value.f0)
    27.                 // 滚动窗口，窗口长度5s
    28.                 .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    29.                 .sideOutputLateData(outputTag)
    30.                 .process(new ShowProcessWindowFunction());
    31.  
    32.         // 输出主流
    33.         processDataStream.print("主流输出");
    34.  
    35.         // 从主流获取侧输出流，打印迟到的数据
    36.         processDataStream.getSideOutput(outputTag).printToErr("关窗后的迟到数据");
    37.  
    38.         // 3.触发程序执行
    39.         env.execute();
    40.     }
    41. }

    运行结果：

     

  • 相关阅读:
    判断当前shell版本
    TBarCode SDK 11.15.1 Crack
    k8s笔记资源限制，亲和和性 污点和容忍
    el-popover和el-tooltip样式修改(普通的组件样式修改方法，对popover是不生效的)
    解决Windows 10更新安装失败的问题
    TS以及webpack的es module
    Pytest教程：Fixture详解
    如何在 Ubuntu 22.04.1 上安装并使用 PostgreSQL 14.5？
    开源多系统启动工具-ventoy
    「实用技巧」后端如何使用 Eolink Apikit 快速调试接口？
  • 原文地址：https://blog.csdn.net/weixin_42845827/article/details/131541254