Flink之SideOutput(数据分流)

Flink在早期版本有一个split算子用来做数据分流使用的,但是在flink-1.12开始这个API就已经被删除了,在1.12版本以后我们是通过process算子来做数据分流的,这里就介绍一下如何使用prodess进行数据分流.

代码

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;/*** @Author: J* @Version: 1.0* @CreateTime: 2023/8/7* @Description: 测流输出**/
public class FlinkSideOutput {public static void main(String[] args) throws Exception {// 构建流环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度env.setParallelism(3);// 这里使用的是自定义数据源为了方便测试,具体数据源根据自己的实际情况进行更换DataStreamSource<CustomizeBean> customizeSourceStream = env.addSource(new CustomizeSource());/*** 需求* 1. 将性别为M且爱好为'羽毛球运动爱好者'分到一个流* 2. 将性别为W且爱好为'篮球运动爱好者'或'钓鱼爱好者'分到一个流* 3. 其他保留到主流**/SingleOutputStreamOperator<CustomizeBean> processedStream = customizeSourceStream.process(new ProcessFunction<CustomizeBean, CustomizeBean>() {@Overridepublic void processElement(CustomizeBean value, ProcessFunction<CustomizeBean, CustomizeBean>.Context ctx, Collector<CustomizeBean> out) throws Exception {String gender = value.getGender(); // 性别String hobbit = value.getHobbit(); // 爱好if (gender.equals("M") && hobbit.equals("羽毛球运动爱好者")) {// 将性别为M且爱好为'羽毛球运动爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("M-羽毛球", TypeInformation.of(CustomizeBean.class)), value);} else if (gender.equals("W") && (hobbit.equals("篮球运动爱好者") || hobbit.equals("钓鱼爱好者"))) {// 将性别为W且爱好为'篮球运动爱好者'或'钓鱼爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("W-篮球/钓鱼", TypeInformation.of(CustomizeBean.class)), value);} else {// 将剩下的数据保留在主流中out.collect(value);}}});// 获取'M-羽毛球'分流数据,这里也要加上类型声明DataStream<CustomizeBean> mSideOutput = processedStream.getSideOutput(new OutputTag<CustomizeBean>("M-羽毛球", TypeInformation.of(CustomizeBean.class)));// 打印'M-羽毛球'结果mSideOutput.print("M-羽毛球");// 获取'W-篮球/钓鱼'分流数据,这里也要加上类型声明DataStream<CustomizeBean> wSideOutput = processedStream.getSideOutput(new OutputTag<CustomizeBean>("W-篮球/钓鱼", TypeInformation.of(CustomizeBean.class)));// 打印结果wSideOutput.print("W-篮球/钓鱼");// 主流数据打印结果processedStream.print("主数据流");env.execute("Side Output");}
}

结果数据

主数据流:2> CustomizeBean(name=AAA-641, age=44, gender=W, hobbit=非遗文化爱好者)
主数据流:3> CustomizeBean(name=AAA-17, age=62, gender=M, hobbit=书法爱好者)
主数据流:1> CustomizeBean(name=AAA-429, age=25, gender=W, hobbit=非遗文化爱好者)
主数据流:2> CustomizeBean(name=AAA-218, age=33, gender=M, hobbit=旅游爱好者)
主数据流:3> CustomizeBean(name=AAA-826, age=39, gender=M, hobbit=篮球运动爱好者)
主数据流:1> CustomizeBean(name=AAA-190, age=31, gender=M, hobbit=旅游爱好者)
主数据流:2> CustomizeBean(name=AAA-266, age=32, gender=W, hobbit=网吧战神)
主数据流:3> CustomizeBean(name=AAA-106, age=70, gender=M, hobbit=书法爱好者)
主数据流:1> CustomizeBean(name=AAA-911, age=50, gender=M, hobbit=网吧战神)
M-羽毛球:2> CustomizeBean(name=AAA-925, age=65, gender=M, hobbit=羽毛球运动爱好者)
主数据流:3> CustomizeBean(name=AAA-20, age=59, gender=M, hobbit=书法爱好者)
主数据流:1> CustomizeBean(name=AAA-409, age=79, gender=W, hobbit=天文知识爱好者)
主数据流:2> CustomizeBean(name=AAA-865, age=58, gender=W, hobbit=天文知识爱好者)
主数据流:3> CustomizeBean(name=AAA-898, age=33, gender=M, hobbit=天文知识爱好者)
主数据流:1> CustomizeBean(name=AAA-85, age=38, gender=W, hobbit=非遗文化爱好者)
主数据流:2> CustomizeBean(name=AAA-883, age=51, gender=M, hobbit=美食爱好者)
主数据流:3> CustomizeBean(name=AAA-243, age=37, gender=M, hobbit=钓鱼爱好者)
主数据流:1> CustomizeBean(name=AAA-430, age=28, gender=W, hobbit=旅游爱好者)
主数据流:2> CustomizeBean(name=AAA-127, age=65, gender=W, hobbit=网吧战神)
W-篮球/钓鱼:3> CustomizeBean(name=AAA-986, age=52, gender=W, hobbit=钓鱼爱好者)
主数据流:1> CustomizeBean(name=AAA-840, age=50, gender=W, hobbit=旅游爱好者)
M-羽毛球:2> CustomizeBean(name=AAA-196, age=34, gender=M, hobbit=羽毛球运动爱好者)
主数据流:3> CustomizeBean(name=AAA-142, age=46, gender=W, hobbit=乒乓球运动爱好者)
主数据流:1> CustomizeBean(name=AAA-985, age=78, gender=W, hobbit=美食爱好者)
W-篮球/钓鱼:2> CustomizeBean(name=AAA-490, age=50, gender=W, hobbit=钓鱼爱好者)
主数据流:3> CustomizeBean(name=AAA-295, age=77, gender=M, hobbit=篮球运动爱好者)
主数据流:1> CustomizeBean(name=AAA-754, age=50, gender=M, hobbit=天文知识爱好者)
主数据流:2> CustomizeBean(name=AAA-249, age=35, gender=W, hobbit=羽毛球运动爱好者)
W-篮球/钓鱼:3> CustomizeBean(name=AAA-908, age=27, gender=W, hobbit=钓鱼爱好者)
主数据流:1> CustomizeBean(name=AAA-674, age=73, gender=M, hobbit=非遗文化爱好者)

通过结果内容可以看到数据完全按照我们分流的逻辑进行输出的,如果想在主数据流中讲所有数据保留下来,Collector<Object> out单独拎出来即可,也就是不加到判断逻辑中,代码如下,这里就只展示部分代码了

SingleOutputStreamOperator<CustomizeBean> processedStream = customizeSourceStream.process(new ProcessFunction<CustomizeBean, CustomizeBean>() {@Overridepublic void processElement(CustomizeBean value, ProcessFunction<CustomizeBean, CustomizeBean>.Context ctx, Collector<CustomizeBean> out) throws Exception {String gender = value.getGender(); // 性别String hobbit = value.getHobbit(); // 爱好// 将所有数据保留在主流中out.collect(value);// 开始进行分流处理if (gender.equals("M") && hobbit.equals("羽毛球运动爱好者")) {// 将性别为M且爱好为'羽毛球运动爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("M-羽毛球", TypeInformation.of(CustomizeBean.class)), value);} else if ((gender.equals("W") && (hobbit.equals("篮球运动爱好者")) || (gender.equals("W") && hobbit.equals("钓鱼爱好者")))) {// 将性别为W且爱好为'篮球运动爱好者'或'钓鱼爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("W-篮球/钓鱼", TypeInformation.of(CustomizeBean.class)), value);}}});