转：

Flink处理函数实战之五：CoProcessFunction(双流处理)

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

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

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos

Flink处理函数实战系列链接

1. 深入了解ProcessFunction的状态操作(Flink-1.10)；
2. ProcessFunction；
3. KeyedProcessFunction类；
4. ProcessAllWindowFunction(窗口处理)；
5. CoProcessFunction(双流处理)；

本篇概览

• 本文是《Flink处理函数实战》系列的第五篇，学习内容是如何同时处理两个数据源的数据；
• 试想在面对两个输入流时，如果这两个流的数据之间有业务关系，该如何编码实现呢，例如下图中的操作，同时监听9998和9999端口，将收到的输出分别处理后，再由同一个sink处理(打印)：
  
• Flink支持的方式是扩展CoProcessFunction来处理，为了更清楚认识，我们把KeyedProcessFunction和CoProcessFunction的类图摆在一起看，如下所示：
  
• 从上图可见，CoProcessFunction和KeyedProcessFunction的继承关系一样，另外CoProcessFunction自身也很简单，在processElement1和processElement2中分别处理两个上游流入的数据即可，并且也支持定时器设置；

编码实战

接下来咱们开发一个应用来体验CoProcessFunction，功能非常简单，描述如下：

1. 建两个数据源，数据分别来自本地9998和9999端口；
2. 每个端口收到类似aaa,123这样的数据，转成Tuple2实例，f0是aaa，f1是123；
3. 在CoProcessFunction的实现类中，对每个数据源的数据都打日志，然后全部传到下游算子；
4. 下游操作是打印，因此9998和9999端口收到的所有数据都会在控制台打印出来；
5. 整个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算子

1. 做一个map算子，用来将字符串aaa,123转成Tuple2实例，f0是aaa，f1是123；
2. 算子名为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]));
    }
}