flink dataset 和datastream（Flink教程DataStream上的Join操作）
flink dataset 和datastream（Flink教程DataStream上的Join操作）一般滴，INNER JOIN只对两个数据源都出现的元素做Join，形成一个数据对，即数据源input1中的某个元素与数据源input2中的所有元素逐个配对。当数据源某个窗口内没数据时，比如图中的第三个窗口，Join的结果也是空的。窗口内数据INNER JOIN示意图input1.join(input2) .where(<KeySelector>) <- input1使用哪个字段作为Key .equalTo(<KeySelector>) <- input2使用哪个字段作为Key .window(<WindowAssigner>) <- 指定WindowAssigner [.trigger(<Trigger>)] <- 指定Trigger（可选） [.ev
批处理经常要解决的问题是将两个数据源做关联Join操作。比如，很多手机APP都有一个用户数据源User，同时APP会记录用户的行为，我们称之为Behavior，两个表按照userId来进行Join。在流处理场景下，Flink也支持了Join，只不过Flink是在一个时间窗口上来进行两个表的Join。
Join示例图
目前，Flink支持了两种Join：window Join（窗口连接）和Interval Join（时间间隔连接。
Window Join从名字中能猜到，Window Join主要在Flink的窗口上进行操作，它将两个流中落在相同窗口的元素按照某个Key进行Join。一个Window Join的大致骨架结构为：
input1.join(input2)
.where(<KeySelector>) <- input1使用哪个字段作为Key
.equalTo(<KeySelector>) <- input2使用哪个字段作为Key
.window(<WindowAssigner>) <- 指定WindowAssigner
[.trigger(<Trigger>)] <- 指定Trigger（可选）
[.evictor(<Evictor>)] <- 指定Evictor（可选）
.apply(<JoinFunction>) <- 指定JoinFunction
下图展示了Join的大致过程。两个输入数据流先分别按Key进行分组，然后将元素划分到窗口中。窗口的划分需要使用WindowAssigner来定义，这里可以使用Flink提供的滚动窗口、滑动窗口或会话窗口等默认的WindowAssigner。随后两个数据流中的元素会被分配到各个窗口上，也就是说一个窗口会包含来自两个数据流的元素。相同窗口内的数据会以INNER JOIN的语义来相互关联，形成一个数据对。当窗口的时间结束，Flink会调用JoinFunction来对窗口内的数据对进行处理。当然，我们也可以使用Trigger或Evictor做一些自定义优化，他们的使用方法和普通窗口的使用方法一样。
Join的大致流程
接下来我们重点分析一下两个数据流是如何INNER JOIN的：
窗口内数据INNER JOIN示意图
一般滴，INNER JOIN只对两个数据源都出现的元素做Join，形成一个数据对，即数据源input1中的某个元素与数据源input2中的所有元素逐个配对。当数据源某个窗口内没数据时，比如图中的第三个窗口，Join的结果也是空的。
class MyJoinFunction extends JoinFunction[(String Int) (String Int) String] {
override def join(input1: (String Int) input2: (String Int)): String = {
"input 1 :" input1._2 " input 2 :" input2._2
}
}
val input1: DataStream[(String Int)] = ...
val input2: DataStream[(String Int)] = ...
val joinResult = input1.join(input2)
.where(i1 => i1._1)
.equalTo(i2 => i2._1)
.window(TumblingprocessingTimeWindows.of(Time.seconds(60)))
.apply(new MyJoinFunction)
上面的代码自定义了JoinFunction，并将Join结果打印出来。无论代码中演示的滚动窗口，还是滑动窗口或会话窗口，其原理都是一样的。除了JoinFunction，Flink还提供了FlatJoinFunction，其功能是输出零到多个结果。
如果INNER JOIN不能满足我们的需求，CoGroupFunction提供了更多可自定义的功能。需要注意的是，在调用时，要写成input1.coGroup(input2).where(<KeySelector>).equalTo(<KeySelecotr>)。
class MyCoGroupFunction extends CoGroupFunction[(String Int) (String Int) String] {
// 这里的类型是Java的Iterable，需要引用 collection.JavaConverters._ 并转成Scala
override def coGroup(input1: lang.Iterable[(String Int)] input2: lang.Iterable[(String Int)] out: Collector[String]): Unit = {
input1.asScala.foreach(element => out.collect("input1 :" element.toString()))
input2.asScala.foreach(element => out.collect("input2 :" element.toString()))
}
}
val input1: DataStream[(String Int)] = ...
val input2: DataStream[(String Int)] = ...
val coGroupResult = input1.coGroup(input2)
.where(i1 => i1._1)
.equalTo(i2 => i2._1)
.window(TumblingProcessingTimeWindows.of(Time.seconds(60)))
.apply(new MyCoGroupFunction)Interval Join
与Window Join不同，Interval Join不依赖Flink的WindowAssigner，而是根据一个时间间隔（Interval）界定时间。Interval需要一个时间下界（lower bound）和上界（upper bound），如果我们将input1和input2进行Interval Join，input1中的某个元素为input1.element1，时间戳为input1.element1.ts，那么一个Interval就是[input1.element1.ts lower bound input1.element1.ts upper bound]，input2中落在这个时间段内的元素将会和input1.element1组成一个数据对。用数学公式表达为，凡是符合下面公式input1.element1.ts lower bound <= input2.elementx.ts <=input1.element1.ts upper bound的元素使用INNER JOIN语义，两两组合在一起。上下界可以是正数也可以是负数。
注意，目前Flink（1.9）的Interval Join只支持Event Time语义。
Interval Join示意图
下面的代码展示了如何对两个数据流进行Interval Join：
class MyProcessFunction extends ProcessJoinFunction[(String Long Int) (String Long Int) String] {
override def processElement(input1: (String Long Int)
input2: (String Long Int)
context: ProcessJoinFunction[(String Long Int) (String Long Int) String]#Context
out: Collector[String]): Unit = {
out.collect("input 1: " input1.toString() " input 2: " input2.toString)
}
}
// 数据流有三个字段：（key 时间戳 数值）
val input1: DataStream[(String Long Int)] = ...
val input2: DataStream[(String Long Int)] = ...
val intervalJoinResult = input1.keyBy(_._1)
.intervalJoin(input2.keyBy(_._1))
.between(Time.milliseconds(-5) Time.milliseconds(10))
.process(new MyProcessFunction)
默认的时间间隔是包含上下界的，我们可以使用.lowerBoundExclusive() 和.upperBoundExclusive来确定是否需要包含上下界。
val intervalJoinResult = input1.keyBy(_._1)
.intervalJoin(input2.keyBy(_._1))
.between(Time.milliseconds(-5) Time.milliseconds(10))
.upperBoundExclusive()
.lowerBoundExclusive()
.process(new MyProcessFunction)
Interval Join内部是用缓存来存储所有数据的，因此需要注意缓存数据不能太大，以免对内存造成绝大压力。
