Window Join（窗口连接）
Window Join 连接共享公共 Key 并位于同一窗口中的两个流的数据元。可以使用 WindowAssigner 定义这些窗口，并对来自两个流的数据元共同进行计算。
然后将来自双方的数据元传递给用户定义的 JoinFunction 或者 FlatJoinFunction，可以发出满足连接条件的结果。
一般用法如下：
stream.join(otherStream)
.where(<KeySelector>)
.equalTo(<KeySelector>)
.window(<WindowAssigner>)
.apply(<JoinFunction>)
关于语义的一些注释：
- 两个流的数据元的成对组合的创建表现得像内连接（inner-join），这意味着如果一个流中的数据元没有连接到另一个流的对应数据元，则不会发出这些数据元。
- 那些关联成功的数据元，其时间戳中为仍位于相应窗口中的最大时间戳。例如，窗口 [5, 10) 将导致连接的数据元使用9作为其时间戳。
Tumbling Window Join（滚动窗口连接）
当执行滚动窗口连接时，具有公共 Key 和公共窗口的所有数据元以成对组合的形式创建，并传递给 JoinFunction 或 FlatJoinFunction。连接行为类似于 inner join，发出的数据元都是连接成功的。
如图所示，定义一个大小为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 和公共窗口的所有数据元以成对组合的形式创建，并传递给 JoinFunction 或 FlatJoinFunction。同样的发出的数据元都是连接成功的，有些元素可能会连接到一个滑动窗口中，但不会连接到另一个滑动窗口中。
如图所示，定义一个大小为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 的所有数据元以成对组合的形式创建，并传递给 JoinFunction 或 FlatJoinFunction。如果会话窗口只包含一个流中的元素，依然不会发出任何输出。
如图所示，定义一个会话窗口，其中每个会话有至少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 时，它们将被分配两个元素的较大时间戳。
间隔连接当前只支持事件时间
在上面的例子中，我们连接橙色和绿色两个流，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
