黑龙江省建设会计协会网站wordpress 默认自带插件

        在实际应用中，我们可能要将多个不同来源的数据连接合并在一起进行处理，也有可能要将一条流拆分成多条流进行处理，这就涉及到了Flink的多流转换问题。简单来说，就是分流和合流两大操作，分流主要通过侧输出流实现，合流的算子就比较丰富了，有union、connect、join等。

一、分流

        所谓分流，就是通过定义一些筛选条件，将一个dataStream拆分成多个子dataStream的过程，每条子数据流之间完全独立。Flink中的分流主要通过侧输出流来实现。

        通过调用底层的处理函数，可以获取到上下文信息，调用上下文的.output方法就可以实施分流操作了。.output方法需要传入一个“输出标签"(OutputTag)，用来标记侧输出流（相当于给侧输出流盖了个戳，指明他的名称和类型），之后也可以通过.getSideOutput()方法传入OutputTag获取到相应的侧输出流。

二、合流

        对多个来源的多条流进行联合处理时，需要用到合流操作，具体有如下几种合流算子：

1. union

        union操作要求不同流中的数据类型必须一致, 类似sql语言中的union，是纵向的合并。对datastream调用.union方法即可实现多流合并，合并后的流类型仍然是datastream。这里要注意，多条流合并后的水位线应以最小的那个为准（类似多个并行子任务向下游传递）。

stream1.union(stream2, stream3, ...)

2. connect

        union操作简单，但要求流的数据类型一致，实际应用中实用性不高。针对两条数据类型不一样的流，Flink还提供了connect合流操作，connect操作只能连接两条流。

(1) 两个dataStream进行connect -> 连接流（ConnectedStreams）

        对于两条数据类型不一致的dataStream进行连接，调用.connect()方法，所得到的是一个连接流ConnectedStreams，然后再调用同处理方法分别对两条流进行处理，得到一个统一类型的dataStream。这里的同处理方法可以是map、flatmap也可以是底层的处理函数process，只是在传入参数时跟以往的单流不同，如map方法传入的不再是MapFunction而是CoMapFunction，可以实现对两条流分别做map操作。

        对ConnectedStreams也可以先调用keyBy进行按键分区操作后，再调用同处理方法。这里调用KeyBy后得到的仍然是ConnectedStreams，keyBy要传入两个参数keySelector1和keySelector2类似于sql中两表之间的 join操作的关联字段。

connectedStreams.keyBy(keySelector1, keySelector2);

(2) dataStream与广播流（broadcastStream）进行connect -> 广播连接流

       当需要动态定义某些规则或配置时，如维度表配置信息是动态变化的，存储在MySQL数据库中，我们用maxwell实时对它进行了监控，当发生变化时，这个配置信息是要完整的告知原始数据流的（从业务数据库中抽取的原始数据），即若原始数据流分为了多个并行子任务，则每个并行子任务上都应该知道配置信息的变化，因此需要对配置信息进行广播连接。

        对dataStream调用.broadcast()方法就可以得到广播流，将要处理的数据流与这条广播流进行connect，得到的就是广播连接流，可以调用.process方法进行动态处理，同样要实现的是一个类似CoProcessFunction的抽象类，对两条流分别进行处理。

3. join

        connect方法已经能够实现各种需求了，但是其支持的处理函数太过于底层，在很多场景下太过于抽象了，flink还为datastream提供了内置的join算子和coGroup算子来简化一些特定场景下的合流操作。

(1) 窗口联结（window join)

        当我们不仅需要对两条流进行连接，还需要对连接后的流进行窗口操作，Flink为这种场景专门提供了一个窗口联结算子。如下操作可将两条流基于联结字段进行配对，并将key相同的放入一个窗口进行窗口计算。

stream1.join(stream2).where(<KeySelector>)    // stream1的联结字段.equalTo(<KeySelector>)    // stream2的联结字段.window(<WindowAssigner>).apply(<JoinFunction>)

        注意 这里调用窗口函数只能通过.apply()方法。

        窗口join的具体流程如下：两条流根据key进行分组，分别进入对应的窗口存储；到达窗口时间时，会先统计窗口内两条流的笛卡尔积，然后进行遍历，遍历到一对匹配的数据就调用一次窗口函数并输出结果。

(2) 间隔联结（interval join）

        间隔联结为数据流中的每一条数据单独开辟属于自己的时间窗口。试想这样一个场景，对于一条流A中的一条数据a，它只想和自己时间戳的前后一段时间间隔的B数据流进行连接，这样窗口联结就无法做到，需要间隔联结。

        间隔联结的两条流必须基于相同的key，且需要给定间隔上界和间隔下界，则数据a的窗口大小就是[a.timestamp+lowbound, a.timestamp+upperbound]，其中lowbound<upperbound，两者都可正可负。

stream1.keyBy(<KeySelector>).intervalJoin(stream2.keyBy(<KeySelector>)).between(Time.milliseconds(-2), Time.milliseconds(1)).process(new ProcessJoinFunction(){})

4. coGroup

        coGroup 与窗口联结类似，也是将两条流合并后开窗处理匹配元素，调用时只需将.join()方法换成.coGroup()方法即可。

stream1.coGroup(stream2).where(<KeySelector>)    // stream1的联结字段.equalTo(<KeySelector>)    // stream2的联结字段.window(<WindowAssigner>).apply(<CoGroupFunction>)

        在window join中，数据在窗口中是先做笛卡尔积，再遍历是否匹配, 只有匹配的数据才会去调用apply方法，因此，window join实现的是类似sql中的inner join功能。而在coGroup函数中，数据不会做笛卡尔积，而是将所有搜集到的数据都传入到apply方法中，用户可以自定义匹配逻辑，因此可以实现任意外连接或是其他用户想要的连接方式。