星空科学探索频道: CoProcessFunction实战三部曲之二：状态处理

2020年12月4日星期五

CoProcessFunction实战三部曲之二：状态处理

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

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

本篇概览

本文是《CoProcessFunction实战三部曲》的第二篇，咱们要实战的是双流连接场景下，处理一号流中的数据时，还要结合该key在二号流中的情况；
最简单的例子：aaa在一号流中的value和二号流的value相加，再输出到下游，如下图所示，一号流中的value存入state，在二号流中取出并相加，将结果输出给下游：
本篇的内容就是编码实现上图的功能；

参考文章

理解状态：《深入了解ProcessFunction的状态操作(Flink-1.10)》

源码下载

如果您不想写代码，整个系列的源码可在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文件夹下，如下图红框所示：
在这里插入图片描述

编码

字符串转Tuple2的Map函数，以及抽象类AbstractCoProcessFunctionExecutor都和上一篇《CoProcessFunction实战三部曲之一：基本功能》一模一样；
新增AbstractCoProcessFunctionExecutor的子类AddTwoSourceValue.java，源码如下，稍后会说明几个关键点：

package com.bolingcavalry.coprocessfunction;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.functions.co.CoProcessFunction;import org.apache.flink.util.Collector;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * @author will * @email zq2599@gmail.com * @date 2020-11-11 09:48 * @description 功能介绍 */public class AddTwoSourceValue extends AbstractCoProcessFunctionExecutor { private static final Logger logger = LoggerFactory.getLogger(AddTwoSourceValue.class); @Override protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() {  return new CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() {   // 某个key在processElement1中存入的状态   private ValueState<Integer> state1;   // 某个key在processElement2中存入的状态   private ValueState<Integer> state2;   @Override   public void open(Configuration parameters) throws Exception {    // 初始化状态    state1 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState1", Integer.class));    state2 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState2", Integer.class));   }   @Override   public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {    logger.info("处理元素1：{}", value);    String key = value.f0;    Integer value2 = state2.value();    // value2为空，就表示processElement2还没有处理或这个key，    // 这时候就把value1保存起来    if(null==value2) {     logger.info("2号流还未收到过[{}]，把1号流收到的值[{}]保存起来", key, value.f1);     state1.update(value.f1);    } else {     logger.info("2号流收到过[{}]，值是[{}]，现在把两个值相加后输出", key, value2);     // 输出一个新的元素到下游节点     out.collect(new Tuple2<>(key, value.f1 + value2));     // 把2号流的状态清理掉     state2.clear();    }   }   @Override   public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {    logger.info("处理元素2：{}", value);    String key = value.f0;    Integer value1 = state1.value();    // value1为空，就表示processElement1还没有处理或这个key，    // 这时候就把value2保存起来    if(null==value1) {     logger.info("1号流还未收到过[{}]，把2号流收到的值[{}]保存起来", key, value.f1);     state2.update(value.f1);    } else {     logger.info("1号流收到过[{}]，值是[{}]，现在把两个值相加后输出", key, value1);     // 输出一个新的元素到下游节点     out.collect(new Tuple2<>(key, value.f1 + value1));     // 把1号流的状态清理掉     state1.clear();    }   }  }; } public static void main(String[] args) throws Exception {  new AddTwoSourceValue().execute(); }}

关键点之一：对于aaa这个key，无法确定会先出现在一号源还是二号源，如果先出现在一号源，就应该在processElement1中将value保存在state1中，这样等到aaa再次出现在二号源时，processElement2就可以从state1中取出一号源的value，相加后输出到下游；
关键点之二：如果输出到下游，就表示数据已经处理完毕，此时要把保存的状态清理掉；
如果您想了解低阶函数中的状态存取的更多细节，请参考《深入了解ProcessFunction的状态操作(Flink-1.10)》

验证

分别开启本机的9998和9999端口，我这里是MacBook，执行nc -l 9998和nc -l 9999
启动Flink应用，如果您和我一样是Mac电脑，直接运行AddTwoSourceValue.main方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
在监听9998端口的控制台输入aaa,111，此时flink控制台输出如下，可见processElement1方法中，读取state2为空，表示aaa在二号流还未出现过，此时的aaa是首次出现，应该放入state中保存：

22:35:12,135 INFO AddTwoSourceValue - 处理元素1：(aaa,111)22:35:12,136 INFO AddTwoSourceValue - 2号流还未收到过[aaa]，把1号流收到的值[111]保存起来

在监听9999端口的控制台输入bbb,123，flink日志如下所示，表示bbb也是首次出现，把值保存在state中：

22:35:34,473 INFO AddTwoSourceValue - 处理元素2：(bbb,123)22:35:34,473 INFO AddTwoSourceValue - 1号流还未收到过[bbb]，把2号流收到的值[123]保存起来

在监听9999端口的控制台输入aaa,222，flink日志如下，很明显，之前保存在state中的值被取出来了，因此processElement2方法中，aaa在两个数据源的值111和222会被相加后输出到下游，下游是print，直接打印出来了：

22:35:38,072 INFO AddTwoSourceValue - 处理元素2：(aaa,222)22:35:38,072 INFO AddTwoSourceValue - 1号流收到过[aaa]，值是[111]，现在把两个值相加后输出(aaa,333)

至此，双流场景下的状态互通实践咱们已经完成了，接下来的文章，会加上定时器和旁路输出，将双流场景的数据处理考虑得更加全面；

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

Java系列
Spring系列
Docker系列
kubernetes系列
数据库+中间件系列
DevOps系列

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

微信搜索「程序员欣宸」，我是欣宸，期待与您一同畅游Java世界...
https://github.com/zq2599/blog_demos

原文转载：http://www.shaoqun.com/a/494808.html

camel：https://www.ikjzd.com/w/331.html

patpat：https://www.ikjzd.com/w/1079.html

houzz：https://www.ikjzd.com/w/236

欢迎访问我的GitHubhttps://github.com/zq2599/blog_demos内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；本篇概览本文是《CoProcessFunction实战三部曲》的第二篇，咱们要实战的是双流连接场景下，处理一号流中的数据时，还要结合该key在二号流中的情况；最简单的例子：aaa在一号流中的value
wangwei：wangwei
review：review
自由行去越南什么时间最佳：自由行去越南什么时间最佳
跨境电商shopify 独立站如何引流量下篇（免费流量版）：跨境电商shopify 独立站如何引流量下篇（免费流量版）
3种超实用的精选亚马逊邀评邮件模板：3种超实用的精选亚马逊邀评邮件模板

星空科学探索频道