古之立大事者，不惟有超世之才，亦必有坚忍不拔之志。这篇文章主要讲述Flink处理函数实战之四：窗口处理相关的知识，希望能为你提供帮助。
欢迎访问我的GitHub
本篇概览

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

1. ProcessAllWindowFunction：处理每个窗口内的所有元素；
2. 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< String, Integer> 对象，对象的f0字段在aaa和bbb之间变化，f1字段固定为1；
2. 设置5秒的滚动窗口；
3. 自定义ProcessAllWindowFunction扩展类，功能是统计每个窗口内元素的数量，将统计结果发给下游算子；
4. 下游算子将统计结果打印出来；
5. 核对发出的数据和统计信息，看是否一致； 开始编码
   
6. 继续使用《Flink处理函数实战之二：ProcessFunction类》一文中创建的工程flinkstudy；
7. 新建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 = https://www.songbingjia.com/android/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));

【Flink处理函数实战之四（窗口处理）】

3. 关于ProcessAllWindowFunctionDemo，有几点需要注意： a. 滚动窗口设置用timeWindowAll方法； b. ProcessAllWindowFunction的匿名子类的process方法中，context.window().getStart()方法可以取得当前窗口的起始时间，getEnd()方法可以取得当前窗口的结束时间； 4. 编码结束，执行ProcessAllWindowFunctionDemo类验证数据，如下图，检查其中一个窗口的元素详情和ProcessAllWindowFunction执行结果，可见符合预期： ![在这里插入图片描述](https://s4.51cto.com/images/blog/202204/05193908_624c2a5c4defb30507.png?x-oss-process=image/watermark,size_14,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_20,type_ZmFuZ3poZW5naGVpdGk=) 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 = https://www.songbingjia.com/android/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;

2. 上述代码有几处需要关注：
   a. 静态类KeyCount.java，是用来保存每个key元素总数的数据结构；
   b. timeWindow方法设置了市场为5秒的滚动窗口；
   c. 每个Tuple2元素以f0为key进行分区；
   d. open方法对名为myState的自定义状态进行注册；
   e. process方法中，state.value()取得当前key的状态，tate.update(current)更新当前key的状态；
3. 接下来运行ProcessWindowFunctionDemo类检查数据，如下图，process方法内，对窗口内元素的统计和数据源打印的一致，并且从backend取得的总数在累加后和数据源的统计信息也一致：
   
   文章图片
   
   • 至此，处理函数中窗口处理相关的实战已经完成，如果您也在学习Flink的处理函数，希望本文能给您一些参考；

欢迎关注51CTO博客：程序员欣宸

推荐阅读

• 

  什么是职场？
• 

  百日练，一百天读一百本书，打卡第9天，《你只是看上去很努力》35分钟，理解50%，2000/分钟
• 

  空调主机跟遥控器显示温度不同步是怎么回事 新科柜机空调不制冷调节温度显示不同步
• 

  等闲识得东风面中的东风是什么意思 等闲识得东风面释义
• 

  redis 集合的底层实现 redis集合存bean
• 

  Java中关于Collections集合工具类的详细介绍
• 

  福州贵安水世界好玩吗 2023福州贵安水世界门票多少钱一张
• 

  蜗居背景音乐歌词蜗居背景音乐歌词翻译
• 

  甘肃所有大学名单 甘肃省大学排名
• 

  雷殿生名字打分99分 雷王姓取名
• 

  python|python 的生产者和消费者模式
• 

  quiet怎么读 quiet怎么造句
• 

  龙眼肉罐头做法 新鲜的龙眼肉怎么做罐头
• 

  海藻面膜敷多长时间好
• 

  mysql my.cnf配置模板参考
• 

  苹果扣费怎么退 iPhone扣费怎么退款
• 

  你认为孤独的根源是什么？
• 

  减肥期间,有哪些做起来简单又好吃的减肥食谱？
• 

  创维55寸4k智能电视多少钱一台 哪个电视牌子好
• 

  二甲双胍|糖尿病天天吃二甲双胍，请花5分钟弄清这6个注意事项，很重要

• 如何利用MHA+ProxySQL实现读写分离和负载均衡
• Java多线程与线程池技术
• MapReduce编程模型和计算框架
• 三高Mysql - Mysql索引和查询优化（偏实战部分）
• 浅析$nextTick和$forceUpdate
• 听说你在写Python爬虫，你对浏览器的开发者工具了解多少（多图预警）
• Spring认证指南（了解如何以最少的配置构建应用程序）
• MySQL优化的5个维度
• Prometheus监控Harbor实战（全网首发）