关于云计算:Flink处理函数实战之四窗口处理

8次阅读

共计 9425 个字符，预计需要花费 24 分钟才能阅读完成。

https://github.com/zq2599/blog_demos

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

深刻理解 ProcessFunction 的状态操作(Flink-1.10)；
ProcessFunction；
KeyedProcessFunction 类；
ProcessAllWindowFunction(窗口解决)；
CoProcessFunction(双流解决)；

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

ProcessAllWindowFunction：解决每个窗口内的所有元素；

ProcessWindowFunction：解决指定 key 的每个窗口内的所有元素；

ProcessAllWindowFunction 和《Flink 处理函数实战之二：ProcessFunction 类》中的 ProcessFunction 类类似，都是用来对上游过去的元素做解决，不过 ProcessFunction 是每个元素执行一次 processElement 办法，ProcessAllWindowFunction 是每个窗口执行一次 process 办法（办法内能够遍历该窗口内的所有元素）；
用类图比照能够更形象的意识差异，下图左侧是 ProcessFunction，右侧是 ProcessAllWindowFunction：

ProcessWindowFunction 和 KeyedProcessFunction 相似， 都是解决分区的数据 ，不过 KeyedProcessFunction 是每个元素执行一次 processElement 办法，而 ProcessWindowFunction 是每个窗口执行一次 process 办法（办法内能够遍历该 key 以后窗口内的所有元素）；
用类图比照能够更形象的意识差异，下图左侧是 KeyedProcessFunction，右侧是 ProcessWindowFunction：

另外还一个差别：ProcessWindowFunction.process 办法的入参就有分区的 key 值，而 KeyedProcessFunction.processElement 办法的入参没有这个参数，而是须要 Context.getCurrentKey()能力取到分区的 key 值；

窗口处理函数的 process 办法，以 ProcessAllWindowFunction 为例，如下图红框所示，其入参能够遍历以后窗口内的所有元素，这意味着以后窗口的所有元素都保留在堆内存中，所以 请在设计阶段就严格控制窗口内元素的内存使用量 ，防止耗尽 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 我的项目中有多个文件夹，本章的利用在 flinkstudy 文件夹下，如下图红框所示：

接下来通过以下形式验证 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 性能：

每隔 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

正文完

云计算

发表至：云计算

2020-11-23

0

关于云计算:浪潮云说直播间第九期今晚2000准时开播

关于云计算:阿里云无影发布生态共荣计划携手伙伴推动终端算力上云

关于云计算:群晖DS218部署Harbor1103

关于云计算:将数据迁移到云端时什么对于企业而言最重要

关于elasticsearch:46elasticsearch搜索引擎scrapy写入数据到elasticsearch中

关于云计算:Flink处理函数实战之四窗口处理

欢送拜访我的 GitHub

Flink 处理函数实战系列链接

本篇概览

对于 ProcessAllWindowFunction

对于 ProcessWindowFunction

注意事项

版本信息

源码下载

如何实战 ProcessAllWindowFunction

开始编码

如何实战 ProcessWindowFunction

开始编码

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

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

站内搜索