当前Spark Streaming-Streaming Join只支持：
- InnerJoin;
- LeftJoin;
- RightJoin;
整体思路
-
将Join的条件分为：
-
preJoinFilter
- leftSideOnly: 只依赖左表的过滤条件，针对左表input输入，先校验该条件，如果不满足该条件一定不会关联上；
- rightSideOnly：只依赖右表的过滤条件，针对右表input输入，先校验该条件，如果不满足该条件一定不会关联上；
postJoinFilter：同时依赖左右表的过滤条件，在满足preJoinFilter并同另外一侧表关联后进行该过滤；
-
-
将满足过滤条件的新增左表数据跟右表状态数据做Join（详细见代码），同时更新所有的新增left表数据到状态 ，将结果输出至leftOutputIter，结果分为两部分：
关联上并满足过滤条件的数据：这部分数据Inner/Left/Right都是相同的；
-
关联不上的数据，分为两种情况：
暂时没有关联上的，这个时候仅仅将左流新增数据写入左表状态表，不会emit数据；
-
永不会关联上的（不满足leftSideOnly），针对Inner/Left/Right产生不同的结果：
a. LeftJoin: 输出join with null;
b. Inner/RightJoin: 输出空；
将满足过滤条件的新增右表数据跟左表状态数据做Join，同时更新所有的新增right表数据到状态，同时会生成新增左表跟新增右表的关联数据（因为左表的新增数据已经保留到了左表的状态数据中），结果输出至rightOutputIter(类似于上步骤)；
针对InnerJoin的输出即为leftOutputIter + rightOutputIter;
-
针对LeftJoin同时还需要考左表已经过期的数据，这些数据分两种情况：
- 同右表状态数据（包括当前批次）没有关联：这些数据应该join with null；
- 同右表状态数据（包括当前批次）有关联： 应该忽略掉，因为这些数据理论上在某些时间点上会Join上，所以不能join with null;
- 同时这些过期数据在该批次会被清理掉；
针对RightJoin同时还需要考左表已经过期的数据，类似LeftOuterJoin；
左右表关联代码实现
def storeAndJoinWithOtherSide(
otherSideJoiner: OneSideHashJoiner)(
generateJoinedRow: (InternalRow, InternalRow) => JoinedRow):
Iterator[InternalRow] = {
// Step1: 过滤不符合watermark要求的数据
val watermarkAttribute = inputAttributes.find(_.metadata.contains(delayKey))
val nonLateRows =
WatermarkSupport.watermarkExpression(watermarkAttribute, eventTimeWatermark) match {
case Some(watermarkExpr) =>
val predicate = newPredicate(watermarkExpr, inputAttributes)
inputIter.filter { row => !predicate.eval(row) }
case None =>
inputIter
}
nonLateRows.flatMap { row =>
val thisRow = row.asInstanceOf[UnsafeRow]
// Step2: 如果输入不满足preJoinFilter条件，即针对Left表不满足只依赖左表的Join条件时：
// - 这种场景下该Row不会满足Join的条件，所以不会保存到状态数据中；
// - 同时根据Join的类型生成关联不上后的数据；
if (preJoinFilter(thisRow)) {
val key = keyGenerator(thisRow)
// Step3: 从另外一个状态表中获取关联数据进行postJoinFilter过滤后，作为结果输出
val outputIter = otherSideJoiner.joinStateManager.get(key).map { thatRow =>
generateJoinedRow(thisRow, thatRow)
}.filter(postJoinFilter)
// Step4: 将满足条件的状态数据更新至状态结果中；
val shouldAddToState = // add only if both removal predicates do not match
!stateKeyWatermarkPredicateFunc(key) && !stateValueWatermarkPredicateFunc(thisRow)
if (shouldAddToState) {
joinStateManager.append(key, thisRow)
updatedStateRowsCount += 1
}
outputIter
} else {
// 无法关联的数据，根据Join的类型生成相应的数据
joinSide match {
case LeftSide if joinType == LeftOuter =>
Iterator(generateJoinedRow(thisRow, nullRight))
case RightSide if joinType == RightOuter =>
Iterator(generateJoinedRow(thisRow, nullLeft))
case _ => Iterator()
}
}
}
}
