赞
踩
所谓“分流”,就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于DataStream,得到完全平等的多个子 DataStream,一般来说,我们会定义一些筛选条件,将符合条件的数据拣选出来放到对应的流里。
- public class SplitStreamByFilter {
- public static void main(String[] args) throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());
- // 筛选 Mary 的浏览行为放入 MaryStream 流中
- DataStream<Event> MaryStream = stream.filter(new FilterFunction<Event>() {
- @Override
- public boolean filter(Event value) throws Exception {
- return value.user.equals("Mary");
- }
- });
- // 筛选 Bob 的购买行为放入 BobStream 流中
- DataStream<Event> BobStream = stream.filter(new FilterFunction<Event>() {
- @Override
- public boolean filter(Event value) throws Exception {
- return value.user.equals("Bob");
- }
- });
- // 筛选其他人的浏览行为放入 elseStream 流中
- DataStream<Event> elseStream = stream.filter(new FilterFunction<Event>() {
- @Override
- public boolean filter(Event value) throws Exception {
- return !value.user.equals("Mary") && !value.user.equals("Bob");
- }
- });
- MaryStream.print("Mary pv");
- BobStream.print("Bob pv");
- elseStream.print("else pv");
- env.execute();
- }
- }
这种实现非常简单,但代码显得有些冗余——我们的处理逻辑对拆分出的三条流其实是一样的,却重复写了三次。而且这段代码背后的含义,是将原始数据流 stream 复制三份,然后对每一份分别做筛选;这明显是不够高效的。
- public class SplitStreamByOutputTag {
- // 定义输出标签,侧输出流的数据类型为三元组(user, url, timestamp)
- private static OutputTag<Tuple3<String, String, Long>> MaryTag = new OutputTag<Tuple3<String, String, Long>>("Mary-pv") {};
- private static OutputTag<Tuple3<String, String, Long>> BobTag = new OutputTag<Tuple3<String, String, Long>>("Bob-pv") {};
-
- public static void main(String[] args) throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());
- SingleOutputStreamOperator<Event> processedStream = stream.process(new ProcessFunction<Event, Event>() {
- @Override
- public void processElement(Event value, Context ctx, Collector<Event> out) throws Exception {
- if (value.user.equals("Mary")) {
- ctx.output(MaryTag, new Tuple3<>(value.user, value.url, value.timestamp));
- } else if (value.user.equals("Bob")) {
- ctx.output(BobTag, new Tuple3<>(value.user, value.url, value.timestamp));
- } else {
- out.collect(value);
- }
- }
- });
- processedStream.getSideOutput(MaryTag).print("Mary pv");
- processedStream.getSideOutput(BobTag).print("Bob pv");
- processedStream.print("else");
- env.execute();
- }
- }
这里我们定义了两个侧输出流,分别拣选 Mary 的浏览事件和 Bob 的浏览事件;由于类型已经确定,我们可以只保留(用户 id, url, 时间戳)这样一个三元组。而剩余的事件则直接输出到主流,类型依然保留 Event,就相当于之前的 elseStream。这样的实现方式显然更简洁,也更加灵活。
最简单的合流操作,就是直接将多条流合在一起,叫作流的“联合”(union),联合操作要求必须流中的数据类型必须相同,合并之后的新流会包括所有流中的元素,数据类型不变。这种合流方式非常简单粗暴,就像公路上多个车道汇在一起一样。
stream1.union(stream2, stream3, ...)
注意:对于合流之后的水位线,也是要以最小的那个为准,这样才可以保证所有流都不会再传来之前的数据
连接得到的并不是 DataStream,而是一个“连接流”(ConnectedStreams)。连接流可以看成是两条流形式上的“统一”,被放在了一个同一个流中;事实上内部仍保持各自的数据形式不变,彼此之间是相互独立的。要想得到新的 DataStream,还需要进一步定义一个“同处理”(co-process)转换操作,用来说明对于不同来源、不同类型的数据,怎样分别进行处理转换、得到统一的输出类型。所以整体上来,两条流的连接就像是“一国两制”,两条流可以保持各自的数据类型、处理方式也可以不同,不过最终还是会统一到同一个 DataStream 中。
- public class CoMapExample {
- public static void main(String[] args) throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(1);
- DataStream<Integer> stream1 = env.fromElements(1, 2, 3);
- DataStream<Long> stream2 = env.fromElements(1L, 2L, 3L);
- ConnectedStreams<Integer, Long> connectedStreams = stream1.connect(stream2);
- SingleOutputStreamOperator<String> result = connectedStreams.map(new CoMapFunction<Integer, Long, String>() {
- @Override
- public String map1(Integer value) {
- return "Integer: " + value;
- }
-
- @Override
- public String map2(Long value) {
- return "Long: " + value;
- }
- });
- result.print();
- env.execute();
- }
- }
注意:ConnectedStreams 也可以直接调用.keyBy()进行按键分区的操作,得到的还是一个 ConnectedStreams,可用理解为一个join的操作。
connectedStreams.keyBy(keySelector1, keySelector2);
CoProcessFunction()方法
对于连接流 ConnectedStreams 的处理操作,需要分别定义对两条流的处理转换,因此接口中就会有两个相同的方法需要实现,用数字“1”“2”区分,在两条流中的数据到来时分别调用。我们把这种接口叫作“协同处理函数”(co-process function)。与 CoMapFunction 类似,如果是调用.flatMap()就需要传入一个 CoFlatMapFunction,需要实现 flatMap1()、flatMap2()两个方法;而调用.process()时,传入的则是一个 CoProcessFunction。
广播连接流(BroadcastConnectedStream)
关于两条流的连接,还有一种比较特殊的用法:DataStream 调用.connect()方法时,传入的参数也可以不是一个 DataStream,而是一个“广播流”(BroadcastStream),这时合并两条流得到的就变成了一个“广播连接流”(BroadcastConnectedStream),要用普通数据流connect广播流。
- DataStream<Integer> orangeStream = ...
- DataStream<Integer> greenStream = ...
-
- orangeStream.join(greenStream)
- .where(<KeySelector>)
- .equalTo(<KeySelector>)
- .window(TumblingEventTimeWindows.of(Time.milliseconds(2)))
- .apply(new JoinFunction<Integer, Integer, String> (){
- @Override
- public String join(Integer first, Integer second) {
- return first + "," + second;
- }
- });
缺点:可能存在数据丢失的问题
- DataStream<Integer> orangeStream = ...
- DataStream<Integer> greenStream = ...
-
- orangeStream.join(greenStream)
- .where(<KeySelector>)
- .equalTo(<KeySelector>)
- .window(SlidingEventTimeWindows.of(Time.milliseconds(2) /* size */, Time.milliseconds(1) /* slide */))
- .apply(new JoinFunction<Integer, Integer, String> (){
- @Override
- public String join(Integer first, Integer second) {
- return first + "," + second;
- }
- });
缺点:可能存在数据重复的问题
- DataStream<Integer> orangeStream = ...
- DataStream<Integer> greenStream = ...
-
- orangeStream.join(greenStream)
- .where(<KeySelector>)
- .equalTo(<KeySelector>)
- .window(EventTimeSessionWindows.withGap(Time.milliseconds(1)))
- .apply(new JoinFunction<Integer, Integer, String> (){
- @Override
- public String join(Integer first, Integer second) {
- return first + "," + second;
- }
- });
缺点:如果数据一直传入,session不断开,导致长时间的没输出,失去了实时的功能。
右流相对左流偏移的时间区间进行关联,即:
right.timestamp ∈ [left.timestamp + lowerBound; left.timestamp + upperBound]
- DataStream<Integer> orangeStream = ...
- DataStream<Integer> greenStream = ...
-
- orangeStream
- .keyBy(<KeySelector>)
- .intervalJoin(greenStream.keyBy(<KeySelector>))
- .between(Time.milliseconds(-2), Time.milliseconds(1))
- .process(new ProcessJoinFunction<Integer, Integer, String(){
- @Override
- public void processElement(Integer left, Integer right, Context ctx, Collector<String> out) {
- out.collect(first + "," + second);
- }
- });
注意:
- stream1.coGroup(stream2)
- .where(<KeySelector>)
- .equalTo(<KeySelector>)
- .window(TumblingEventTimeWindows.of(Time.hours(1)))
- .apply(<CoGroupFunction>)
- public interface CoGroupFunction<IN1, IN2, O> extends Function, Serializable {
- void coGroup(Iterable<IN1> first, Iterable<IN2> second, Collector<O> out)
- throws Exception;
- }
可以理解为传入为俩个自定义的收集器,输出为一个自定义的收集器。
onnect操作比join操作更通用。Connect确保两个流( key控的或未锁定的)在同一个位置(在coxxx函数中的同一个并行实例)相遇。
一个流可以是操纵应用于另一个流的行为的控制流。例如,您可以输入新的机器学习模型或其他业务规则。或者,可以使用 key控并在同一位置汇合的两个流的属性进行 join。Flink提供了一些预定义的 join运算符。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。