大佬教程收集整理的这篇文章主要介绍了CoProcessFunction实战三部曲之二:状态处理,大佬教程大佬觉得挺不错的,现在分享给大家,也给大家做个参考。
https://github.com/zq2599/blog_demos
内容:所有原创文章分类汇总及配套源码,涉及Java、Docker、Kubernetes、DevOPS等;
理解状态:《深入了解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文件夹下,如下图红框所示:
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<@R_772_10495@ng, Integer>, Tuple2<@R_772_10495@ng, Integer>, Tuple2<@R_772_10495@ng, Integer>> getCoProcessFunctionInstance() {
return new CoProcessFunction<Tuple2<@R_772_10495@ng, Integer>, Tuple2<@R_772_10495@ng, Integer>, Tuple2<@R_772_10495@ng, Integer>>() {
// 某个key在procesSELER_620_11845@ent1中存入的状态
private ValueState<Integer> state1;
// 某个key在procesSELER_620_11845@ent2中存入的状态
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 procesSELER_620_11845@ent1(Tuple2<@R_772_10495@ng, Integer> value, Context ctx, Collector<Tuple2<@R_772_10495@ng, Integer>> out) throws Exception {
logger.info("处理元素1:{}", value);
@R_772_10495@ng key = value.f0;
Integer value2 = state2.value();
// value2为空,就表示procesSELER_620_11845@ent2还没有处理或这个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 procesSELER_620_11845@ent2(Tuple2<@R_772_10495@ng, Integer> value, Context ctx, Collector<Tuple2<@R_772_10495@ng, Integer>> out) throws Exception {
logger.info("处理元素2:{}", value);
@R_772_10495@ng key = value.f0;
Integer value1 = state1.value();
// value1为空,就表示procesSELER_620_11845@ent1还没有处理或这个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(@R_772_10495@ng[] args) throws Exception {
new AddTwosourceValue().execute();
}
}
22:35:12,135 INFO AddTwosourceValue - 处理元素1:(aaa,111)
22:35:12,136 INFO AddTwosourceValue - 2号流还未收到过[aaa],把1号流收到的值[111]保存起来
22:35:34,473 INFO AddTwosourceValue - 处理元素2:(bbb,123)
22:35:34,473 INFO AddTwosourceValue - 1号流还未收到过[bbb],把2号流收到的值[123]保存起来
22:35:38,072 INFO AddTwosourceValue - 处理元素2:(aaa,222)
22:35:38,072 INFO AddTwosourceValue - 1号流收到过[aaa],值是[111],现在把两个值相加后输出
(aaa,333)
以上是大佬教程为你收集整理的CoProcessFunction实战三部曲之二:状态处理全部内容,希望文章能够帮你解决CoProcessFunction实战三部曲之二:状态处理所遇到的程序开发问题。
如果觉得大佬教程网站内容还不错,欢迎将大佬教程推荐给程序员好友。
本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ:384754419,请注明来意。