关于云计算:CoProcessFunction实战三部曲之一基本功能

https://github.com/zq2599/blog_demos

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

《CoProcessFunction 实战三部曲》旨在通过三次实战，由浅入深的学习和把握 Flink 低阶处理函数 CoProcessFunction 的用法；
整个系列的开篇先介绍 CoProcessFunction，而后迅速进入实战，理解 CoProcessFunction 的基本功能；
下一篇会联合状态，让双流元素的解决彼此放弃关系；
终篇的实战会退出定时器性能，确保同一个 key 的数据在双流场景下可能及时处理；

开发环境操作系统：MacBook Pro 13 寸，macOS Catalina 10.15.3
开发工具：IDEA ULTIMATE 2018.3
JDK：1.8.0_211
Maven：3.6.0
Flink：1.9.2

基本功能
状态解决
定时器和侧输入

CoProcessFunction 的作用是同时解决两个数据源的数据；
试想在面对两个输出流时，如果这两个流的数据之间有业务关系，该如何编码实现呢，例如下图中的操作，同时监听 9998 和 9999 端口，将收到的输入别离解决后，再由同一个 sink 解决(打印)：
Flink 反对的形式是扩大 CoProcessFunction 来解决，为了更分明意识，咱们把 KeyedProcessFunction 和 CoProcessFunction 的类图摆在一起看，如下所示：
从上图可见，CoProcessFunction 和 KeyedProcessFunction 的继承关系一样，另外 CoProcessFunction 本身也很简略，在 processElement1 和 processElement2 中别离解决两个上游流入的数据即可，并且也反对定时器设置；

本篇实战性能简介

本篇咱们要开发的利用，其性能非常简单，形容如下：
建两个数据源，数据别离来自本地 9998 和 9999 端口；
每个端口收到相似 aaa,123 这样的数据，转成 Tuple2 实例，f0 是 aaa，f1 是 123；
在 CoProcessFunction 的实现类中，对每个数据源的数据都打日志，而后全副传到上游算子；
上游操作是打印，因而 9998 和 9999 端口收到的所有数据都会在控制台打印进去；
整个 demo 的性能如下图所示：
接下来开始编码；

源码下载

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

开发一个 Map 算子，将字符串转成 Tuple2；
再开发抽象类 AbstractCoProcessFunctionExecutor，性能包含：flink 启动、监听端口、调用算子解决数据、双流连贯、将双流处理结果打印进去；
从下面的形容可见，AbstractCoProcessFunctionExecutor 做了很多事件，唯独没有实现双流连贯后的具体业务逻辑，这些没有做的是留给子类来实现的，整个三部曲系列的重点都集中在 AbstractCoProcessFunctionExecutor 的子类上，把双流连贯后的业务逻辑做好，如下图所示，红色为 CoProcessFunction 的业务代码，其余的都在抽象类中实现：

Map 算子
做一个 map 算子，用来将字符串 aaa,123 转成 Tuple2 实例，f0 是 aaa，f1 是 123；
算子名为 WordCountMap.java：

package com.bolingcavalry.coprocessfunction;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.StringUtils;

public class WordCountMap implements MapFunction<String, Tuple2<String, Integer>> {
    @Override
    public Tuple2<String, Integer> map(String s) throws Exception {if(StringUtils.isNullOrWhitespaceOnly(s)) {System.out.println("invalid line");
            return null;
        }

        String[] array = s.split(",");

        if(null==array || array.length<2) {System.out.println("invalid line for array");
            return null;
        }

        return new Tuple2<>(array[0], Integer.valueOf(array[1]));
    }
}

抽象类 AbstractCoProcessFunctionExecutor.java，源码如下，稍后会阐明几个关键点：

package com.bolingcavalry.coprocessfunction;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;

/**
 * @author will
 * @email zq2599@gmail.com
 * @date 2020-11-09 17:33
 * @description 串起整个逻辑的执行类，用于体验 CoProcessFunction
 */
public abstract class AbstractCoProcessFunctionExecutor {

    /**
     * 返回 CoProcessFunction 的实例，这个办法留给子类实现
     * @return
     */
    protected abstract CoProcessFunction<
            Tuple2<String, Integer>,
            Tuple2<String, Integer>,
            Tuple2<String, Integer>> getCoProcessFunctionInstance();

    /**
     * 监听依据指定的端口，* 失去的数据先通过 map 转为 Tuple2 实例，* 给元素退出工夫戳，* 再按 f0 字段分区，* 将分区后的 KeyedStream 返回
     * @param port
     * @return
     */
    protected KeyedStream<Tuple2<String, Integer>, Tuple> buildStreamFromSocket(StreamExecutionEnvironment env, int port) {
        return env
                // 监听端口
                .socketTextStream("localhost", port)
                // 失去的字符串 "aaa,3" 转成 Tuple2 实例，f0="aaa"，f1=3
                .map(new WordCountMap())
                // 将单词作为 key 分区
                .keyBy(0);
    }

    /**
     * 如果子类有侧输入须要解决，请重写此办法，会在主流程执行结束后被调用
     */
    protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) { }

    /**
     * 执行业务的办法
     * @throws Exception
     */
    public void execute() throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 并行度 1
        env.setParallelism(1);

        // 监听 9998 端口的输出
        KeyedStream<Tuple2<String, Integer>, Tuple> stream1 = buildStreamFromSocket(env, 9998);

        // 监听 9999 端口的输出
        KeyedStream<Tuple2<String, Integer>, Tuple> stream2 = buildStreamFromSocket(env, 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream = stream1
                // 两个流连贯
                .connect(stream2)
                // 执行低阶处理函数，具体解决逻辑在子类中实现
                .process(getCoProcessFunctionInstance());

        // 将低阶处理函数输入的元素全副打印进去
        mainDataStream.print();

        // 侧输入相干逻辑，子类有侧输入需要时重写此办法
        doSideOutput(mainDataStream);

        // 执行
        env.execute("ProcessFunction demo : CoProcessFunction");
    }
}

关键点之一：一共有两个数据源，每个源的解决逻辑都封装到 buildStreamFromSocket 办法中；
关键点之二：stream1.connect(stream2) 将两个流连接起来；
关键点之三：process 接管 CoProcessFunction 实例，合并后的流的解决逻辑就在这外面；
关键点之四：getCoProcessFunctionInstance 是形象办法，返回 CoProcessFunction 实例，交给子类实现，所以 CoProcessFunction 中做什么事件齐全由子类决定；
关键点之五：doSideOutput 办法中啥也没做，然而在主流程代码的开端会被调用，如果子类有侧输入 (SideOutput) 的需要，重写此办法即可，此办法的入参是解决过的数据集，能够从这里获得侧输入；

子类，对连贯后的双流进行操作
本篇子类 CollectEveryOne.java 如下所示，逻辑很简略，将每个源的上游数据间接输入到上游算子：

package com.bolingcavalry.coprocessfunction;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CollectEveryOne extends AbstractCoProcessFunctionExecutor {private static final Logger logger = LoggerFactory.getLogger(CollectEveryOne.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>>() {

            @Override
            public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {logger.info("解决 1 号流的元素：{},", value);
                out.collect(value);
            }

            @Override
            public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {logger.info("解决 2 号流的元素：{}", value);
                out.collect(value);
            }
        };
    }

    public static void main(String[] args) throws Exception {new CollectEveryOne().execute();}
}

上述代码中，CoProcessFunction 前面的泛型定义很长：<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>，一共三个 Tuple2，别离代表一号数据源输出、二号数据源输出、上游输入的类型；
编码实现，运行起来试试；

别离开启本机的 9998 和 9999 端口，我这里是 MacBook，执行 nc -l 9998 和 nc -l 9999
启动 Flink 利用，如果您和我一样是 Mac 电脑，间接运行 CollectEveryOne.main 办法即可（如果是 windows 电脑，我这没试过，不过做成 jar 在线部署也是能够的）；
在监听 9998 和 9999 端口的控制台别离输出 aaa,111 和 bbb,222
以下是 flink 控制台输入的内容，可见 processElement1 和 processElement2 办法的日志代码曾经执行，并且 print 办法作为最上游，将两个数据源的数据都打印进去了，合乎预期：

12:45:38,774 INFO CollectEveryOne - 解决 1 号流的元素：(aaa,111),
(aaa,111)
12:45:43,816 INFO CollectEveryOne - 解决 2 号流的元素：(bbb,222)
(bbb,222)

至此，咱们的第一个双流解决低阶函数就实现了，对 CoProcessFunction 也有了最根本的意识，当然 CoProcessFunction 的作用远不迭此，下一篇咱们借助状态让 processElement1 和 processElement2 别离对方解决过的状态，让每个元素的解决都和另一个流关联，不再孤立；

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

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

关于云计算:CoProcessFunction实战三部曲之一基本功能

欢送拜访我的 GitHub

对于《CoProcessFunction 实战三部曲》系列

版本信息

系列文章链接

对于 CoProcessFunction

本篇实战性能简介

源码下载

代码简介

Map 算子

抽象类

子类，对连贯后的双流进行操作

验证

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

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