欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；

Flink处理函数实战系列链接

1. 深入了解ProcessFunction的状态操作(Flink-1.10)；
2. ProcessFunction；
3. KeyedProcessFunction类；
4. ProcessAllWindowFunction(窗口处理)；
5. CoProcessFunction(双流处理)；

本篇概览

本文是《Flink处理函数实战》系列的第四篇，内容是学习以下两个窗口相关的处理函数：

1. ProcessAllWindowFunction：处理每个窗口内的所有元素；
2. ProcessWindowFunction：处理指定key的每个窗口内的所有元素；

关于ProcessAllWindowFunction

1. ProcessAllWindowFunction和《Flink处理函数实战之二：ProcessFunction类》中的ProcessFunction类相似，都是用来对上游过来的元素做处理，不过ProcessFunction是每个元素执行一次processElement方法，ProcessAllWindowFunction是每个窗口执行一次process方法（方法内可以遍历该窗口内的所有元素）；
2. 用类图对比可以更形象的认识差别，下图左侧是ProcessFunction，右侧是ProcessAllWindowFunction：
   
   

关于ProcessWindowFunction

1. ProcessWindowFunction和KeyedProcessFunction类似，都是处理分区的数据，不过KeyedProcessFunction是每个元素执行一次processElement方法，而ProcessWindowFunction是每个窗口执行一次process方法（方法内可以遍历该key当前窗口内的所有元素）；
2. 用类图对比可以更形象的认识差别，下图左侧是KeyedProcessFunction，右侧是ProcessWindowFunction：
   
   
3. 另外还一个差异：ProcessWindowFunction.process方法的入参就有分区的key值，而KeyedProcessFunction.processElement方法的入参没有这个参数，而是需要Context.getCurrentKey()才能取到分区的key值；

注意事项

窗口处理函数的process方法，以ProcessAllWindowFunction为例，如下图红框所示，其入参可以遍历当前窗口内的所有元素，这意味着当前窗口的所有元素都保存在堆内存中，所以请在设计阶段就严格控制窗口内元素的内存使用量，避免耗尽TaskManager节点的堆内存：



接下来通过实战学习ProcessAllWindowFunction和ProcessWindowFunction；

版本信息

1. 开发环境操作系统：MacBook Pro 13寸， macOS Catalina 10.15.4
2. 开发工具：IntelliJ IDEA 2019.3.2 (Ultimate Edition)
3. JDK：1.8.0_121
4. Maven：3.3.9
5. Flink：1.9.2

源码下载

如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog_demos)：

名称	链接	备注
项目主页	https://github.com/zq2599/blog_demos	该项目在GitHub上的主页
git仓库地址(https)	https://github.com/zq2599/blog_demos.git	该项目源码的仓库地址，https协议
git仓库地址(ssh)	git@github.com:zq2599/blog_demos.git	该项目源码的仓库地址，ssh协议

这个git项目中有多个文件夹，本章的应用在flinkstudy文件夹下，如下图红框所示：

如何实战ProcessAllWindowFunction

接下来通过以下方式验证ProcessAllWindowFunction功能：

1. 每隔1秒发出一个Tuple2<String, Integer>对象，对象的f0字段在aaa和bbb之间变化，f1字段固定为1；
2. 设置5秒的滚动窗口；
3. 自定义ProcessAllWindowFunction扩展类，功能是统计每个窗口内元素的数量，将统计结果发给下游算子；
4. 下游算子将统计结果打印出来；
5. 核对发出的数据和统计信息，看是否一致；

开始编码

1. 继续使用《Flink处理函数实战之二：ProcessFunction类》一文中创建的工程flinkstudy；
2. 新建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));

    }

}

1. 关于ProcessAllWindowFunctionDemo，有几点需要注意：

a. 滚动窗口设置用timeWindowAll方法；

b. ProcessAllWindowFunction的匿名子类的process方法中，context.window().getStart()方法可以取得当前窗口的起始时间，getEnd()方法可以取得当前窗口的结束时间；

1. 编码结束，执行ProcessAllWindowFunctionDemo类验证数据，如下图，检查其中一个窗口的元素详情和ProcessAllWindowFunction执行结果，可见符合预期：
   
   
2. ProcessAllWindowFunction已经了解，接下来尝试ProcessWindowFunction；

如何实战ProcessWindowFunction

接下来通过以下方式验证ProcessWindowFunction功能：

1. 每隔1秒发出一个Tuple2<String, Integer>对象，对象的f0字段在aaa和bbb之间变化，f1字段固定为1；
2. 以f0字段为key进行分区；
3. 分区后的数据进入5秒的滚动窗口；
4. 自定义ProcessWindowFunction扩展类，功能之一是统计每个key在每个窗口内元素的数量，将统计结果发给下游算子；
5. 功能之二是在更新当前key的元素总量，然后在状态后端(backend)保存，这是验证KeyedStream在处理函数中的状态读写能力；
6. 下游算子将统计结果打印出来；
7. 核对发出的数据和统计信息（每个窗口的和总共的分别核对），看是否一致；

开始编码

1. 新建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;

    }

}

1. 上述代码有几处需要关注：

a. 静态类KeyCount.java，是用来保存每个key元素总数的数据结构；

b. timeWindow方法设置了市场为5秒的滚动窗口；

c. 每个Tuple2元素以f0为key进行分区；

d. open方法对名为myState的自定义状态进行注册；

e. process方法中，state.value()取得当前key的状态，tate.update(current)更新当前key的状态；

1. 接下来运行ProcessWindowFunctionDemo类检查数据，如下图，process方法内，对窗口内元素的统计和数据源打印的一致，并且从backend取得的总数在累加后和数据源的统计信息也一致：
   
   
   
   至此，处理函数中窗口处理相关的实战已经完成，如果您也在学习Flink的处理函数，希望本文能给您一些参考；

你不孤单，欣宸原创一路相伴

1. Java系列
2. Spring系列
3. Docker系列
4. kubernetes系列
5. 数据库+中间件系列
6. DevOps系列

欢迎关注公众号：程序员欣宸

微信搜索「程序员欣宸」，我是欣宸，期待与您一同畅游Java世界...

https://github.com/zq2599/blog_demos

最新文章

1. 解决eclipse端口占用问题
2. .Net Core CLI在CentOS7的安装及使用简介
3. espcms /public/class_connector.php intval truncation Vul Arbitrary User Login
4. css实现微信信息背景qq聊天气泡
5. Ubuntu 12.04(32位)安装Oracle 11g(32位)
6. Oracle备份表结构和数据
7. 将文件读取到内存、打印pe结构
8. 【CodeVS 1014】装箱问题
9. Unity3d之协程自实现测试
10. delphi处理消息的几种方式
11. hdu1540线段树
12. asp.net MVC下使用rest
13. Java内存泄露监控工具：JVM监控工具介绍
14. 第三个spring冲刺第10天
15. C++ Boost库简介（转载）
16. Ubuntu 14.10 下安装伪分布式hbase 0.99.0
17. [Canvas]RPG游戏雏形 （地图加载，英雄出现并移动）
18. libevent安装总结
19. JS函数的几种用法
20. JQuery中event的preventDefault和stopPropagation介绍

热门文章

1. python 爬虫 循环分页
2. Windows环境下vscode Live Server插件如何开启https
3. Helium文档1-WebUI自动化-环境准备与入门
4. Curl可以模拟浏览器
5. poj1655 Balancing Act (dp? dfs?)
6. 【bug录】安装项目编译环境bug录
7. Redis学习笔记（四）——数据结构之List
8. 拖拽编写SVG图形化工具(二)
9. MongoDB 监控 --- MongoDB基础用法（八）
10. 【转】Setting up SDL 2 on Visual Studio 2019 Community