乐趣区

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

jiezi

欢送拜访我的 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

退出移动版