概述
窗口的长度(大小): 决定了要计算最近多长时间的数据
窗口的间隔: 决定了每隔多久计算一次
举例：每隔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 FunctionNon-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>>() {
@Override
public 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>() {
@Override
public 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>>() {
@Override
public 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>() {
@Override
public 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>>() {
@Override
public Tuple2<String,Integer> map(String value) throws Exception {
return Tuple2.of(value,1);
}
}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public 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>() {
@Override
public 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>>() {
// 初始化一个 累加器
@Override
public Tuple3<String, Integer, Long> createAccumulator() {
return Tuple3.of(null,0,0L);
}
// 累加器和输入的值进行累加
// Tuple3<String, String, Long> value 第一个是传入的值
// Tuple3<String, Integer, Long> accumulator 第二个是累加器的值
@Override
public 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);
}
// 获取结果——在不同节点的结果进行汇总后实现
@Override
public Tuple2<String, Double> getResult(Tuple3<String, Integer, Long> accumulator) {
return Tuple2.of(accumulator.f0, (double) accumulator.f2 / accumulator.f1);
}
// 由于flink是分布式，所以在别的节点也会进行累加 ，该方法是不同节点的结果进行汇总
// 即累加器之间的累加
@Override
public 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>>() {
@Override
public 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>() {
@Override
public 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>>() {
@Override
public 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();
}
}
