點擊上方“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