CoProcessFunction实战三部曲之一：基本功能，软件技术java工程师方向

CoProcessFunction实战三部曲之一：基本功能，软件技术java工程师方向

• CoProcessFunction的作用是同时处理两个数据源的数据；

• 试想在面对两个输入流时，如果这两个流的数据之间有业务关系，该如何编码实现呢，例如下图中的 *** 作，同时监听9998和9999端口，将收到的输出分别处理后，再由同一个sink处理(打印)：

• Flink支持的方式是扩展CoProcessFunction来处理，为了更清楚认识，我们把KeyedProcessFunction和CoProcessFunction的类图摆在一起看，如下所示：

• 从上图可见，CoProcessFunction和KeyedProcessFunction的继承关系一样，另外CoProcessFunction自身也很简单，在processElement1和processElement2中分别处理两个上游流入的数据即可，并且也支持定时器设置；

[](

)本篇实战功能简介

本篇咱们要开发的应用，其功能非常简单，描述如下：

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文件夹下，如下图红框所示：

[](

)代码简介

1. 开发一个Map算子，将字符串转成Tuple2；

2. 再开发抽象类AbstractCoProcessFunctionExecutor，功能包括：flink启动、监听端口、调用算子处理数据、双流连接、将双流处理结果打印出来；

3. 从上面的描述可见，AbstractCoProcessFunctionExecutor做了很多事情，唯独没有实现双流连接后的具体业务逻辑，这些没有做的是留给子类来实现的，整个三部曲系列的重点都集中在AbstractCoProcessFunctionExecutor的子类上，把双流连接后的业务逻辑做好，如下图所示，红色为CoProcessFunction的业务代码，其他的都在抽象类中完成：

[](

)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> {

@Override

public Tuple2 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;

public abstract class AbstractCoProcessFunctionExecutor {

protected abstract CoProcessFunction<

Tuple2,

Tuple2,

Tuple2> getCoProcessFunctionInstance();

protected KeyedStream, 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> mainDataStream) {

}

public void execute() throws Exception {

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 并行度1

env.setParallelism(1);

// 监听9998端口的输入

KeyedStream, Tuple> stream1 = buildStreamFromSocket(env, 9998);

// 监听9999端口的输入

KeyedStream, Tuple> stream2 = buildStreamFromSocket(env, 9999);

SingleOutputStreamOperator> mainDataStream = stream1

// 两个流连接

.connect(stream2)

// 执行低阶处理函数，具体处理逻辑在子类中实现