乐趣区

关于云计算:CoProcessFunction实战三部曲之二状态处理

欢送拜访我的 GitHub

https://github.com/zq2599/blog_demos

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

本篇概览

  • 本文是《CoProcessFunction 实战三部曲》的第二篇,咱们要实战的是双流连贯场景下,解决一号流中的数据时,还要联合该 key 在二号流中的状况;
  • 最简略的例子:<font color=”blue”>aaa</font> 在一号流中的 value 和二号流的 value 相加,再输入到上游,如下图所示,一号流中的 value 存入 state,在二号流中取出并相加,将后果输入给上游:
  • 本篇的内容就是编码实现上图的性能;

    参考文章

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

    源码下载

    如果您不想写代码,整个系列的源码可在 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 我的项目中有多个文件夹,本章的利用在 <font color=”blue”>flinkstudy</font> 文件夹下,如下图红框所示:

编码

  1. 字符串转 Tuple2 的 Map 函数,以及抽象类 <font color=”blue”>AbstractCoProcessFunctionExecutor</font> 都和上一篇《CoProcessFunction 实战三部曲之一:基本功能》截然不同;
  2. 新增 AbstractCoProcessFunctionExecutor 的子类 <font color=”blue”>AddTwoSourceValue.java</font>,源码如下,稍后会阐明几个关键点:
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();}
}
  1. 关键点之一:对于 <font color=”blue”>aaa</font> 这个 key,无奈确定会先呈现在一号源还是二号源,如果先呈现在一号源,就应该在 processElement1 中将 value 保留在 state1 中,这样等到 aaa 再次出现在二号源时,processElement2 就能够从 state1 中取出一号源的 value,相加后输入到上游;
  2. 关键点之二:如果输入到上游,就示意数据曾经处理完毕,此时要把保留的状态清理掉;
  3. 如果您想理解低阶函数中的状态存取的更多细节,请参考《深刻理解 ProcessFunction 的状态操作(Flink-1.10)》

验证

  1. 别离开启本机的 <font color=”blue”>9998</font> 和 <font color=”blue”>9999</font> 端口,我这里是 MacBook,执行 <font color=”blue”>nc -l 9998</font> 和 <font color=”blue”>nc -l 9999</font>
  2. 启动 Flink 利用,如果您和我一样是 Mac 电脑,间接运行 <font color=”blue”>AddTwoSourceValue.main</font> 办法即可(如果是 windows 电脑,我这没试过,不过做成 jar 在线部署也是能够的);
  3. 在监听 9998 端口的控制台输出 <font color=”blue”>aaa,111</font>,此时 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] 保存起来
  1. 在监听 9999 端口的控制台输出 <font color=”blue”>bbb,123</font>,flink 日志如下所示,示意 bbb 也是首次呈现,把值保留在 state 中:
22:35:34,473 INFO  AddTwoSourceValue - 解决元素 2:(bbb,123)
22:35:34,473 INFO  AddTwoSourceValue - 1 号流还未收到过 [bbb],把 2 号流收到的值[123] 保存起来
  1. 在监听 9999 端口的控制台输出 <font color=”blue”>aaa,222</font>,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)
  • 至此,双流场景下的状态互通实际咱们曾经实现了,接下来的文章,会加上定时器和旁路输入,将双流场景的数据处理思考得更加全面;

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

  1. Java 系列
  2. Spring 系列
  3. Docker 系列
  4. kubernetes 系列
  5. 数据库 + 中间件系列
  6. DevOps 系列

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

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

退出移动版