Spark结构化流之join

妖狐艹你老母 2023-02-16 00:48 262阅读 0赞

## 一.join操作 ##

结构化流支持将流数据集/数据帧与静态数据集/数据帧以及另一个流数据集/数据帧连接在一起。流连接的结果是增量生成的，类似于流聚合的结果。请注意，在所有受支持的联接类型中，使用流Dataset/DataFrame进行联接的结果将与使用包含流中相同数据的静态Dataset/DataFrame进行联接的结果完全相同。

## 二.流静态联接 ##

自从Spark 2.0引入以来，结构化流已支持流和静态DataFrame / Dataset之间的联接（内部联接和某种类型的外部联接）。如下：

val staticDf = spark.read. ...
    val streamingDf = spark.readStream. ...
    
    streamingDf.join(staticDf, "type")          // inner equi-join with a static DF
    streamingDf.join(staticDf, "type", "right_join")  // right outer join with a static DF

请注意，流静态联接不是有状态的，因此不需要状态管理。但是，尚不支持某些类型的流静态外部联接。

## 三.流流连接 ##

在Spark 2.3中，我们添加了对流流连接的支持，即可以连接两个流Dataset/DataFrame。在两个数据流之间生成联接结果的挑战在于，在任何时间点，联接两侧的数据集视图都不完整，这使得在输入之间查找匹配项变得更加困难。从一个输入流接收到的任何行都可以与另一输入流中将来接收到的任何行匹配。因此，对于两个输入流，我们将过去的输入作为流状态进行缓冲，以便我们可以将每个将来的输入与过去的输入进行匹配，并相应地生成合并的结果。此外，类似于流聚合，我们会自动处理较晚的乱序数据，并可以使用水印限制状态。  
**1.内部联接，带有可选水印**  
支持任何类型的列上的内部连接以及任何类型的连接条件。但是，随着流的运行，流状态的大小将无限期地增长，因为必须保存所有过去的输入，因为任何新输入都可以与过去的任何输入匹配。为了避免无界状态，必须定义其他联接条件，以使无限长的旧输入不能与将来的输入匹配，因此可以从状态中清除它们。换句话说，将必须在连接中执行以下其他步骤。

1.  在两个输入上定义水印延迟，以便引擎知道输入的延迟（类似于流聚合）。
2.  定义两个输入之间的事件时间约束，以便引擎可以确定何时不需要与另一个输入匹配的一个输入的旧行（即，不满足时间约束）。可以通过以下两种方式之一来定义此约束。
    
    1.  时间范围加入条件（JOIN ON leftTime BETWEEN rightTime AND rightTime + INTERVAL 1 HOUR）。
    2.  加入事件时间窗口（JOIN ON leftTimeWindow = rightTimeWindow）。

让我们通过一个例子来理解这一点。

假设我们想将广告印象流（在显示广告时）与用户点击广告的另一流结合起来，以在印象带来可货币化点击时进行关联。要在此流-流连接中允许状态清除，将必须指定水印延迟和时间限制，如下所示。

1.  水印延迟：在事件时间内，印象和相应的点击可能延迟/乱序最多分别为2和3个小时。
2.  事件时间范围条件：假设在相应的展示之后0秒到1小时的时间范围内可以发生点击。

代码看起来像这样。

import org.apache.spark.sql.functions.expr
    
    val impressions = spark.readStream. ...
    val clicks = spark.readStream. ...
    
    // Apply watermarks on event-time columns
    val impressionsWithWatermark = impressions.withWatermark("impressionTime", "2 hours")
    val clicksWithWatermark = clicks.withWatermark("clickTime", "3 hours")
    
    // Join with event-time constraints
    impressionsWithWatermark.join(
      clicksWithWatermark,
      expr("""
        clickAdId = impressionAdId AND
        clickTime >= impressionTime AND
        clickTime <= impressionTime + interval 1 hour
        """)
    )

**2.流流内部连接的语义保证与水印**  
这类似于在聚合上加水印所提供的保证。水印延迟为“2小时”可确保引擎永远不会丢弃任何少于2小时的数据。但是延迟超过2小时的数据可能会或可能不会得到处理。  
**3.外加水印**  
对于内部联接，水印+事件时间约束是可选的，对于左右外部联接，必须指定它们。这是因为为了在外部联接中生成NULL结果，引擎必须知道将来什么时候输入行不匹配任何内容。因此，必须指定水印+事件时间约束以生成正确的结果。因此，带有外部联接的查询看起来与前面的广告获利示例非常相似，不同之处在于，将有一个附加参数将其指定为外部联接。

impressionsWithWatermark.join(
      clicksWithWatermark,
      expr("""
        clickAdId = impressionAdId AND
        clickTime >= impressionTime AND
        clickTime <= impressionTime + interval 1 hour
        """),
      joinType = "leftOuter"      // can be "inner", "leftOuter", "rightOuter"
     )

**4.带有流水印的流外连接的语义保证**  
关于水印延迟以及数据是否会丢失，外连接与内连接具有相同的保证。  
**5.注意事项**  
关于如何生成外部结果，有一些重要的特性需要注意。

*  外部NULL结果的生成延迟取决于指定的水印延迟和时间范围条件。这是因为引擎必须等待很长时间才能确保没有匹配项，并且将来将不再有匹配项。
 *  在微批处理引擎的当前实现中，水印在微批处理的末尾进行，下一个微批处理使用更新的水印来清理状态并输出外部结果。由于我们仅在有新数据要处理时才触发微批处理，因此如果流中未接收到新数据，则外部结果的生成可能会延迟。 简而言之，如果连接的两个输入流中的任何一个在一段时间内未接收到数据，则外部（两种情况，左或右）输出可能会延迟。

**6.流查询中的联接支持矩阵**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]  
有关支持的联接的其他详细信息：

*  联接可以级联，即可以执行df1.join(df2, …).join(df3, …).join(df4, …)。
 *  从Spark 2.4开始，仅当查询处于Append输出模式时才能使用联接。尚不支持其他输出模式。
 *  从Spark 2.4开始，不能在join之前使用其他非类map操作。以下是一些无法使用的示例。
    
     *  无法使用流式聚合。
     *  无法在更新模式下使用mapGroupsWithState和flatMapGroupsWithState。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020060621245750.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200606212526470.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz,size_16,color_FFFFFF,t_70