flink多流关联方法

  |   0 评论   |   0 浏览

背景

在业务中遇到了，需要实时的将多个流关联起来，在一定的窗口内，按某一个字段聚合成一条记录。

Flink中的多个流关联方式有 join, union, cogroup, connect等。他们的区别的使用场景是什么呢？

三种窗口

  • 滚动窗口（Tumbling Window）
  • 滑动窗口（Sliding Window
  • 会话窗口（Session Window）

滚动窗口

依据固定的窗口长度对数据进行切片。

特点：时间对齐，窗口长度固定，没有重叠。

滚动窗口分配器将每个元素分配到一个特定的窗口大小的窗口中，滚动窗口有一个固定的大小，并且不会出现重叠。例如：如果你指定了一个5分钟大小的滚动窗口，窗口的创建如下图所示：

2020051217040985.png

适用场景：适合做BI统计等。（做每个时间段的聚合计算）。

滑动窗口

滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成。

特点：时间对齐，窗口长度固定，可以有重叠。

滑动窗口分配器将元素分配到固定长度的窗口中，与滚动窗口类似，窗口的大小由窗
口大小参数来配置，另一个窗口滑动参数控制滑动窗口滑动的步长。

若滑动窗口的滑动参数小于窗口大小，则窗口出现重叠。在这种情况下，元素被分配到多个窗口中。

例如：若你有10秒钟的窗口和5秒钟的滑动，则每个窗口中5秒钟的窗口里包含着上个10秒钟产生的数据，如下图：

20200512170442625.png

适用场景：对最近一个时间段内的统计（根据某接口最近5分钟的失败率来决定是否报警。）

会话窗口

由一系列事件组合一个指定时间长度的timeout间隙组成，类似于web应用的session，也就是一段时间没有接收到新数据就会生成新的窗口。

特点：时间无对齐。

会话窗口分配器通过session活动来对元素进行分组，会话窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况。相反，当它在一个固定的时间周期内不再收到元素，即非活动间隔产生，那这个窗口就会关闭。一个会话窗口通过一个session间隔来配置，这个session间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去。

20200512170519416.png

窗口API

// Keyed Window
stream
       .keyBy(...)               <-  按照一个Key进行分组
       .window(...)              <-  将数据流中的元素分配到相应的窗口中
      [.trigger(...)]            <-  指定触发器Trigger（可选）
      [.evictor(...)]            <-  指定清除器Evictor(可选)
       .reduce/aggregate/process()      <-  窗口处理函数Window Function
 
// Non-Keyed Window
stream
       .windowAll(...)           <-  不分组，将数据流中的所有元素分配到相应的窗口中
      [.trigger(...)]            <-  指定触发器Trigger（可选）
      [.evictor(...)]            <-  指定清除器Evictor(可选)
       .reduce/aggregate/process()      <-  窗口处理函数Window Function

TimeWindow

TimeWindow是将指定时间范围内的所有数据组成一个Window，一次对一个Window里面的所有数据进行计算。

滚动窗口

Flink默认的时间窗口根据Processing Time进行窗口的划分，将Flink获取到的数据根据进入Flink的时间划分到不同的窗口中。

val minTempPerWindow = dataStream
.map(data => (data.id, data.temperature) )
.keyBy(_._1)
.timeWindow(Time.seconds(15))
.reduce((d1,d2) => (d1._1,d1._2.min(d2._2)))

时间间隔可以通过Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。

滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。

下面代码中的sliding_size设置为了5s，也就是说，窗口每5s就计算一次，每一次计算的window范围是15s内的所有元素。

val minTempPerWindow = dataStream
      .map(data => (data.id, data.temperature) )
      .keyBy(_._1)
      .timeWindow(Time.seconds(15),Time.seconds(5))
      .reduce((d1,d2) => (d1._1,d1._2.min(d2._2)))

时间间隔可以通过Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其中的一个来指定。

会话窗口

CountWindow

CountWindow根据窗口中相同key元素的数量来触发执行，执行时只计算元素数量达到窗口大小的key对应的结果。

注：CountWindow的window_size指的是相同key的元素的个数，不是输入的所有元素的总数。

滚动窗口

默认的CountWindow是一个滚动的窗口，只需要指定窗口大小即可，当元素数量达到窗口大小时，就会触发窗口的执行。

val minTempPerWindow = dataStream
     .map(data => (data.id, data.temperature) )
     .keyBy(_._1)
.countWindow(5)
.reduce((d1,d2) => (d1._1,d1._2.min(d2._2)))

滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。

下面代码中的sliding_size设置为2，也就是说，每收到两个相同key的数据就计算一次，每一次计算的window范围是5个元素。

val minTempPerWindow = dataStream
     .map(data => (data.id, data.temperature) )
     .keyBy(_._1)
.countWindow(5,2)
.reduce((d1,d2) => (d1._1,d1._2.min(d2._2)))

window function

Window function定义了要对窗口中收集的数据做的计算操作，主要可以分为两类：
① 增量聚合函数：每条数据到来就进行计算，保持一个简单的状态。典型的增量聚合函数有ReduceFunction，AggreateFunction。

② 全窗口函数：先把窗口所有数据收集起来，等到计算的时候会遍历所有数据。典型的全窗口函数有ProcessWindowFunction。

其他可选API

触发器——.trigger()

定义window什么时候关闭，触发计算并输出结果。
注：GlobalWindow默认的触发器时NeverTrigger，该触发器从不出发，所以在使用GlobalWindow时必须自定义触发器。

移除器——.evitor()

定义移除某些数据的逻辑。Evictor可以在触发器触发之后以及窗口函数被应用之前或之后可选择地移除元素。使用Evictor可以防止预聚合，因为窗口的所有元素都必须在应用计算逻辑之前先传给Evictor进行处理。

允许处理迟到多久的数据——.allowedLateness()

将迟到的数据放入侧输入流——.sideOutputLateData()

获取侧输出流——.getSideOutput()

介绍

  • cogroup: 按key做group，没有匹配上的也会输出。
  • join：类比数据库中的inner join，按key做group，没有匹配上的不输出。
  • connect：通常使用于处理一个BroadCastStream和另一个DataStream中的数据。

cogroup

该操作是将两个数据流/集合按照key进行group，然后将相同key的数据进行处理，但是它和join操作稍有区别，它在一个流/数据集中没有找到与另一个匹配的数据还是会输出。

Join操作

有四种

  1. Tumbling Window Join
  2. Sliding Window Join
  3. Session Window Join
  4. Interval Join

编程模型如下：

stream.join(otherStream)
    .where(<KeySelector>)
    .equalTo(<KeySelector>)
    .window(<WindowAssigner>)
    .apply(<JoinFunction>)

Interval Join会将两个数据流按照相同的key，并且在其中一个流的时间范围内的数据进行join处理。通常用于把一定时间范围内相关的分组数据拉成一个宽表。我们通常可以用类似下面的表达式来使用interval Join来处理两个数据流：

key1 == key2 && e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound

附录

随笔一

Flink怎么实现多个实时流的关联。官方文档，很多教程 给的Demo都是两个实时流的关联。

  1. 分析业务数据源，很多需要多个流的join的场景 是伪命题，用union即可。
  2. union + group by ，在基于key的流中可以取代join。优势： 在join 发生数据倾斜或者反压，很难 checkpoint时，用union可以回避这个问题。
  3. 例如三个流join，可以 tempstream = stream1.join(stream2) ResultStream = tempstream.join(stream3)。语法支持，看了下生成的图，不确定是不是想要的效果。
  4. 将第三点的 join 换成 cogroup操作。这个是社区直播中，提问多流join后 得到的回复
  5. 基于blink引擎，多表join 直接用sql表达出来。select * from table1 a left join table2 b on a.id = b.id left join table3 c on b.id = c.id 这个方案也是在社区直播，提问多流join后 得到的回复。

参考