点击上方“IT那活儿”公众号，关注后了解更多内容，不管IT什么活儿，干就完了！！！

结构化数据流支持streaming DataFrame/DataSet与静态DataFrame/DataSet进行JOIN操作，也支持两个streaming DataFrame/DataSet进行JOIN操作，流连接的结果以增量方式生成，与流聚合结果类似。

下面主要讨论流连接支持的类型，inner,outer,semi连接等。

流数据集-静态数据集JOIN操作

自Spark 2.0引入以来，支持流数据集与静态数据集进行JOIN操作，例如：

请注意流-静态连接不是有状态的，所以不需要状态管理。

流数据集-流数据集JOIN操作

在Spark 2.3中，我们添加了对流连接的支持，也就是说，您可以连接两个流数据集/数据帧。在两个数据流之间生成连接结果的挑战在于，在任何时间点，数据集的视图对于连接的两侧都是不完整的，这使得查找输入之间的匹配变得更加困难。从一个输入流接收到的任何行都可以与将来从另一个输入流接收到的任何行相匹配。

因此，对于这两个输入流，我们将过去的输入缓冲为流状态，这样我们就可以将每个未来的输入与过去的输入匹配，并相应地生成连接的结果。

此外，与流式聚合类似，我们自动处理延迟、无序的数据，并可以使用水印限制状态。让我们讨论支持的不同类型的流连接以及如何使用它们。

3.1 基于水印（watermarking）的INNER JOIN操作

支持任何类型的列上的内部连接以及任何类型的连接条件，但是，随着流的运行，流状态的大小将无限期的增长，因为所有过去的输入都必须保存，因为任何输入都有可能与过去的输入匹配，为了避免无线的状态，必须定义额外的连接条件，以便过去的旧输入无法与将来的输入匹配，因此可以删除旧状态，也就是说，你必须在连接中执行以下附加步骤：

1）在两个流输入上定义水印，以便引擎知道输入的延迟程度（类似于流聚合）。

2）定义跨两个流输入的事件时间约束，以便引擎知道何时不需要一个旧的输入行（即不满足时间约束）来匹配另一个输入，可以通过以下两种方式定义此约束：

• 时间范围连接条件（例如lefttime between righttime and right+1hour）。

• 基于事件时间窗口进行JOIN。

举例来进行说明

假如我们希望将一系列广告播放与另一系列用户点击广告的行为连接起来，要允许此连接中的状态清理，必须指定水印延迟和时间约束，如下：

• 水印延迟（watermar delays）：比如说在活动期间内，广告印象和点击事件可能分别延迟2小时/3小时。

• 事件时间范围条件：再广告播放0秒到1小时范围内可能发生点击。

代码如下：

3.2 基于水印（watermarking）的OUTER JOIN操作

虽然水印+事件时间约束条件对于inner join 不是必须的，但是对于outer join就必须指定，因为outer join会生成NULL，引擎必须知道输入行将来何时与任何输入都不匹配，因此指定水印和事件时间约束来生成正确结果。

代码如下：

关于如何生成外部连接结果，有几个重要的特征需要注意：

• 根据指定的水印延迟和事件时间范围条件，将会生成NULL 结果，这是因为引擎必须等待很长时间以确保没有匹配项，并且将来不会有更多的匹配项。

• 在当前微批处理引擎中的实现中，水印在微批处理结束时被提前，下一个微批处理使用更新后的水印来清除状态并输出外部结果。

  由于我们仅在有新数据要处理时触发微批处理，因此如果流中没有接收到新数据，则外部结果的生成可能会延迟。

  简言之，如果正在连接的两个输入流中的任何一个在一段时间内没有接收数据，则外部（两种情况下，左或右）输出可能会延迟。

3.3 基于水印（watermarking）的SEMI JOIN操作

半连接返回左侧返回值，也被称为左半连接，与外部连接类似，其也必须指定水印和事件时间约束，引擎必须知道左侧的输入行将来何时与右侧的输入行都不匹配。

流数据集、静态数据集之间各种连接总结

• 支持级联连接操作，例如df1.join(df2).join(df3).....................

• 自spark2.4版本之后，连接操作只支持Append输出模式。

• 自spark2.4版本之后，在连接之前不能使用类似于non-map-like的操作，例如在连接之前不能进行流聚合操作，在连接之前不能再Update输出模式下使用mapGroupsWithState 、flatMapGroupsWithState 操作。

END