CoProcessFunction 实战三部曲之一：基本功能，软件技术 java 工程师方向
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/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 文件夹下，如下图红框所示：
[](
)代码简介
开发一个 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)
// 执行低阶处理函数，具体处理逻辑在子类中实现
还未添加个人签名 2021.03.18 加入
还未添加个人简介
评论