Flink处理函数实战之四:窗口处理

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
简介: 学习Flink低阶处理函数中的ProcessAllWindowFunction和ProcessWindowFunction

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos

Flink处理函数实战系列链接

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

本篇概览

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

  1. ProcessAllWindowFunction:处理每个窗口内的所有元素;
    1. ProcessWindowFunction:处理指定key的每个窗口内的所有元素;

关于ProcessAllWindowFunction

  1. ProcessAllWindowFunction和《Flink处理函数实战之二:ProcessFunction类》中的ProcessFunction类相似,都是用来对上游过来的元素做处理,不过ProcessFunction是每个元素执行一次processElement方法,ProcessAllWindowFunction是每个窗口执行一次process方法(方法内可以遍历该窗口内的所有元素);
  2. 用类图对比可以更形象的认识差别,下图左侧是ProcessFunction,右侧是ProcessAllWindowFunction:
    在这里插入图片描述

    关于ProcessWindowFunction

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

    注意事项

    窗口处理函数的process方法,以ProcessAllWindowFunction为例,如下图红框所示,其入参可以遍历当前窗口内的所有元素,这意味着当前窗口的所有元素都保存在堆内存中,所以请在设计阶段就严格控制窗口内元素的内存使用量,避免耗尽TaskManager节点的堆内存:
    在这里插入图片描述
    接下来通过实战学习ProcessAllWindowFunction和ProcessWindowFunction;

    版本信息

  6. 开发环境操作系统:MacBook Pro 13寸, macOS Catalina 10.15.4
  7. 开发工具:IntelliJ IDEA 2019.3.2 (Ultimate Edition)
  8. JDK:1.8.0_121
  9. Maven:3.3.9
  10. 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对象,对象的f0字段在aaa和bbb之间变化,f1字段固定为1;
  2. 设置5秒的滚动窗口;
  3. 自定义ProcessAllWindowFunction扩展类,功能是统计每个窗口内元素的数量,将统计结果发给下游算子;
  4. 下游算子将统计结果打印出来;
  5. 核对发出的数据和统计信息,看是否一致;

    开始编码

  6. 继续使用《Flink处理函数实战之二:ProcessFunction类》一文中创建的工程flinkstudy;
  7. 新建ProcessAllWindowFunctionDemo类,如下:
    ```java
    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));
}

}

3. 关于ProcessAllWindowFunctionDemo,有几点需要注意:
a. 滚动窗口设置用timeWindowAll方法;
b. ProcessAllWindowFunction的匿名子类的process方法中,context.window().getStart()方法可以取得当前窗口的起始时间,getEnd()方法可以取得当前窗口的结束时间;
4. 编码结束,执行ProcessAllWindowFunctionDemo类验证数据,如下图,检查其中一个窗口的元素详情和ProcessAllWindowFunction执行结果,可见符合预期:
![在这里插入图片描述](https://img2023.cnblogs.com/blog/485422/202308/485422-20230805210302887-901698821.png)
5. 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:
```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的状态;
  2. 接下来运行ProcessWindowFunctionDemo类检查数据,如下图,process方法内,对窗口内元素的统计和数据源打印的一致,并且从backend取得的总数在累加后和数据源的统计信息也一致:
    在这里插入图片描述
    至此,处理函数中窗口处理相关的实战已经完成,如果您也在学习Flink的处理函数,希望本文能给您一些参考;

欢迎关注阿里云开发者社区:程序员欣宸

学习路上,你不孤单,欣宸原创一路相伴...

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
4月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
在数字化转型中,企业亟需从海量数据中快速提取价值并转化为业务增长动力。5月15日19:00-21:00,阿里云三位技术专家将讲解Kafka与Flink的强强联合方案,帮助企业零门槛构建分布式实时分析平台。此组合广泛应用于实时风控、用户行为追踪等场景,具备高吞吐、弹性扩缩容及亚秒级响应优势。直播适合初学者、开发者和数据工程师,参与还有机会领取定制好礼!扫描海报二维码或点击链接预约直播:[https://developer.aliyun.com/live/255088](https://developer.aliyun.com/live/255088)
340 35
直播预告|Kafka+Flink双引擎实战:手把手带你搭建分布式实时分析平台!
|
4月前
|
消息中间件 运维 Kafka
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
直播预告|Kafka+Flink 双引擎实战:手把手带你搭建分布式实时分析平台!
168 11
|
10月前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
698 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
11月前
|
SQL 消息中间件 分布式计算
大数据-120 - Flink Window 窗口机制-滑动时间窗口、会话窗口-基于时间驱动&基于事件驱动
大数据-120 - Flink Window 窗口机制-滑动时间窗口、会话窗口-基于时间驱动&基于事件驱动
248 0
|
11月前
|
SQL 分布式计算 大数据
大数据-119 - Flink Window总览 窗口机制-滚动时间窗口-基于时间驱动&基于事件驱动
大数据-119 - Flink Window总览 窗口机制-滚动时间窗口-基于时间驱动&基于事件驱动
380 0
|
6月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
探索Flink动态CEP:杭州银行的实战案例
210 5
|
8月前
|
缓存 监控 数据处理
Flink 四大基石之窗口(Window)使用详解
在流处理场景中,窗口(Window)用于将无限数据流切分成有限大小的“块”,以便进行计算。Flink 提供了多种窗口类型,如时间窗口(滚动、滑动、会话)和计数窗口,通过窗口大小、滑动步长和偏移量等属性控制数据切分。窗口函数包括增量聚合函数、全窗口函数和ProcessWindowFunction,支持灵活的数据处理。应用案例展示了如何使用窗口进行实时流量统计和电商销售分析。
1759 28
|
9月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
本文由杭州银行大数据工程师唐占峰、欧阳武林撰写,介绍Flink动态CEP的定义、应用场景、技术实现及使用方式。Flink动态CEP是基于Flink的复杂事件处理库,支持在不重启服务的情况下动态更新规则,适应快速变化的业务需求。文章详细阐述了其在反洗钱、反欺诈和实时营销等金融领域的应用,并展示了某金融机构的实际应用案例。通过动态CEP,用户可以实时调整规则,提高系统的灵活性和响应速度,降低维护成本。文中还提供了具体的代码示例和技术细节,帮助读者理解和使用Flink动态CEP。
1106 2
探索Flink动态CEP:杭州银行的实战案例
|
9月前
|
数据处理 数据安全/隐私保护 流计算
Flink 三种时间窗口、窗口处理函数使用及案例
Flink 是处理无界数据流的强大工具,提供了丰富的窗口机制。本文介绍了三种时间窗口(滚动窗口、滑动窗口和会话窗口)及其使用方法,包括时间窗口的概念、窗口处理函数的使用和实际案例。通过这些机制,可以灵活地对数据流进行分析和计算,满足不同的业务需求。
1006 27
|
1月前
|
存储 分布式计算 数据处理
「48小时极速反馈」阿里云实时计算Flink广招天下英雄
阿里云实时计算Flink团队,全球领先的流计算引擎缔造者,支撑双11万亿级数据处理,推动Apache Flink技术发展。现招募Flink执行引擎、存储引擎、数据通道、平台管控及产品经理人才,地点覆盖北京、杭州、上海。技术深度参与开源核心,打造企业级实时计算解决方案,助力全球企业实现毫秒洞察。
364 0
「48小时极速反馈」阿里云实时计算Flink广招天下英雄

热门文章

最新文章