底層原理簡(jiǎn)介

• LState:存儲左邊數據流中的數據。
• RState：存儲右邊數據流中的數據。
• 當左邊數據流數據到達的時(shí)候會(huì )保存到LState，并且到RState中進(jìn)行Join。將Join生成的結果數據發(fā)送到下游。
• 右邊數據流中數據到達的時(shí)候，會(huì )保存到RState當中，并且到LState中進(jìn)行Join，然后將Join之嚄胡的結果數據發(fā)送到下游。

為了保障左右兩邊流中需要Join的數據出現在相同節點(diǎn)，Flink SQL會(huì )利用Join中的on的關(guān)聯(lián)條件進(jìn)行分區，把相同關(guān)聯(lián)條件
的數據分發(fā)到同一個(gè)分區里面。

普通雙流Join

現有訂單表A和支付表B進(jìn)行關(guān)聯(lián)得到匯總表C。訂單表和支付表初始數據如下：

表A：訂單表數據

表B：支付表數據

inner join

當A表中每一條數據到達時(shí)，都會(huì )和B表中的數據進(jìn)行關(guān)聯(lián)：

• 當能夠關(guān)聯(lián)到數據時(shí)，將結果輸出到結果表里面；
• 當不能關(guān)聯(lián)到數據時(shí)，不會(huì )將結果輸出到結果表里面；

所以上述A表和B表的Join結果為：

當表B中1001新數據到達時(shí)，新數據如下所示:

此時(shí)結果表的數據為:

注意

Inner Join 不會(huì )產(chǎn)生回撤流。

left join

當A表數據到達時(shí)會(huì )主動(dòng)和B表中數據進(jìn)行關(guān)聯(lián)查詢(xún)，沒(méi)有關(guān)聯(lián)到數據。也會(huì )輸出結果，缺失的字段使用null進(jìn)行補全。

B表中的數據1002到達之后且A表中的數據1001和1002已經(jīng)到達，關(guān)聯(lián)之后表C的數據如下:

當B表中數據1001到達之后，也會(huì )主動(dòng)和表A中的數據進(jìn)行關(guān)聯(lián)，如果表中的數據已經(jīng)輸出過(guò)結果了且缺失字段為null，此時(shí)
會(huì )產(chǎn)生一個(gè)回撤流，將之前輸出的數據會(huì )撤掉-D，在重新輸出完整的數據+I。

注意

left Join會(huì )產(chǎn)生回撤流。

Right Join

當表B中1001到達時(shí)，A表中的數據沒(méi)有到達，則還是會(huì )輸出數據，缺失字段使用null代替。當表B中數據1002到達時(shí)，A表中的
數據1002已經(jīng)到達此時(shí)可以關(guān)聯(lián)到數據，關(guān)聯(lián)結果如下：

當表A中數據1001到達時(shí)，會(huì )主動(dòng)到B表中進(jìn)行關(guān)聯(lián)，此時(shí)結果中已經(jīng)輸出過(guò)關(guān)于1001的數據，此時(shí)會(huì )產(chǎn)生一個(gè)回撤流。

注意

Right Join會(huì )產(chǎn)生回撤流。

Full Join

當表B中數據1001先到達時(shí)，會(huì )主動(dòng)到A表中進(jìn)行關(guān)聯(lián)查詢(xún)，關(guān)聯(lián)不到數據，還是會(huì )輸出結果。

當表A中數據到達時(shí)，會(huì )主動(dòng)和B表中的數據進(jìn)行關(guān)聯(lián)查詢(xún)，此時(shí)B表中只有1001的數據，灌籃不到數據，還是會(huì )輸出結果。

所以此時(shí)關(guān)聯(lián)結果如下：

當表A中的1001到達時(shí)，會(huì )和B表進(jìn)行關(guān)聯(lián)查詢(xún)，當表B的1002到達時(shí)，會(huì )和表A進(jìn)行關(guān)聯(lián)查詢(xún)，此時(shí)結果如下：

注意

Full Join 會(huì )產(chǎn)生回撤流。

Interval Join

Interval JOIN 相對于UnBounded的雙流JOIN來(lái)說(shuō)是Bounded JOIN。就是每條流的每一條數據會(huì )與另一條流上的不同時(shí)間區域
的數據進(jìn)行JOIN。

語(yǔ)法

SELECT ... FROM t1 JOIN t2  ON t1.key = t2.key AND TIMEBOUND_EXPRESSION

TIMEBOUND_EXPRESSION 有兩種寫(xiě)法，如下：

• L.time between LowerBound(R.time) and UpperBound(R.time)
• R.time between LowerBound(L.time) and UpperBound(L.time)
• 帶有時(shí)間屬性(L.time/R.time)的比較表達式。

Interval JOIN 的語(yǔ)義就是每條數據對應一個(gè)時(shí)間區間的數據區間，比如有一個(gè)訂單表Orders(orderId, productName,
orderTime)和付款表Payment(orderId, payType, payTime)。假設我們要統計在下單一小時(shí)內付款的訂單信息。SQL查詢(xún)如下:

SELECT 
  o.orderId,
  o.productName,
  p.payType,
  o.orderTime,
  cast(payTime as timestamp) as payTime
FROM
  Orders AS o JOIN Payment AS p ON 
  o.orderId = p.orderId AND 
  p.payTime BETWEEN orderTime AND 
  orderTime + INTERVAL '1' HOUR

Orders訂單數據

Payment付款數據

符合語(yǔ)義的預期結果是 訂單id為003的信息不出現在結果表中，因為下單時(shí)間 2018-12-26 04:53:24.0, 付款時(shí)間是
2018-12-26 05:53:30.0超過(guò)了1小時(shí)付款。

那么預期的結果信息如下：

這樣Id為003的訂單是無(wú)效訂單，可以更新庫存繼續售賣(mài)。

接下來(lái)我們以圖示的方式直觀(guān)說(shuō)明Interval JOIN的語(yǔ)義，我們對上面的示例需求稍微變化一下： 訂單可以預付款(不管是
否合理，我們只是為了說(shuō)明語(yǔ)義）也就是訂單 前后 1小時(shí)的付款都是有效的。SQL語(yǔ)句如下：

SELECT
  ...
FROM
  Orders AS o JOIN Payment AS p ON
  o.orderId = p.orderId AND
  p.payTime BETWEEN orderTime - INTERVAL '1' HOUR AND
  orderTime + INTERVAL '1' HOUR

總結

• Flink的流關(guān)聯(lián)當前只能支持兩條流的關(guān)聯(lián)
• Flink同時(shí)支持基于EventTime和ProcessingTime的流流join。
• Interval join 已經(jīng)支持inner ,left outer, right outer , full outer 等類(lèi)型的join，由此來(lái)看官網(wǎng)對interval join
  類(lèi)型支持的說(shuō)明不夠準確。
• 當前版本Interval join的兩條流的消息清理是基于兩條流共有的combinedWatermark（較小的流的watermark）。
• 流的watermark不會(huì )用于將消息直接過(guò)濾掉，即時(shí)消息在本流中的watermark表示中已經(jīng)遲到，但會(huì )直接將遲到的消息根據
  相應的join類(lèi)型或輸出或丟棄。

維表Join

維表（Dimension Table）是來(lái)自數倉建模的概念。在數倉模型中，事實(shí)表（Fact Table）是指存儲有事實(shí)記錄的表，如系統
日志、銷(xiāo)售記錄等，而維表是與事實(shí)表相對應的一種表，它保存了事實(shí)表中指定屬性的相關(guān)詳細信息，可以跟事實(shí)表做關(guān)
聯(lián)；相當于將事實(shí)表上經(jīng)常重復出現的屬性抽取、規范出來(lái)用一張表進(jìn)行管理。

在實(shí)際生產(chǎn)中，我們經(jīng)常會(huì )有這樣的需求，以原始數據流作為基礎，關(guān)聯(lián)大量的外部表來(lái)補充一些屬性。這種查詢(xún)操作就是
典型的維表 JOIN。

使用維表的好處

• 縮小了事實(shí)表的大小。
• 便于維度的管理和維護，增加、刪除和修改維度的屬性，不必對事實(shí)表的大量記錄進(jìn)行改動(dòng)。
• 維度表可以為多個(gè)事實(shí)表重用，以減少重復工作。

維表JOIN使用

由于維表是一張不斷變化的表（靜態(tài)表視為動(dòng)態(tài)表的一種特例），因此在維表 JOIN 時(shí)，需指明這條記錄關(guān)聯(lián)維表快照的對
應時(shí)刻。Flink SQL 的維表 JOIN 語(yǔ)法引入了 Temporal Table 的標準語(yǔ)法，用于聲明流數據關(guān)聯(lián)的是維表哪個(gè)時(shí)刻的快照。

需要注意是，目前原生 Flink SQL 的維表 JOIN 僅支持事實(shí)表對當前時(shí)刻維表快照的關(guān)聯(lián)（處理時(shí)間語(yǔ)義），而不支持事實(shí)
表 rowtime 所對應的維表快照的關(guān)聯(lián)（事件時(shí)間語(yǔ)義）。

語(yǔ)法說(shuō)明

Flink SQL 中使用語(yǔ)法 for SYSTEM_TIME as of PROC_TIME()來(lái)標識維表JOIN。僅支持 INNER JOIN和 LEFT JOIN。

SELECT 
  column-namesFROM 
  table1 [AS <alias1>][LEFT] 
JOIN table2 FOR SYSTEM_TIME AS OF table1.proctime [AS <alias2>] 
ON table1.column-name1 = table2.key-name1

注意：
table1.proctime表示 table1的 proctime字段。

使用示例

下面用一個(gè)簡(jiǎn)單的示例來(lái)展示維表 JOIN 語(yǔ)法。假設我們有一個(gè) Orders 訂單數據流，希望根據用戶(hù) ID 補全訂單中的用戶(hù)
信息，因此需要跟 Customer 維度表進(jìn)行關(guān)聯(lián)。

CREATE TABLE Orders (  
  id   INT,  
  price   DOUBLE,  
  quantity   INT,  
  proc_time AS PROCTIME(),  
  PRIMARY KEY(id) NOT ENFORCED
) WITH ( 
  'connector' = 'datagen',  
  'fields.id.kind' = 'sequence',  
  'rows-per-second' = '10'
);
CREATE TABLE Customers (
  id   INT,
  name   STRING,
  country   STRING,
  zip   STRING,
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',
  'table-name' = 'customers'
);
CREATE TABLE OrderDetails (
  id   INT,
  total_price   DOUBLE,
  country   STRING,
  zip   STRING,
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://mysqlhost:3306/orderdb',
  'table-name' = 'orderdetails'
);
-- enrich each order with customer information
INSERT INTO OrderDetails 
SELECT 
  o.id, 
  o.price,
  o.quantity, 
  c.country, 
  c.zipFROM 
Orders AS o 
JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c  
ON o.id = c.id;

Flink SQL 執行流程

Apache Calcite 是一款開(kāi)源的 SQL 解析工具，被廣泛使用于各個(gè)大數據項目中，主要用于解析 SQL 語(yǔ)句。SQL 的執行流程
一般分為四個(gè)主要階段：

• Parse：語(yǔ)法解析，把 SQL 語(yǔ)句轉換成抽象語(yǔ)法樹(shù)（AST），在 Calcite 中用 SqlNode 來(lái)表示；
• Validate：語(yǔ)法校驗，根據元數據信息進(jìn)行驗證，例如查詢(xún)的表、使用的函數是否存在等，校驗之后仍然是 SqlNode 構
  成的語(yǔ)法樹(shù)；
• Optimize：查詢(xún)計劃優(yōu)化，包含兩個(gè)階段，1）將 SqlNode 語(yǔ)法樹(shù)轉換成關(guān)系表達式 RelNode 構成的邏輯樹(shù)，2）使用優(yōu)
  化器基于規則進(jìn)行等價(jià)變換，例如謂詞下推、列裁剪等，經(jīng)過(guò)優(yōu)化器優(yōu)化后得到最優(yōu)的查詢(xún)計劃；
• Execute：將邏輯查詢(xún)計劃翻譯成物理執行計劃，生成對應的可執行代碼，提交運行。