flink学习（7）——window

 概述

窗口的长度(大小): 决定了要计算最近多长时间的数据

窗口的间隔: 决定了每隔多久计算一次

举例：每隔10min,计算最近24h的热搜词，24小时是长度，每隔10分钟是间隔。

窗口的分类

1、根据window前是否调用keyBy分为键控窗口和非键控窗口

2、根据window中参数的配置分为基于时间的，基于条数的，会话窗口

SlidingProcessingTimeWindows —— 滑动窗口，按照处理时间

TumblingProcessingTimeWindows —— 滚动窗口，按照处理时间

ProcessingTimeSessionWindows —— 会话窗口

 Keyed Window --键控窗口

// Keyed Window
stream.keyBy(...)              <-  按照一个Key进行分组.window(...)            <-  将数据流中的元素分配到相应的窗口中[.trigger(...)]            <-  指定触发器Trigger（可选）[.evictor(...)]            <-  指定清除器Evictor(可选).reduce/aggregate/process/apply()      <-  窗口处理函数Window Function

Non-Keyed Window

// Non-Keyed Window
stream.windowAll(...)         <-  不分组，将数据流中的所有元素分配到相应的窗口中[.trigger(...)]            <-  指定触发器Trigger（可选）[.evictor(...)]            <-  指定清除器Evictor(可选).reduce/aggregate/process()      <-  窗口处理函数Window Function

方括号([…]) 中的命令是可选的。

首先：我们要决定是否对一个DataStream按照Key进行分组，这一步必须在窗口计算之前进行。

经过keyBy的数据流将形成多组数据，下游算子的多个实例可以并行计算。

windowAll不对数据流进行分组，所有数据将发送到下游算子单个实例上。

决定是否分组之后，窗口的后续操作基本相同。

经过windowAll的算子是不分组的窗口（Non-Keyed Window），它们的原理和操作与Keyed Window类似，唯一的区别在于所有数据将发送给下游的单个实例，或者说下游算子的并行度为1。

Flink窗口的骨架结构中有两个必须的两个操作：

• 使用窗口分配器（WindowAssigner）将数据流中的元素分配到对应的窗口。

• 当满足窗口触发条件后，对窗口内的数据使用窗口处理函数（Window Function）进行处理，常用的Window Function有reduce、aggregate、process。

其他的trigger、evictor则是窗口的触发和销毁过程中的附加选项，主要面向需要更多自定义的高级编程者，如果不设置则会使用默认的配置。

 

 

基于时间的窗口 

滚动窗口- TumblingWindow概念

package com.bigdata.day04;public class _01_windows {/*** 1、实时统计每个红绿灯通过的汽车数量* 2、实时统计每个红绿灯每个1分钟，统计最近1分钟通过的汽车数量 ——滚动* 3、实时统计每个红绿灯每个1分钟，统计最近2分钟通过的汽车数量 ——滑动*/public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStreamSource<String> socketStream = env.socketTextStream("localhost", 8889);//3. transformation-数据处理转换socketStream.map(new MapFunction<String, Tuple2<Integer,Integer>>() {@Overridepublic Tuple2<Integer,Integer> map(String line) throws Exception {String[] words = line.split(" ");return new Tuple2<>(Integer.parseInt(words[0]),Integer.parseInt(words[1]));}}).keyBy(new KeySelector<Tuple2<Integer, Integer>, Integer>() {@Overridepublic Integer getKey(Tuple2<Integer, Integer> value) throws Exception {return value.f0;}})// 基于这个部分实现 滚动窗口 每一分钟 统计前一分钟的数据.window(TumblingProcessingTimeWindows.of(Time.minutes(1))).sum(1).print();env.execute();}
}

滑动窗口– SlidingWindow概念 

package com.bigdata.day04;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.Window;/*** @基本功能:* @program:flinkProject* @author: jinnian* @create:2024-11-25 10:13:46**/
public class _01_windows {/*** 1、实时统计每个红绿灯通过的汽车数量* 2、实时统计每个红绿灯每个1分钟，统计最近1分钟通过的汽车数量 ——滚动* 3、实时统计每个红绿灯每个1分钟，统计最近2分钟通过的汽车数量 ——滑动*/public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStreamSource<String> socketStream = env.socketTextStream("localhost", 8889);//3. transformation-数据处理转换socketStream.map(new MapFunction<String, Tuple2<Integer,Integer>>() {@Overridepublic Tuple2<Integer,Integer> map(String line) throws Exception {String[] words = line.split(" ");return new Tuple2<>(Integer.parseInt(words[0]),Integer.parseInt(words[1]));}}).keyBy(new KeySelector<Tuple2<Integer, Integer>, Integer>() {@Overridepublic Integer getKey(Tuple2<Integer, Integer> value) throws Exception {return value.f0;}})// 基于这一部分实现，每30秒统计前一分钟的数据，大的在前，小的在后.window(SlidingProcessingTimeWindows.of(Time.minutes(1),Time.seconds(30))).sum(1).print();//5. execute-执行env.execute();}
}

如何显示窗口时间——apply

——apply将reduce替代

kafka生产数据

package com.bigdata.day04;public class _02_kafka生产数据 {public static void main(String[] args) throws InterruptedException {// Properties 它是map的一种Properties properties = new Properties();// 设置连接kafka集群的ip和端口properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"bigdata01:9092");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");KafkaProducer<String, String> producer = new KafkaProducer<>(properties);String[] arr = {"联通换猫","遥遥领先","恒大歌舞团","恒大足球队","郑州烂尾楼"};Random random = new Random();for (int i = 0; i < 5000; i++) {int index = random.nextInt(arr.length);ProducerRecord<String, String> record = new ProducerRecord<>("edu", arr[index]);producer.send(record);Thread.sleep(30);}}
}

flink消费数据

package com.bigdata.day04;public class _02_flink消费数据 {public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);Properties properties = new Properties();properties.setProperty("bootstrap.servers", "bigdata01:9092");properties.setProperty("group.id", "gw2");FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>("edu",new SimpleStringSchema(),properties);DataStreamSource<String> source = env.addSource(consumer);source.map(new MapFunction<String, Tuple2<String,Integer>>() {@Overridepublic Tuple2<String,Integer> map(String value) throws Exception {return Tuple2.of(value,1);}}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}}).window(SlidingProcessingTimeWindows.of(Time.minutes(1),Time.seconds(30)))//.window(TumblingProcessingTimeWindows.of(Time.seconds(30)))/****/.apply(new WindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {@Overridepublic void apply(String key, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) throws Exception {StringBuilder sb = new StringBuilder();long start = window.getStart();long end = window.getEnd();String startStr = DateFormatUtils.format(start, "yyyy-MM-dd HH:mm:ss");String endStr = DateFormatUtils.format(end, "yyyy-MM-dd HH:mm:ss");int sum = 0;for (Tuple2<String, Integer> tuple2 : input) {sum +=tuple2.f1;}sb.append("开始时间："+startStr+",").append("结束时间："+endStr+",").append("key: "+key+ ",").append("数量："+sum);out.collect(sb.toString());}}).print();env.execute();}
}

基于条数的窗口——countWindow

package com.bigdata.day04;public class _04_agg函数 {public static final Tuple3[] ENGLISH = new Tuple3[] {Tuple3.of("class1", "张三", 100L),Tuple3.of("class1", "李四", 40L),Tuple3.of("class1", "王五", 60L),Tuple3.of("class2", "赵六", 20L),Tuple3.of("class2", "小七", 30L),Tuple3.of("class2", "小八", 50L),};public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStreamSource<Tuple3<String,String,Long>> dataStreamSource = env.fromElements(ENGLISH);// 此时我要获取每个班级的平均成绩// 输入数据的类型(IN)、累加器的类型（ACC）和输出数据的类型（OUT）// IN——Tuple3<String, String, Long>// ACC——Tuple3<String, Integer,Long> 第一个是班级（key）第二个是数量，第三个是总的成绩// OUT —— Tuple2<String,Double> 第一个是班级 第二个是平均成绩dataStreamSource.countWindowAll(3).aggregate(new AggregateFunction<Tuple3<String, String, Long>, Tuple3<String, Integer,Long>, Tuple2<String,Double>>() {// 初始化一个 累加器@Overridepublic Tuple3<String, Integer, Long> createAccumulator() {return Tuple3.of(null,0,0L);}// 累加器和输入的值进行累加// Tuple3<String, String, Long> value 第一个是传入的值// Tuple3<String, Integer, Long> accumulator 第二个是累加器的值@Overridepublic Tuple3<String, Integer, Long> add(Tuple3<String, String, Long> value, Tuple3<String, Integer, Long> accumulator) {return Tuple3.of(value.f0,accumulator.f1+1,accumulator.f2+value.f2);}// 获取结果——在不同节点的结果进行汇总后实现@Overridepublic Tuple2<String, Double> getResult(Tuple3<String, Integer, Long> accumulator) {return Tuple2.of(accumulator.f0, (double) accumulator.f2 / accumulator.f1);}// 由于flink是分布式，所以在别的节点也会进行累加 ，该方法是不同节点的结果进行汇总// 即累加器之间的累加@Overridepublic Tuple3<String, Integer, Long> merge(Tuple3<String, Integer, Long> a, Tuple3<String, Integer, Long> b) {return Tuple3.of(a.f0,a.f1+b.f1,a.f2+b.f2);}}).print();//4. sink-数据输出//5. execute-执行env.execute();}
}

会话窗口

package com.bigdata.day04;public class _03_会话窗口 {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);env.setParallelism(1);DataStreamSource<String> source = env.socketTextStream("localhost", 8889);source.map(new MapFunction<String, Tuple2<String,Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] s = value.split(" ");return Tuple2.of(s[0],Integer.valueOf(s[1]));}}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}// 1、主要就是 ProcessingTimeSessionWindows 参数的使用// 2、使用 EventTimeSessionWindows的时候，若没有水印就不会有结果}).window(ProcessingTimeSessionWindows.withGap(Time.seconds(5))).reduce(new ReduceFunction<Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {return Tuple2.of(value1.f0,value1.f1+value2.f1);}}).print();env.execute();}
}