Flink Window Join

Window Join(窗口连接)

Window Join 连接共享公共 Key 并位于同一窗口中的两个流的数据元。可以使用 WindowAssigner 定义这些窗口,并对来自两个流的数据元共同进行计算。

然后将来自双方的数据元传递给用户定义的 JoinFunction 或者 FlatJoinFunction,可以发出满足连接条件的结果。

一般用法如下:

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

关于语义的一些注释:

  1. 两个流的数据元的成对组合的创建表现得像内连接(inner-join),这意味着如果一个流中的数据元没有连接到另一个流的对应数据元,则不会发出这些数据元。
  2. 那些关联成功的数据元,其时间戳中为仍位于相应窗口中的最大时间戳。例如,窗口 [5, 10) 将导致连接的数据元使用9作为其时间戳。

Tumbling Window Join(滚动窗口连接)

当执行滚动窗口连接时,具有公共 Key 和公共窗口的所有数据元以成对组合的形式创建,并传递给 JoinFunctionFlatJoinFunction。连接行为类似于 inner join,发出的数据元都是连接成功的。

Tumbling

如图所示,定义一个大小为2毫秒的滚动窗口,产生窗口 [0,1] [2,3] …,图中显示了每个窗口中所有数据元的成对组合,这些数据元将被传递给 JoinFunction。在窗口 [6,7] 中没有任何数据被发出,因为绿色流中不存在与橙色流数据元连接的数据元。

import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

...

val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...

orangeStream.join(greenStream)
    .where(elem => /* select key */)
    .equalTo(elem => /* select key */)
    .window(TumblingEventTimeWindows.of(Time.milliseconds(2)))
    .apply { (e1, e2) => e1 + "," + e2 }

Sliding Window Join(滑动窗口连接)

执行滑动窗口连接时,具有公共 Key 和公共窗口的所有数据元以成对组合的形式创建,并传递给 JoinFunctionFlatJoinFunction。同样的发出的数据元都是连接成功的,有些元素可能会连接到一个滑动窗口中,但不会连接到另一个滑动窗口中。

Sliding

如图所示,定义一个大小为2毫秒的滚动窗口,并滑动一毫秒,产生窗口 [-1, 0] [0,1] [1,2] [2,3]…,图中显示了每个窗口中所有数据元的成对组合,这些数据元将被传递给 JoinFunction。橙色2在窗口 [1,2] 中没有连接数据,在窗口 [2,3] 中与绿色3连接。

import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

...

val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...

orangeStream.join(greenStream)
    .where(elem => /* select key */)
    .equalTo(elem => /* select key */)
    .window(SlidingEventTimeWindows.of(Time.milliseconds(2) /* size */, Time.milliseconds(1) /* slide */))
    .apply { (e1, e2) => e1 + "," + e2 }

Session Window Join(会话窗口连接)

在执行会话窗口连接时,具有相同 Key 的所有数据元以成对组合的形式创建,并传递给 JoinFunctionFlatJoinFunction。如果会话窗口只包含一个流中的元素,依然不会发出任何输出。

Session

如图所示,定义一个会话窗口,其中每个会话有至少1ms的间隙。图中显示有三个会话,在前两个会话中,两个流的连接数据元都传递给 JoinFunction。在第三个会话中,没有发出任何输出。

import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

...

val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...

orangeStream.join(greenStream)
    .where(elem => /* select key */)
    .equalTo(elem => /* select key */)
    .window(EventTimeSessionWindows.withGap(Time.milliseconds(1)))
    .apply { (e1, e2) => e1 + "," + e2 }

Interval Join(间隔连接)

间隔连接用一个公共 Key 连接两个流的数据元(现在将它们称为A和B),其中B流的元素具有时间戳,该时间戳位于A流中数据元的时间戳的相对时间间隔中。

这也可以更正式地表达为 b.timestamp ∈ [a.timestamp + lowerBound; a.timestamp + upperBound]a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound

其中a和b是有公共 Key 的A和B流的数据元。lower bound 和 upper bound 可以是负数或正数,只要 lower bound 总是小于或等于 upper bound。该连接行为同样是 inner join。

当一对数据元传递给 ProcessJoinFunction 时,它们将被分配两个元素的较大时间戳。

间隔连接当前只支持事件时间

Interval

在上面的例子中,我们连接橙色和绿色两个流,lower bound 为-2毫秒,upper bound 为+1毫秒。这些界限默认是包含的,但 .lowerBoundExclusive().upperBoundExclusive 可以改变默认行为。

orangeElem.ts + lowerBound <= greenElem.ts <= orangeElem.ts + upperBound

import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;

...

val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...

orangeStream
    .keyBy(elem => /* select key */)
    .intervalJoin(greenStream.keyBy(elem => /* select key */))
    .between(Time.milliseconds(-2), Time.milliseconds(1))
    .process(new ProcessJoinFunction[Integer, Integer, String] {
        override def processElement(left: Integer, right: Integer, ctx: ProcessJoinFunction[Integer, Integer, String]#Context, out: Collector[String]): Unit = {
         out.collect(left + "," + right); 
        }
      });
    });

Reference:
https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/stream/operators/joining.html

Add a Comment

电子邮件地址不会被公开。 必填项已用*标注

16 + 11 =