Flink多流转换（二）

分手后的思念是犯贱 2023-09-27 10:18 109阅读 0赞

**目录**

8.3 基于时间的合流——双流联结（Join）

8.3.1 窗口联结（Window Join）

8.3.2 间隔联结（Interval Join）

8.3.3 窗口同组联结（Window CoGroup）

8.4 本章总结

--------------------

**3. 广播连接流（BroadcastConnectedStream）**

关于两条流的连接，还有一种比较特殊的用法：DataStream 调用.connect()方法时，传入的参数也可以不是一个 DataStream，而是一个“广播流”（BroadcastStream），这时**合并两条流得 到的就变成了一个“广播连接流”（BroadcastConnectedStream）。**

这种连接方式往往用在需要动态定义某些规则或配置的场景。因为规则是实时变动的，所 以我们可以用一个单独的流来获取规则数据；而这些规则或配置是对整个应用全局有效的，所 以不能只把这数据传递给一个下游并行子任务处理，而是要“广播”（broadcast）给所有的并 行子任务。而下游子任务收到广播出来的规则，会把它保存成一个状态，这就是所谓的“广播 状态”（broadcast state）。

广播状态底层是用一个“映射”（map）结构来保存的。在代码实现上，可以直接调用

DataStream 的.broadcast()方法，传入一个**“映射状态描述器”（MapStateDescriptor）说明状态 的名称和类型**，就可以得到规则数据的“广播流”（BroadcastStream）：

MapStateDescriptor<String, Rule> ruleStateDescriptor = new 
    MapStateDescriptor<>(...); 
    BroadcastStream<Rule> ruleBroadcastStream = ruleStream 
     .broadcast(ruleStateDescriptor);

接下来我们就可以将要处理的数据流，与这条广播流进行连接（connect），得到的就是所 谓的“广播连接流”（BroadcastConnectedStream）。基于 BroadcastConnectedStream 调用.process()

方法，就可以同时获取规则和数据，进行动态处理了。

这里既然调用了.process()方法，当然传入的参数也应该是处理函数大家族中一员——如果 对数据流调用过 keyBy 进行了按键分区，那么要传入的就是 KeyedBroadcastProcessFunction； 如果没有按键分区，就传入 BroadcastProcessFunction。

DataStream<String> output = stream 
     .connect(ruleBroadcastStream) 
     .process( new BroadcastProcessFunction<>() {...} );

BroadcastProcessFunction 与 CoProcessFunction 类似，同样是一个抽象类，需要实现两个 方法，针对合并的两条流中元素分别定义处理操作。区别在于这里一条流是正常处理数据，而 另一条流则是要用新规则来更新广播状态，所以对应的两个方法叫作.processElement()和.processBroadcastElement()。源码中定义如下：

public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends 
    BaseBroadcastProcessFunction { 
    ... 
     public abstract void processElement(IN1 value, ReadOnlyContext ctx, 
    Collector<OUT> out) throws Exception; 
     public abstract void processBroadcastElement(IN2 value, Context ctx, 
    Collector<OUT> out) throws Exception; 
    ... 
    }

## 8.3 基于时间的合流——双流联结（Join） ##

### 8.3.1 窗口联结（Window Join） ###

我们希望将两条流的数据进行 合并、且同样针对某段时间进行处理和统计，又该怎么做呢？

Flink 为这种场景专门提供了一个**窗口联结（window join）算子**，可以定义时间窗口，并 **将两条流中共享一个公共键（key）的数据放在窗口中进行配对处理。**

**1. 窗口联结的调用**

窗口联结在代码中的实现，首先需要调用 DataStream 的.join()方法来合并两条流，得到一 个 JoinedStreams；接着通过.where()和.equalTo()方法指定两条流中联结的 key；然后通 过.window()开窗口，并调用.apply()传入联结窗口函数进行处理计算。

stream1.join(stream2) 
     .where(<KeySelector>) 
     .equalTo(<KeySelector>) 
     .window(<WindowAssigner>) 
     .apply(<JoinFunction>)

.where()的参数是键选择器（KeySelector），用来指定第一条流中的 key； 而.equalTo()传入的 KeySelector 则指定了第二条流中的 key。两者相同的元素，如果在同一窗 口中，就可以匹配起来，并通过一个“联结函数”（JoinFunction）进行处理了。

这里.window()传入的就是窗口分配器，之前讲到的三种时间窗口都可以用在这里：滚动窗口（tumbling window）、滑动窗口（sliding window）和会话窗口（session window）。

而后面调用.apply()可以看作实现了一个特殊的窗口函数。注意这里只能调用.apply()，没有其他替代的方法。

传入的 JoinFunction 也是一个函数类接口，使用时需要实现内部的.join()方法。这个方法 有两个参数，分别表示两条流中成对匹配的数据。JoinFunction 在源码中的定义如下：

public interface JoinFunction<IN1, IN2, OUT> extends Function, Serializable { 
     OUT join(IN1 first, IN2 second) throws Exception; 
    }

**2. 窗口联结的处理流程**

两条流的数据到来之后，首先会**按照 key 分组**、**进入对应的窗口中存储**；当**到达窗口结束时间**时，**算子会先统计出窗口内两条流的数据的所有组合**，也就是**对两条流中的数据做一个笛卡尔积**（相当于表的交叉连接，cross join），然后进行遍历，把**每一对匹配的数据，作为参数(first，second)传入 JoinFunction 的.join()方法进行计算处理**。所以窗口中每有一对数据成功联结匹配，JoinFunction 的.join()方法就会被调用一次，并输出一个结果。

![3e0e5dc805be4fc1a3b66d48a103c00c.png][]

> 除了 JoinFunction，在.apply()方法中还可以传入 **FlatJoinFunction**，用法非常类似，只是内部需要实现的**.join()方法没有返回值**。**结果的输出是通过收集器（Collector）来实现的**，所以对于一对匹配数据可以输出任意条结果。
> 
> 如果某个窗口中一条流的数据没有任何另一条流的数据匹配，那么就不会调用 JoinFunction 的.join()方法，也就没有任何输出了。

**3. 窗口联结实例**

在电商网站中，往往需要统计用户不同行为之间的转化，这就需要对不同的行为数据流， 按照用户 ID 进行分组后再合并，以分析它们之间的关联。如果这些是以固定时间周期来统计的，那我们就可以使用窗口 join 来实现这样的需求

package com.atguigu.chapter08;
    
    import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    import org.apache.flink.api.common.functions.JoinFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    
    public class WindowJoinTest {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env =
                    StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            SingleOutputStreamOperator<Tuple2<String, Long>> stream1 = env.fromElements(
                            Tuple2.of("a", 1000L),
                            Tuple2.of("b", 1000L),
                            Tuple2.of("a", 2000L),
                            Tuple2.of("b", 2000L)
                    )
                    .assignTimestampsAndWatermarks(
                            WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                                    .withTimestampAssigner(
                                            new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                                        @Override
                                                        public long extractTimestamp(Tuple2<String,
                                                                Long> stringLongTuple2, long l) {
                                                            return stringLongTuple2.f1;
                                                        }
                                                    }
                                    )
                    );
    
            SingleOutputStreamOperator<Tuple2<String, Integer>> stream2 = env.fromElements(
                            Tuple2.of("a", 3000),
                            Tuple2.of("b", 4000),
                            Tuple2.of("a", 4500),
                            Tuple2.of("b", 5000)
                    )
                    .assignTimestampsAndWatermarks(
                            WatermarkStrategy.<Tuple2<String, Integer>>forMonotonousTimestamps()
                                    .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Integer>>() {
                                                @Override
                                                public long extractTimestamp(Tuple2<String, Integer> stringLongTuple2, long l) {
                                                    return stringLongTuple2.f1;
                                                }
                                            })
                    );
            stream1.join(stream2)
                            .where(data -> data.f0)
                                    .equalTo(data -> data.f0)
                                            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                                                    .apply(new JoinFunction<Tuple2<String, Long>, Tuple2<String, Integer>, String>() {
                                                        @Override
                                                        public String join(Tuple2<String, Long> first, Tuple2<String, Integer> second) throws Exception {
                                                            return first+" -> "+second;
                                                        }
                                                    })
                                                            .print();
    
    
    
    
            env.execute();
        }
    }

输出结果：

(a,1000) -> (a,3000)
    (a,1000) -> (a,4500)
    (a,2000) -> (a,3000)
    (a,2000) -> (a,4500)
    (b,1000) -> (b,4000)
    (b,2000) -> (b,4000)

可以看到，窗口的联结是笛卡尔积。

### 8.3.2 间隔联结（Interval Join） ###

在有些场景下，我们要处理的时间间隔可能并不是固定的。比如，在交易系统中，需要实 时地对每一笔交易进行核验，保证两个账户转入转出数额相等，也就是所谓的“实时对账”。 两次转账的数据可能写入了不同的日志流，它们的时间戳应该相差不大，所以我们可以考虑只 统计一段时间内是否有出账入账的数据匹配。这时显然不应该用滚动窗口或滑动窗口来处理— —**因为匹配的两个数据有可能刚好“卡在”窗口边缘两侧，于是窗口内就都没有匹配了**；会话 窗口虽然时间不固定，但也明显不适合这个场景。 基于时间的窗口联结已经无能为力了。

为了应对这样的需求，Flink 提供了一种叫作“间隔联结”（interval join）的合流操作。顾 名思义，**间隔联结的思路就是针对一条流的每个数据，开辟出其时间戳前后的一段时间间隔， 看这期间是否有来自另一条流的数据匹配。**

**1. 间隔联结的原理**

间隔联结具体的定义方式是，我们**给定两个时间点**，分别叫作**间隔的“上界”**（upperBound） **和“下界”**（lowerBound）；于是对于一条流（不妨叫作 A）中的任意一个数据元素 a，就可以开辟一段时间间隔：**\[a.timestamp + lowerBound, a.timestamp + upperBound\]**,即以 a 的时间戳为 中心，下至下界点、上至上界点的一个闭区间：我们就把这段时间作为可以匹配另一条流数据 的“窗口”范围。所以对于另一条流（不妨叫 B）中的数据元素 b，如果它的时间戳落在了这 个区间范围内，a 和 b 就可以成功配对，进而进行计算输出结果。所以匹配的条件为： a.timestamp + lowerBound

这里需要注意，**做间隔联结的两条流 A 和 B，也必须基于相同的 key**；**下界 lowerBound应该小于等于上界 upperBound**，**两者都可正可负**；间隔联结目前**只支持事件时间语义**。

![c953ece277444300bf3e63fabfd50f39.png][]

**2. 间隔联结的调用**

间隔联结在代码中，是基于 KeyedStream 的联结（join）操作。DataStream 在 **keyBy** 得到KeyedStream 之后，可以调用.**intervalJoin()**来合并两条流，传入的参数同样是一个 KeyedStream， 两者的 **key 类型应该一致**；得到的是一个 **IntervalJoin 类型**。后续的操作同样是完全固定的： 先通过.between()方法指定间隔的上下界，再调用.process()方法，定义对匹配数据对的处理操 作。调用.process()需要传入一个处理函数，这是处理函数家族的最后一员：“处理联结函数”ProcessJoinFunction。

通用调用形式如下：

stream1 
     .keyBy(<KeySelector>) 
     .intervalJoin(stream2.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(left + "," + right); 
     } 
     });

可以看到，抽象类 ProcessJoinFunction 就像是 ProcessFunction 和 JoinFunction 的结合，内 部同样有一个抽象方法.processElement()。与其他处理函数不同的是，它多了一个参数，这自 然是因为有来自两条流的数据。参数中 **left 指的就是第一条流中的数据**，**right 则是第二条流中 与它匹配的数据。**每当检测到一组匹配，就会调用这里的.processElement()方法，经处理转换 之后输出结果。

**3. 间隔联结实例**

在电商网站中，某些用户行为往往会有短时间内的强关联。我们这里举一个例子，我们有 两条流，一条是下订单的流，一条是浏览数据的流。我们可以针对同一个用户，来做这样一个 联结。也就是使用一个用户的下订单的事件和这个用户的最近十分钟的浏览数据进行一个联结 查询。

package com.atguigu.chapter08;
    
    import com.atguigu.chapter05.Event;
    import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    import java.time.Duration;
    
    public class IntervalJoinTest {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            SingleOutputStreamOperator<Tuple3<String, String, Long>> orderStream = env.fromElements(
                    Tuple3.of("Mary", "order-1", 5000L),
                    Tuple3.of("Alice", "order-2", 5000L),
                    Tuple3.of("Bob", "order-3", 20000L),
                    Tuple3.of("Alice", "order-4", 20000L),
                    Tuple3.of("Cary", "order-5", 51000L)
            ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                    .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                        @Override
                        public long extractTimestamp(Tuple3<String, String, Long> stringStringLongTuple3, long l) {
                            return stringStringLongTuple3.f2;
                        }
                    }));
    
            SingleOutputStreamOperator<Event> clickStream = env.fromElements(
                    new Event("Bob", "./cart", 2000L),
                    new Event("Alice", "./prod?id=100", 3000L),
                    new Event("Alice", "./prod?id=200", 3500L),
                    new Event("Bob", "./prod?id=2", 2500L),
                    new Event("Alice", "./prod?id=300", 36000L),
                    new Event("Bob", "./home", 30000L),
                    new Event("Bob", "./prod?id=1", 23000L),
                    new Event("Bob", "./prod?id=3", 33000L)
            ).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                    .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                        @Override
                        public long extractTimestamp(Event event, long l) {
                            return event.timestamp;
                        }
                    }));
    
            orderStream.keyBy(data -> data.f0)
                            .intervalJoin(clickStream.keyBy(data -> data.user))
                                    .between(Time.seconds(-5),Time.seconds(10))
                                            .process(new ProcessJoinFunction<Tuple3<String, String, Long>, Event, String>() {
                                                @Override
                                                public void processElement(Tuple3<String, String, Long> left, Event right, ProcessJoinFunction<Tuple3<String, String, Long>, Event, String>.Context context, Collector<String> collector) throws Exception {
                                                    collector.collect(right+ " => "+left);
                                                }
                                            })
                                                    .print();
    
            env.execute();
        }
    }

### 8.3.3 窗口同组联结（Window CoGroup） ###

除窗口联结和间隔联结之外，Flink 还提供了一个“窗口同组联结”（window coGroup）操 作。它的用法跟 window join 非常类似，**也是将两条流合并之后开窗处理匹配的元素**，调用时 只需要将.join()换为.coGroup()就可以了

stream1.coGroup(stream2) 
     .where(<KeySelector>) 
     .equalTo(<KeySelector>) 
     .window(TumblingEventTimeWindows.of(Time.hours(1))) 
     .apply(<CoGroupFunction>)

与 window join 的区别在于，调用.apply()方法定义具体操作时，传入的是一个CoGroupFunction。这也是一个函数类接口，源码中定义如下：

public interface CoGroupFunction<IN1, IN2, O> extends Function, Serializable { 
     void coGroup(Iterable<IN1> first, Iterable<IN2> second, Collector<O> out) 
    throws Exception; 
    }

内部的**.coGroup()方法**，有些类似于 FlatJoinFunction 中.join()的形式，同样有**三个参数**， **分别代表两条流中的数据以及用于输出的收集器（Collector）**。不同的是，这里的前两个参数不再是单独的每一组“配对”数据了，而是传入了可遍历的数据集合。也就是说，现在不会再去计算窗口中两条流数据集的笛卡尔积，而是直接把收集到的所有数据一次性传入，至于要怎 样配对完全是自定义的。这样**.coGroup()方法只会被调用一次，而且即使一条流的数据没有任 何另一条流的数据匹配，也可以出现在集合中、当然也可以定义输出结果了。**

> 所以能够看出，coGroup 操作比窗口的 join 更加通用，不仅可以实现类似 SQL 中的“内 连接”（inner join），也可以实现左外连接（left outer join）、右外连接（right outer join）和全外 连接（full outer join）。事实上，窗口 join 的底层，也是通过 coGroup 来实现的。

package com.atguigu.chapter08;
    
    import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    import org.apache.flink.api.common.functions.CoGroupFunction;
    import org.apache.flink.api.common.functions.JoinFunction;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    public class CoGroupTest {
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env =
                    StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            SingleOutputStreamOperator<Tuple2<String, Long>> stream1 = env.fromElements(
                            Tuple2.of("a", 1000L),
                            Tuple2.of("b", 1000L),
                            Tuple2.of("a", 2000L),
                            Tuple2.of("b", 2000L)
                    )
                    .assignTimestampsAndWatermarks(
                            WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                                    .withTimestampAssigner(
                                            new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                                @Override
                                                public long extractTimestamp(Tuple2<String,
                                                        Long> stringLongTuple2, long l) {
                                                    return stringLongTuple2.f1;
                                                }
                                            }
                                    )
                    );
    
            SingleOutputStreamOperator<Tuple2<String, Integer>> stream2 = env.fromElements(
                            Tuple2.of("a", 3000),
                            Tuple2.of("b", 4000),
                            Tuple2.of("a", 4500),
                            Tuple2.of("b", 5000)
                    )
                    .assignTimestampsAndWatermarks(
                            WatermarkStrategy.<Tuple2<String, Integer>>forMonotonousTimestamps()
                                    .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Integer>>() {
                                        @Override
                                        public long extractTimestamp(Tuple2<String, Integer> stringLongTuple2, long l) {
                                            return stringLongTuple2.f1;
                                        }
                                    })
                    );
    
            stream1.coGroup(stream2)
                            .where(data -> data.f0)
                                    .equalTo(data -> data.f0)
                                            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                                                    .apply(new CoGroupFunction<Tuple2<String, Long>, Tuple2<String, Integer>, String>() {
                                                        //coGroup方法参数：两个数据源的集合，以及一个输出的数据类型
                                                        @Override
                                                        public void coGroup(Iterable<Tuple2<String, Long>> first, Iterable<Tuple2<String, Integer>> second, Collector<String> collector) throws Exception {
                                                            //在一个窗口中，一个集合根据key匹配另外一个集合中的数据
                                                            collector.collect(first+ " => "+second);
                                                        }
                                                    })
                                                            .print();
            
            env.execute();
        }
    }

输出:

[(a,1000), (a,2000)] => [(a,3000), (a,4500)]
    [(b,1000), (b,2000)] => [(b,4000)]
    [] => [(b,5000)]

## 8.4 本章总结 ##

多流转换是流处理在实际应用中常见的需求，主要包括分流和合流两大类，本章分别做了 详细讲解。在 Flink 中，**分流操作可以通过处理函数的侧输出流（side output）很容易地实现； 而合流则提供不同层级的各种 API。**

最基本的合流方式是**联合（union）**和**连接（connect）**，两者的主要区别在于 **union 可以对 多条流进行合并，数据类型必须一致**；而 **connect 只能连接两条流，数据类型可以不同**。事实 上 connect 提供了最底层的处理函数（process function）接口，可以通过状态和定时器实现任意自定义的合流操作，所以是最为通用的合流方式。

除此之外，Flink 还提供了内置的几个联结（join）操作，它们都是基于某个时间段的双流 合并，是需求特化之后的高层级 API。**主要包括窗口联结（window join）、间隔联结（interval join）**

和窗口同组联结（window coGroup）。其中 window join 和 coGroup 都是基于时间窗口的操作， 窗口分配器的定义与之前介绍的相同，而窗口函数则被限定为一种，通过.apply()来调用；

> **interval join 则与窗口无关，而是基于每个数据元素截取对应的一个时间段来做联结，最终的 处理操作则需调用.process()，由处理函数 ProcessJoinFunction 实现。**

可以看到，基于时间的联结操作的每一步操作都是固定的接口，并没有其他变化，使用起 来“专项专用”，非常方便。

[3e0e5dc805be4fc1a3b66d48a103c00c.png]: https://img-blog.csdnimg.cn/3e0e5dc805be4fc1a3b66d48a103c00c.png
[c953ece277444300bf3e63fabfd50f39.png]: https://img-blog.csdnimg.cn/c953ece277444300bf3e63fabfd50f39.png