欢送拜访我的GitHub
https://github.com/zq2599/blog_demos
内容：所有原创文章分类汇总及配套源码，波及Java、Docker、Kubernetes、DevOPS等；
对于《CoProcessFunction实战三部曲》系列
- 《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
- CoProcessFunction的作用是同时解决两个数据源的数据；
- 试想在面对两个输出流时，如果这两个流的数据之间有业务关系，该如何编码实现呢，例如下图中的操作，同时监听<font color="blue">9998</font>和<font color="blue">9999</font>端口，将收到的输入别离解决后，再由同一个sink解决(打印)：
- Flink反对的形式是扩大CoProcessFunction来解决，为了更分明意识，咱们把<font color="blue">KeyedProcessFunction</font>和<font color="blue">CoProcessFunction</font>的类图摆在一起看，如下所示：
从上图可见，CoProcessFunction和KeyedProcessFunction的继承关系一样，另外CoProcessFunction本身也很简略，在<font color="blue">processElement1</font>和<font color="blue">processElement2</font>中别离解决两个上游流入的数据即可，并且也反对定时器设置；
本篇实战性能简介
本篇咱们要开发的利用，其性能非常简单，形容如下：
- 建两个数据源，数据别离来自本地<font color="red">9998</font>和<font color="red">9999</font>端口；
- 每个端口收到相似<font color="blue">aaa,123</font>这样的数据，转成Tuple2实例，f0是<font color="blue">aaa</font>，f1是<font color="blue">123</font>；
- 在CoProcessFunction的实现类中，对每个数据源的数据都打日志，而后全副传到上游算子；
- 上游操作是打印，因而<font color="red">9998</font>和<font color="red">9999</font>端口收到的所有数据都会在控制台打印进去；
- 整个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我的项目中有多个文件夹，本章的利用在<font color="blue">flinkstudy</font>文件夹下，如下图红框所示：
代码简介
- 开发一个Map算子，将字符串转成Tuple2；
- 再开发抽象类<font color="blue">AbstractCoProcessFunctionExecutor</font>，性能包含：flink启动、监听端口、调用算子解决数据、双流连贯、将双流处理结果打印进去；
从下面的形容可见，<font color="blue">AbstractCoProcessFunctionExecutor</font>做了很多事件，唯独没有实现双流连贯后的具体业务逻辑，这些没有做的是留给子类来实现的，整个三部曲系列的重点都集中在AbstractCoProcessFunctionExecutor的子类上，把双流连贯后的业务逻辑做好，如下图所示，红色为CoProcessFunction的业务代码，其余的都在抽象类中实现：
Map算子
- 做一个map算子，用来将字符串<font color="blue">aaa,123</font>转成Tuple2实例，f0是<font color="red">aaa</font>，f1是<font color="red">123</font>；
- 算子名为<font color="blue">WordCountMap.java</font>：
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"); }}- 关键点之一：一共有两个数据源，每个源的解决逻辑都封装到<font color="blue">buildStreamFromSocket</font>办法中；
- 关键点之二：<font color="blue">stream1.connect(stream2)</font>将两个流连接起来；
- 关键点之三：<font color="blue">process</font>接管CoProcessFunction实例，合并后的流的解决逻辑就在这外面；
- 关键点之四：<font color="blue">getCoProcessFunctionInstance</font>是形象办法，返回<font color="blue">CoProcessFunction</font>实例，交给子类实现，所以CoProcessFunction中做什么事件齐全由子类决定；
关键点之五：doSideOutput办法中啥也没做，然而在主流程代码的开端会被调用，如果子类有侧输入(SideOutput)的需要，重写此办法即可，此办法的入参是解决过的数据集，能够从这里获得侧输入；
子类，对连贯后的双流进行操作
- 本篇子类<font color="blue">CollectEveryOne.java</font>如下所示，逻辑很简略，将每个源的上游数据间接输入到上游算子：
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，别离代表一号数据源输出、二号数据源输出、上游输入的类型；
- 编码实现，运行起来试试；
验证
- 别离开启本机的<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>
- 启动Flink利用，如果您和我一样是Mac电脑，间接运行<font color="blue">CollectEveryOne.main</font>办法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是能够的）；
- 在监听9998和9999端口的控制台别离输出<font color="blue">aaa,111</font>和<font color="blue">bbb,222</font>
- 以下是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的作用远不迭此，下一篇咱们借助状态让<font color="blue">processElement1</font>和<font color="blue">processElement2</font>别离对方解决过的状态，让每个元素的解决都和另一个流关联，不再孤立；
你不孤独，欣宸原创一路相伴
- Java系列
- Spring系列
- Docker系列
- kubernetes系列
- 数据库+中间件系列
- DevOps系列
欢送关注公众号：程序员欣宸
微信搜寻「程序员欣宸」，我是欣宸，期待与您一起畅游Java世界...
https://github.com/zq2599/blog_demos