长春网站建设 信赖吉网传媒,wordpress默认摘要,西宁百姓网免费发布信息,wordpress相册主题在实际应用中#xff0c;我们可能要将多个不同来源的数据连接合并在一起进行处理#xff0c;也有可能要将一条流拆分成多条流进行处理#xff0c;这就涉及到了Flink的多流转换问题。简单来说#xff0c;就是分流和合流两大操作#xff0c;分流主要通过侧输出流实现#x… 在实际应用中我们可能要将多个不同来源的数据连接合并在一起进行处理也有可能要将一条流拆分成多条流进行处理这就涉及到了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后得到的仍然是ConnectedStreamskeyBy要传入两个参数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.timestamplowbound, a.timestampupperbound]其中lowboundupperbound两者都可正可负。
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方法中用户可以自定义匹配逻辑因此可以实现任意外连接或是其他用户想要的连接方式。
