当前位置：首页 > news >正文

公司网站免费申请/seo优化软件免费

news 2025/7/23 19:12:37

公司网站免费申请,seo优化软件免费,马云谈2025的房价3500,江苏建设学院文章目录所谓“分流”，就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个DataStream，得到完全平等的多个子 DataStream 使用侧输出流在 Flink 1.13 版本中，已经弃用了.split()方法，取而代之的是直接用处理函数&a…

文章目录

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

在这里插入图片描述

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

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

代码如下：Gitee上的完整代码

public class SplitStreamTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// 定义输出标签，侧输出流的数据类型为三元组(user, url, timestamp)OutputTag<Tuple3<String, String, Long>> MaryTag = new OutputTag<Tuple3<String, String, Long>>("Mary") {};OutputTag<Tuple3<String, String, Long>> BobTag = new OutputTag<Tuple3<String, String, Long>>("Bob") {};//准备数据源SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());// 准备分流操作SingleOutputStreamOperator<Event> processStream = stream.process(new ProcessFunction<Event, Event>() {@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();}
}