4.1.18 Flink-流处理框架-Window API之窗口函数window function

àì夳堔傛蜴生んèń 2022-11-05 00:48 158阅读 0赞

**目录**

1.window function-窗口函数

2.其他可选API

3.代码演示

3.1 TimeWindow

3.2 CountWindow

--------------------

# 1.window function-窗口函数 #

window function 定义了要对窗口中收集的数据做的计算操作，主要可以分为两类：

*  增量聚合函数（incremental aggregation functions） 每条数据到来就进行计算，保持一个简单的状态。典型的增量聚合函数有 ReduceFunction, AggregateFunction。（来一个处理一个）
 *  全窗口函数（full window functions） 先把窗口所有数据收集起来，等到计算的时候会遍历所有数据。 ProcessWindowFunction和WindowFunction就是全窗口函数。（一个窗口一起处理）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNTUwMzQ_size_16_color_FFFFFF_t_70][]

# 2.其他可选API #

*   .trigger() —— 触发器      定义 window 什么时候关闭，触发计算并输出结果
 *   .evitor() —— 移除器       定义移除某些数据的逻辑
 *   .allowedLateness() —— 允许处理迟到的数据
 *   .sideOutputLateData() —— 将迟到的数据放入侧输出流
 *   .getSideOutput() —— 获取侧输出流

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

# 3.代码演示 #

## 3.1 TimeWindow时间窗口 ##

package com.atguigu.apitest.window;
    
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.commons.collections.IteratorUtils;
    import org.apache.flink.api.common.functions.AggregateFunction;
    import org.apache.flink.api.common.functions.ReduceFunction;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    import org.apache.flink.util.OutputTag;
    
    import java.util.Arrays;
    import java.util.Collections;
    
    
    public class WindowTest1_TimeWindow {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
    //        // 从文件读取数据
    //        DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            // 开窗测试
    
            // 1. 增量聚合函数
            DataStream<Integer> resultStream = dataStream.keyBy("id")
    //                .countWindow(10, 2);
    //                .window(EventTimeSessionWindows.withGap(Time.minutes(1)));
    //                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
                    .timeWindow(Time.seconds(15))
                    //聚合统计总数
                    .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
                        @Override
                        public Integer createAccumulator() {
                            return 0;
                        }
    
                        @Override
                        public Integer add(SensorReading value, Integer accumulator) {
                            return accumulator + 1;
                        }
    
                        @Override
                        public Integer getResult(Integer accumulator) {
                            return accumulator;
                        }
    
                        @Override
                        public Integer merge(Integer a, Integer b) {
                            return a + b;
                        }
                    });
    
            // 2. 全窗口函数
            SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultStream2 = dataStream.keyBy("id")
                    .timeWindow(Time.seconds(15))
    //                .process(new ProcessWindowFunction<SensorReading, Object, Tuple, TimeWindow>() {
    //                })
                    .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() {
                        @Override
                        public void apply(Tuple tuple, TimeWindow window, Iterable<SensorReading> input, Collector<Tuple3<String, Long, Integer>> out) throws Exception {
                            String id = tuple.getField(0);
                            Long windowEnd = window.getEnd();
                            Integer count = IteratorUtils.toList(input.iterator()).size();
                            out.collect(new Tuple3<>(id, windowEnd, count));
                        }
                    });
    
            // 3. 其它可选API
            OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") {
            };
    
            SingleOutputStreamOperator<SensorReading> sumStream = dataStream.keyBy("id")
                    .timeWindow(Time.seconds(15))
    //                .trigger()
    //                .evictor()
                    .allowedLateness(Time.minutes(1))
                    .sideOutputLateData(outputTag)
                    .sum("temperature");
    
            sumStream.getSideOutput(outputTag).print("late");
    
            resultStream2.print();
    
    
            env.execute();
        }
    }

![20210303160629881.png][]

## 3.2 CountWindow计数窗口 ##

package com.atguigu.apitest.window;
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.functions.AggregateFunction;
    import org.apache.flink.api.common.functions.ReduceFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    
    public class WindowTest2_CountWindow {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
    //        // 从文件读取数据
    //        DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            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> avgTempResultStream = dataStream.keyBy("id")
                    .countWindow(10, 2)
                    .aggregate(new MyAvgTemp());
    
            avgTempResultStream.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_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNTUwMzQ_size_16_color_FFFFFF_t_70]: /images/20221023/dd79b0a3201c4f9fba4399d0b824416b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/48c7f6c7e1aa40a5a51ee7ca0ff50fea.png
[20210303160629881.png]: /images/20221023/3924617035ed42f098767ffcbf6e7e0d.png