Flink中的Window API

r囧r小猫 2023-01-16 02:16 203阅读 0赞

## 一、基本概念 ##

### 1.1 Window概念 ###

一般真实的流都是无界的，如何处理这种无界的数据呢？可以把无限的数据流进行切分，得到有限的数据集进行处理（就可以转换成有界流啦）。**窗口（Window）** 就是将无限流切割成有限流的一种方式，它会将流数据分发到**有限大小的桶（bucket）** 中进行分析  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]

### 1.2 Window类型 ###

#### 1.2.1 时间窗口（Time Windows） ####

##### 1.2.1.1 滑动窗口（Sliding Windows） #####

滑动窗口是**固定窗口**的更广义的一种形式，滑动窗口由固定的**窗口长度和滑动间隔**组成。窗口长度固定，可以有**重叠**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]

##### 1.2.1.2 滚动窗口 #####

将数据依据固定的窗口长度对数据进行切分。**时间对齐**，窗口长度固定，**没有重叠**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]

##### 1.2.1.3 会话窗口（Session Windows） #####

由一系列事件组合一个指定时间长度的**timeout间隙**组成，也就是一段时间没有接收到新数据就会生成新的窗口。特点是：**时间无对齐**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]

#### 1.2.2 计数窗口（Count Windows） ####

##### 1.2.2.1 滚动计数窗口 #####

##### 1.2.2.2 滑动计数窗口 #####

### 1.3 Window API ###

可以用 **.window()来定义一个窗口** ，然后基于这个window去做一些聚合或其他处理操作。注意window()方法必须在**keyBy之后**才能用  
Flink提供了更加简单的 **.timeWindow** 和 **.countWindow** 方法，用于定义时间窗口和计数窗口

#### 1.3.1 窗口分配器（window assigner） ####

*  window()方法接收的输入参数是一个WindowAssigner
 *  WindowAssigner负责将每条输入的数据分发到正确的window中
 *  Flink提供了通用的 WindowAssigner  
    （1）滚动窗口（tumbling window）  
    （2）滑动窗口（sliding window）  
    （3）会话窗口（session window）  
    （4）全局窗口（global window）

示例代码：

#### 1.3.2 窗口函数（window function） ####

window function定义了要对窗口中收集的数据做的计算操作

可以分为两类：  
（1）增量聚合函数（incremental aggregation function）

*  每条数据到来就进行计算，保持一个简单的状态
 *  ReduceFunction、AggregateFunction

（2）全窗口函数（full window function）

*  先把窗口所有数据收集起来，等到计算的时候会遍历所有数据
 *  ProcessWindowFunction、WindowFunction

示例代码1：**增量聚合函数**

public class WindowTest1_TimeWindow { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            //开窗测试
            //windowAll()会把所有数据传给同一个下游算子，即把所有数据放到一个窗口里（相当于并行度是1）
            SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
    //            .window(TumblingAlignedProcessingTimeWindows.of(Time.seconds(15))); //方式一：复杂麻烦
                    .timeWindow(Time.seconds(15)) //方式二：滚动窗口，因为没有设置步长；有步长的话就是滑动窗口
    //        .window(EventTimeSessionWindows.withGap(Time.minutes(1))); //session窗口，只有这一种写法
    //        .countWindow(10,2 ); //计数窗口
                    // 以下属于增量聚合
                    .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() { 
                        @Override
                        public Integer createAccumulator() { 
                            return 0;
                        }
    
                        @Override
                        public Integer add(SensorReading sensorReading, Integer integer) { 
                            return integer + 1;
                        }
    
                        @Override
                        public Integer getResult(Integer integer) { 
                            return integer;
                        }
    
                        @Override
                        public Integer merge(Integer integer, Integer acc1) { 
                            return integer + acc1;
                        }
                    });
            resultStream.print();
            env.execute();
        }
    }

示例代码2：**全窗口函数**

public class WindowTest1_TimeWindow { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            //开窗测试
            //windowAll()会把所有数据传给同一个下游算子，即把所有数据放到一个窗口里（相当于并行度是1）
            SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultStream = dataStream.keyBy("id")
    //            .window(TumblingAlignedProcessingTimeWindows.of(Time.seconds(15))); //方式一：复杂麻烦
                    .timeWindow(Time.seconds(15)) //方式二：滚动窗口，因为没有设置步长；有步长的话就是滑动窗口
    //        .window(EventTimeSessionWindows.withGap(Time.minutes(1))); //session窗口，只有这一种写法
    //        .countWindow(10,2 ); //计数窗口
    
            // 全窗口函数
            .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() { 
                @Override
                public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<SensorReading> input, Collector<Tuple3<String, Long, Integer>> out) throws Exception { 
                    String id = tuple.getField(0);
                    Long windowEnd = timeWindow.getEnd();
                    Integer count = IteratorUtils.toList(input.iterator()).size();
                    out.collect(new Tuple3<>(id, windowEnd, count));
                }
            });
    
            resultStream.print();
            env.execute();
        }
    }

示例代码3：

public class WindowTest1_TimeWindow { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            //开窗测试
            //windowAll()会把所有数据传给同一个下游算子，即把所有数据放到一个窗口里（相当于并行度是1）
            SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultStream = dataStream.keyBy("id")
    //            .window(TumblingAlignedProcessingTimeWindows.of(Time.seconds(15))); //方式一：复杂麻烦
                    .timeWindow(Time.seconds(15)) //方式二：滚动窗口，因为没有设置步长；有步长的话就是滑动窗口
    //        .window(EventTimeSessionWindows.withGap(Time.minutes(1))); //session窗口，只有这一种写法
    //        .countWindow(10,2 ); //计数窗口
    // 不完整的ProcessWindowFunction代码
            .process(new ProcessWindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() { 
                // Context中包含Window
                @Override
                public void process(Tuple tuple, Context context, Iterable<SensorReading> iterable, Collector<Tuple3<String, Long, Integer>> collector) throws Exception { 
    
                }
            });
    
            resultStream.print();
            env.execute();
        }
    }

计数窗口测试，示例代码：

public class WindowTest2_CountWindow { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 开计数窗口测试
            SingleOutputStreamOperator<Double> aggregateStream = dataStream.keyBy("id")
                    .countWindow(10, 2) //含义是: 10个数为一个窗口，每两个数滑动一次
                    .aggregate(new MyAvgTemp());
    
            aggregateStream.print();
            env.execute();
        }
    
        // 求平均数
        public static class MyAvgTemp implements AggregateFunction<SensorReading, Tuple2<Double, Integer>, Double>{ 
    
            @Override
            public Tuple2<Double, Integer> createAccumulator() { 
                return new Tuple2<>(0.0, 0);
            }
    
            @Override
            public Tuple2<Double, Integer> add(SensorReading value, Tuple2<Double, Integer> accumulator) { 
                return new Tuple2<>(accumulator.f0 + value.getTemperature(), accumulator.f1 + 1);
            }
    
            @Override
            public Double getResult(Tuple2<Double, Integer> accumulator) { 
                return accumulator.f0/accumulator.f1;
            }
    
            @Override
            public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) { 
                return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]: /images/20221022/a0c3d434cdc8433da37fb3131cb5991e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]: /images/20221022/ba0ddb5343ce472facaecfe4b42dc3e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 2]: /images/20221022/6554a63894974f78bbb0ad04edc48ebf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 3]: /images/20221022/9696c071a8df4a21ab6c6f0d3ab61a32.png