网站建设一般涉及后台功能,做装饰网站公司,东莞做网站 动点官网,网站如何收录Flink在早期版本有一个split算子用来做数据分流使用的,但是在flink-1.12开始这个API就已经被删除了,在1.12版本以后我们是通过process算子来做数据分流的,这里就介绍一下如何使用prodess进行数据分流.
代码
import org.apache.flink.api.common.typeinfo.TypeInformation;
im…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);// 这里使用的是自定义数据源为了方便测试,具体数据源根据自己的实际情况进行更换DataStreamSourceCustomizeBean customizeSourceStream env.addSource(new CustomizeSource());/*** 需求* 1. 将性别为M且爱好为羽毛球运动爱好者分到一个流* 2. 将性别为W且爱好为篮球运动爱好者或钓鱼爱好者分到一个流* 3. 其他保留到主流**/SingleOutputStreamOperatorCustomizeBean processedStream customizeSourceStream.process(new ProcessFunctionCustomizeBean, CustomizeBean() {Overridepublic void processElement(CustomizeBean value, ProcessFunctionCustomizeBean, CustomizeBean.Context ctx, CollectorCustomizeBean out) throws Exception {String gender value.getGender(); // 性别String hobbit value.getHobbit(); // 爱好if (gender.equals(M) hobbit.equals(羽毛球运动爱好者)) {// 将性别为M且爱好为羽毛球运动爱好者进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTagCustomizeBean(M-羽毛球, TypeInformation.of(CustomizeBean.class)), value);} else if (gender.equals(W) (hobbit.equals(篮球运动爱好者) || hobbit.equals(钓鱼爱好者))) {// 将性别为W且爱好为篮球运动爱好者或钓鱼爱好者进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTagCustomizeBean(W-篮球/钓鱼, TypeInformation.of(CustomizeBean.class)), value);} else {// 将剩下的数据保留在主流中out.collect(value);}}});// 获取M-羽毛球分流数据,这里也要加上类型声明DataStreamCustomizeBean mSideOutput processedStream.getSideOutput(new OutputTagCustomizeBean(M-羽毛球, TypeInformation.of(CustomizeBean.class)));// 打印M-羽毛球结果mSideOutput.print(M-羽毛球);// 获取W-篮球/钓鱼分流数据,这里也要加上类型声明DataStreamCustomizeBean wSideOutput processedStream.getSideOutput(new OutputTagCustomizeBean(W-篮球/钓鱼, TypeInformation.of(CustomizeBean.class)));// 打印结果wSideOutput.print(W-篮球/钓鱼);// 主流数据打印结果processedStream.print(主数据流);env.execute(Side Output);}
}结果数据
主数据流:2 CustomizeBean(nameAAA-641, age44, genderW, hobbit非遗文化爱好者)
主数据流:3 CustomizeBean(nameAAA-17, age62, genderM, hobbit书法爱好者)
主数据流:1 CustomizeBean(nameAAA-429, age25, genderW, hobbit非遗文化爱好者)
主数据流:2 CustomizeBean(nameAAA-218, age33, genderM, hobbit旅游爱好者)
主数据流:3 CustomizeBean(nameAAA-826, age39, genderM, hobbit篮球运动爱好者)
主数据流:1 CustomizeBean(nameAAA-190, age31, genderM, hobbit旅游爱好者)
主数据流:2 CustomizeBean(nameAAA-266, age32, genderW, hobbit网吧战神)
主数据流:3 CustomizeBean(nameAAA-106, age70, genderM, hobbit书法爱好者)
主数据流:1 CustomizeBean(nameAAA-911, age50, genderM, hobbit网吧战神)
M-羽毛球:2 CustomizeBean(nameAAA-925, age65, genderM, hobbit羽毛球运动爱好者)
主数据流:3 CustomizeBean(nameAAA-20, age59, genderM, hobbit书法爱好者)
主数据流:1 CustomizeBean(nameAAA-409, age79, genderW, hobbit天文知识爱好者)
主数据流:2 CustomizeBean(nameAAA-865, age58, genderW, hobbit天文知识爱好者)
主数据流:3 CustomizeBean(nameAAA-898, age33, genderM, hobbit天文知识爱好者)
主数据流:1 CustomizeBean(nameAAA-85, age38, genderW, hobbit非遗文化爱好者)
主数据流:2 CustomizeBean(nameAAA-883, age51, genderM, hobbit美食爱好者)
主数据流:3 CustomizeBean(nameAAA-243, age37, genderM, hobbit钓鱼爱好者)
主数据流:1 CustomizeBean(nameAAA-430, age28, genderW, hobbit旅游爱好者)
主数据流:2 CustomizeBean(nameAAA-127, age65, genderW, hobbit网吧战神)
W-篮球/钓鱼:3 CustomizeBean(nameAAA-986, age52, genderW, hobbit钓鱼爱好者)
主数据流:1 CustomizeBean(nameAAA-840, age50, genderW, hobbit旅游爱好者)
M-羽毛球:2 CustomizeBean(nameAAA-196, age34, genderM, hobbit羽毛球运动爱好者)
主数据流:3 CustomizeBean(nameAAA-142, age46, genderW, hobbit乒乓球运动爱好者)
主数据流:1 CustomizeBean(nameAAA-985, age78, genderW, hobbit美食爱好者)
W-篮球/钓鱼:2 CustomizeBean(nameAAA-490, age50, genderW, hobbit钓鱼爱好者)
主数据流:3 CustomizeBean(nameAAA-295, age77, genderM, hobbit篮球运动爱好者)
主数据流:1 CustomizeBean(nameAAA-754, age50, genderM, hobbit天文知识爱好者)
主数据流:2 CustomizeBean(nameAAA-249, age35, genderW, hobbit羽毛球运动爱好者)
W-篮球/钓鱼:3 CustomizeBean(nameAAA-908, age27, genderW, hobbit钓鱼爱好者)
主数据流:1 CustomizeBean(nameAAA-674, age73, genderM, hobbit非遗文化爱好者)通过结果内容可以看到数据完全按照我们分流的逻辑进行输出的,如果想在主数据流中讲所有数据保留下来,CollectorObject out单独拎出来即可,也就是不加到判断逻辑中,代码如下,这里就只展示部分代码了
SingleOutputStreamOperatorCustomizeBean processedStream customizeSourceStream.process(new ProcessFunctionCustomizeBean, CustomizeBean() {Overridepublic void processElement(CustomizeBean value, ProcessFunctionCustomizeBean, CustomizeBean.Context ctx, CollectorCustomizeBean 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 OutputTagCustomizeBean(M-羽毛球, TypeInformation.of(CustomizeBean.class)), value);} else if ((gender.equals(W) (hobbit.equals(篮球运动爱好者)) || (gender.equals(W) hobbit.equals(钓鱼爱好者)))) {// 将性别为W且爱好为篮球运动爱好者或钓鱼爱好者进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTagCustomizeBean(W-篮球/钓鱼, TypeInformation.of(CustomizeBean.class)), value);}}});所有的内容到这里就结束了.
