大佬教程收集整理的这篇文章主要介绍了CoProcessFunction实战三部曲之一：基本功能，大佬教程大佬觉得挺不错的，现在分享给大家，也给大家做个参考。
https://github.com/zq2599/blog_demos
内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；
本篇咱们要开发的应用，其功能非常简单，描述如下：
如果您不想写代码，整个系列的源码可在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文件夹下，如下图红框所示：
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]));
}
}
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");
}
}
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 procesSELER_963_11845@ent1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {
logger.info("处理1号流的元素：{},", value);
out.collect(value);
}
@Override
public void procesSELER_963_11845@ent2(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();
}
}
12:45:38,774 INFO CollectEveryOne - 处理1号流的元素：(aaa,111),
(aaa,111)
12:45:43,816 INFO CollectEveryOne - 处理2号流的元素：(bbb,222)
(bbb,222)
以上是大佬教程为你收集整理的CoProcessFunction实战三部曲之一：基本功能全部内容，希望文章能够帮你解决CoProcessFunction实战三部曲之一：基本功能所遇到的程序开发问题。
如果觉得大佬教程网站内容还不错，欢迎将大佬教程推荐给程序员好友。
本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ：384754419，请注明来意。