欢送拜访我的GitHub
https://github.com/zq2599/blog_demos
内容：所有原创文章分类汇总及配套源码，波及Java、Docker、Kubernetes、DevOPS等；
Flink处理函数实战系列链接
- 深刻理解ProcessFunction的状态操作(Flink-1.10)；
- ProcessFunction；
- KeyedProcessFunction类；
- ProcessAllWindowFunction(窗口解决)；
- CoProcessFunction(双流解决)；
本篇概览
本文是《Flink处理函数实战》系列的第四篇，内容是学习以下两个窗口相干的处理函数：
- ProcessAllWindowFunction：解决每个窗口内的所有元素；
- ProcessWindowFunction：解决指定key的每个窗口内的所有元素；
对于ProcessAllWindowFunction
- ProcessAllWindowFunction和《Flink处理函数实战之二：ProcessFunction类》中的ProcessFunction类类似，都是用来对上游过去的元素做解决，不过ProcessFunction是每个元素执行一次processElement办法，ProcessAllWindowFunction是每个窗口执行一次process办法（办法内能够遍历该窗口内的所有元素）；
- 用类图比照能够更形象的意识差异，下图左侧是ProcessFunction，右侧是ProcessAllWindowFunction：
对于ProcessWindowFunction
- ProcessWindowFunction和KeyedProcessFunction相似，<font color="red">都是解决分区的数据</font>，不过KeyedProcessFunction是每个元素执行一次<font color="blue">processElement</font>办法，而ProcessWindowFunction是每个窗口执行一次<font color="blue">process</font>办法（办法内能够遍历该key以后窗口内的所有元素）；
- 用类图比照能够更形象的意识差异，下图左侧是KeyedProcessFunction，右侧是ProcessWindowFunction：
- 另外还一个差别：ProcessWindowFunction.process办法的入参就有分区的key值，而KeyedProcessFunction.processElement办法的入参没有这个参数，而是须要Context.getCurrentKey()能力取到分区的key值；
注意事项
窗口处理函数的process办法，以ProcessAllWindowFunction为例，如下图红框所示，其入参能够遍历以后窗口内的所有元素，这意味着以后窗口的所有元素都保留在堆内存中，所以<font color="red">请在设计阶段就严格控制窗口内元素的内存使用量</font>，防止耗尽TaskManager节点的堆内存：
接下来通过实战学习ProcessAllWindowFunction和ProcessWindowFunction；
版本信息
- 开发环境操作系统：MacBook Pro 13寸， macOS Catalina 10.15.4
- 开发工具：IntelliJ IDEA 2019.3.2 (Ultimate Edition)
- JDK：1.8.0_121
- Maven：3.3.9
- Flink：1.9.2
源码下载
如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blo...：
| 名称 | 链接 | 备注 |
|---|---|---|
| 我的项目主页 | https://github.com/zq2599/blo... | 该我的项目在GitHub上的主页 |
| git仓库地址(https) | https://github.com/zq2599/blo... | 该我的项目源码的仓库地址，https协定 |
| git仓库地址(ssh) | git@github.com:zq2599/blog_demos.git | 该我的项目源码的仓库地址，ssh协定 |
这个git我的项目中有多个文件夹，本章的利用在<font color="blue">flinkstudy</font>文件夹下，如下图红框所示：
如何实战ProcessAllWindowFunction
接下来通过以下形式验证ProcessAllWindowFunction性能：
- 每隔1秒收回一个Tuple2<String, Integer>对象，对象的f0字段在aaa和bbb之间变动，f1字段固定为1；
- 设置5秒的滚动窗口；
- 自定义ProcessAllWindowFunction扩大类，性能是统计每个窗口内元素的数量，将统计后果发给上游算子；
- 上游算子将统计后果打印进去；
- 核查收回的数据和统计信息，看是否统一；
开始编码
- 持续应用《Flink处理函数实战之二：ProcessFunction类》一文中创立的工程flinkstudy；
- 新建ProcessAllWindowFunctionDemo类，如下：
package com.bolingcavalry.processwindowfunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;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 java.text.SimpleDateFormat;import java.util.Date;public class ProcessAllWindowFunctionDemo { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 应用事件工夫 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // 并行度为1 env.setParallelism(1); // 设置数据源，一共三个元素 DataStream<Tuple2<String,Integer>> dataStream = env.addSource(new SourceFunction<Tuple2<String, Integer>>() { @Override public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception { for(int i=1; i<Integer.MAX_VALUE; i++) { // 只有aaa和bbb两种name String name = 0==i%2 ? "aaa" : "bbb"; // 应用以后工夫作为工夫戳 long timeStamp = System.currentTimeMillis(); // 将数据和工夫戳打印进去，用来验证数据 System.out.println(String.format("source，%s, %s\n", name, time(timeStamp))); // 发射一个元素，并且带上了工夫戳 ctx.collectWithTimestamp(new Tuple2<String, Integer>(name, 1), timeStamp); // 每发射一次就延时1秒 Thread.sleep(1000); } } @Override public void cancel() { } }); // 将数据用5秒的滚动窗口做划分，再用ProcessAllWindowFunction SingleOutputStreamOperator<String> mainDataStream = dataStream // 5秒一次的滚动窗口 .timeWindowAll(Time.seconds(5)) // 统计以后窗口内的元素数量，而后把数量、窗口起止工夫整顿成字符串发送给上游算子 .process(new ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>() { @Override public void process(Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector) throws Exception { int count = 0; // iterable能够拜访以后窗口内的所有数据， // 这里简略解决，只统计了元素数量 for (Tuple2<String, Integer> tuple2 : iterable) { count++; } // 将以后窗口的起止工夫和元素数量整顿成字符串 String value = String.format("window, %s - %s, %d\n", // 以后窗口的起始工夫 time(context.window().getStart()), // 以后窗口的完结工夫 time(context.window().getEnd()), // 以后key在以后窗口内元素总数 count); // 发射到上游算子 collector.collect(value); } }); // 打印后果，通过剖析打印信息，查看ProcessWindowFunction中能够解决所有key的整个窗口的数据 mainDataStream.print(); env.execute("processfunction demo : processallwindowfunction"); } public static String time(long timeStamp) { return new SimpleDateFormat("hh:mm:ss").format(new Date(timeStamp)); }}- 对于ProcessAllWindowFunctionDemo，有几点须要留神：
a. 滚动窗口设置用timeWindowAll办法；
b. ProcessAllWindowFunction的匿名子类的process办法中，context.window().getStart()办法能够获得以后窗口的起始工夫，getEnd()办法能够获得以后窗口的完结工夫；
- 编码完结，执行ProcessAllWindowFunctionDemo类验证数据，如下图，查看其中一个窗口的元素详情和ProcessAllWindowFunction执行后果，可见合乎预期：
- ProcessAllWindowFunction曾经理解，接下来尝试ProcessWindowFunction；
如何实战ProcessWindowFunction
接下来通过以下形式验证ProcessWindowFunction性能：
- 每隔1秒收回一个Tuple2<String, Integer>对象，对象的f0字段在aaa和bbb之间变动，f1字段固定为1；
- 以f0字段为key进行分区；
- 分区后的数据进入5秒的滚动窗口；
- 自定义ProcessWindowFunction扩大类，性能之一是统计每个key在每个窗口内元素的数量，将统计后果发给上游算子；
- 性能之二是在更新以后key的元素总量，而后在状态后端(backend)保留，这是验证KeyedStream在处理函数中的状态读写能力；
- 上游算子将统计后果打印进去；
- 核查收回的数据和统计信息（每个窗口的和总共的别离核查），看是否统一；
开始编码
- 新建ProcessWindowFunctionDemo.java：
package com.bolingcavalry.processwindowfunction;import org.apache.flink.api.common.state.ValueState;import org.apache.flink.api.common.state.ValueStateDescriptor;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;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 java.text.SimpleDateFormat;import java.util.Date;public class ProcessWindowFunctionDemo { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 应用事件工夫 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // 并行度为1 env.setParallelism(1); // 设置数据源，一共三个元素 DataStream<Tuple2<String,Integer>> dataStream = env.addSource(new SourceFunction<Tuple2<String, Integer>>() { @Override public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception { int aaaNum = 0; int bbbNum = 0; for(int i=1; i<Integer.MAX_VALUE; i++) { // 只有aaa和bbb两种name String name = 0==i%2 ? "aaa" : "bbb"; //更新aaa和bbb元素的总数 if(0==i%2) { aaaNum++; } else { bbbNum++; } // 应用以后工夫作为工夫戳 long timeStamp = System.currentTimeMillis(); // 将数据和工夫戳打印进去，用来验证数据 System.out.println(String.format("source，%s, %s, aaa total : %d, bbb total : %d\n", name, time(timeStamp), aaaNum, bbbNum)); // 发射一个元素，并且戴上了工夫戳 ctx.collectWithTimestamp(new Tuple2<String, Integer>(name, 1), timeStamp); // 每发射一次就延时1秒 Thread.sleep(1000); } } @Override public void cancel() { } }); // 将数据用5秒的滚动窗口做划分，再用ProcessWindowFunction SingleOutputStreamOperator<String> mainDataStream = dataStream // 以Tuple2的f0字段作为key，本例中实际上key只有aaa和bbb两种 .keyBy(value -> value.f0) // 5秒一次的滚动窗口 .timeWindow(Time.seconds(5)) // 统计每个key以后窗口内的元素数量，而后把key、数量、窗口起止工夫整顿成字符串发送给上游算子 .process(new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() { // 自定义状态 private ValueState<KeyCount> state; @Override public void open(Configuration parameters) throws Exception { // 初始化状态，name是myState state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", KeyCount.class)); } @Override public void process(String s, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector) throws Exception { // 从backend获得以后单词的myState状态 KeyCount current = state.value(); // 如果myState还从未没有赋值过，就在此初始化 if (current == null) { current = new KeyCount(); current.key = s; current.count = 0; } int count = 0; // iterable能够拜访该key以后窗口内的所有数据， // 这里简略解决，只统计了元素数量 for (Tuple2<String, Integer> tuple2 : iterable) { count++; } // 更新以后key的元素总数 current.count += count; // 更新状态到backend state.update(current); // 将以后key及其窗口的元素数量，还有窗口的起止工夫整顿成字符串 String value = String.format("window, %s, %s - %s, %d, total : %d\n", // 以后key s, // 以后窗口的起始工夫 time(context.window().getStart()), // 以后窗口的完结工夫 time(context.window().getEnd()), // 以后key在以后窗口内元素总数 count, // 以后key呈现的总数 current.count); // 发射到上游算子 collector.collect(value); } }); // 打印后果，通过剖析打印信息，查看ProcessWindowFunction中能够解决所有key的整个窗口的数据 mainDataStream.print(); env.execute("processfunction demo : processwindowfunction"); } public static String time(long timeStamp) { return new SimpleDateFormat("hh:mm:ss").format(new Date(timeStamp)); } static class KeyCount { /** * 分区key */ public String key; /** * 元素总数 */ public long count; }}- 上述代码有几处须要关注：
a. 动态类KeyCount.java，是用来保留每个key元素总数的数据结构；
b. timeWindow办法设置了市场为5秒的滚动窗口；
c. 每个Tuple2元素以f0为key进行分区；
d. open办法对名为myState的自定义状态进行注册；
e. process办法中，state.value()获得以后key的状态，tate.update(current)更新以后key的状态；
- 接下来运行ProcessWindowFunctionDemo类检查数据，如下图，process办法内，对窗口内元素的统计和数据源打印的统一，并且从backend获得的总数在累加后和数据源的统计信息也统一：
至此，处理函数中窗口解决相干的实战曾经实现，如果您也在学习Flink的处理函数，心愿本文能给您一些参考；
你不孤独，欣宸原创一路相伴
- Java系列
- Spring系列
- Docker系列
- kubernetes系列
- 数据库+中间件系列
- DevOps系列
欢送关注公众号：程序员欣宸
微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界...
https://github.com/zq2599/blog_demos