Flink多流转换（一）分流

所谓“分流”&＃xff0c;就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个DataStream&＃xff0c;得到完全平等的多个子 DataStream

使用侧输出流在 Flink 1.13 版本中&＃xff0c;已经弃用了.split()方法&＃xff0c;取而代之的是直接用处理函数&＃xff08;process function&＃xff09;的侧输出流&＃xff08;side output&＃xff09;。

处理函数本身可以认为是一个转换算子&＃xff0c;它的输出类型是单一的&＃xff0c;处理之后得到的仍然是一个 DataStream&＃xff1b;而侧输出流则不受限制&＃xff0c;可以任意自定义输出数据&＃xff0c;它们就像从“主流”上分叉出的“支流”。尽管看起来主流和支流有所区别&＃xff0c;不过实际上它们都是某种类型的 DataStream&＃xff0c;所以本质上还是平等的。利用侧输出流就可以很方便地实现分流操作&＃xff0c;而且得到的多条 DataStream 类型可以不同&＃xff0c;这就给我们的应用带来了极大的便利。
关于处理函数中侧输出流的用法&＃xff0c;简单来说&＃xff0c;只需要调用上下文 ctx 的.output()方法&＃xff0c;就可以输出任意类型的数据了。而侧输出流的标记和提取&＃xff0c;都离不开一个“输出标签”&＃xff08;OutputTag&＃xff09;&＃xff0c;它就相当于 split()分流时的“戳”&＃xff0c;指定了侧输出流的id 和类型。

代码如下&＃xff1a;Gitee上的完整代码

public class SplitStreamTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// 定义输出标签&＃xff0c;侧输出流的数据类型为三元组(user, url, timestamp)OutputTag<Tuple3<String, String, Long>> MaryTag &＃61; new OutputTag<Tuple3<String, String, Long>>("Mary") {};OutputTag<Tuple3<String, String, Long>> BobTag &＃61; new OutputTag<Tuple3<String, String, Long>>("Bob") {};//准备数据源SingleOutputStreamOperator<Event> stream &＃61; env.addSource(new ClickSource());// 准备分流操作SingleOutputStreamOperator<Event> processStream &＃61; stream.process(new ProcessFunction<Event, Event>() {&＃64;Overridepublic void processElement(Event value, ProcessFunction<Event, Event>.Context ctx, Collector<Event> out) throws Exception {//如果数据user 是Mary 则单独打印if (value.user.equals("Mary")) {ctx.output(MaryTag, Tuple3.of(value.user, value.url, value.timestamp));//如果数据user 是Bob 则单独打印} else if (value.user.equals("Bob")) {ctx.output(BobTag, Tuple3.of(value.user, value.url, value.timestamp));//如果数据user 是其他数据,则打印} else {out.collect(new Event(value.user, value.url, value.timestamp));}}});processStream.print("Event");processStream.getSideOutput(MaryTag).print("Mary");processStream.getSideOutput(BobTag).print("Bob");env.execute();}
}

这里我们定义了两个侧输出流&＃xff0c;分别拣选 Mary 的浏览事件和 Bob 的浏览事件&＃xff1b;由于类型已经确定&＃xff0c;我们可以只保留(用户 id, url, 时间戳)这样一个三元组。而剩余的事件则直接输出到主流&＃xff0c;类型依然保留 Event&＃xff0c;这样的实现方式显然更简洁&＃xff0c;也更加灵活