在flink雙流Time-windowed Joins的主要實現(xiàn)是在TimeBoundedStreamJoin中，這個類里面的變量非常的多，所以首先要清楚，這些重要變量或者概念的計算過程。簡單的說整個join過程就是把左流的數(shù)據(jù)和右流的數(shù)據(jù)都通過state保存起來，左流有新的數(shù)據(jù)到，就會根據(jù)key去遍歷右流state中的數(shù)據(jù)，符合關(guān)聯(lián)條件就輸出，關(guān)聯(lián)不上的就保存在左流的state中等待右流數(shù)據(jù)的遍歷，反之亦然。另外會對每個流計算過期時間，以及每個數(shù)據(jù)的清理時間。本文主要根據(jù)代碼的實現(xiàn)過程對清理機(jī)制做一個分步的演算。

基本公式

建表語句

CREATE TABLE LeftTable (
  l_id STRING, 
  l_imsi STRING, 
  l_time TIMESTAMP(3), 
  WATERMARK FOR l_time AS l_time - INTERVAL '5' SECOND 
) WITH ( 
  'connector.type' = 'kafka',   
  'connector.version' = 'universal',  
  'connector.topic' = 'foo',  
  'connector.properties.zookeeper.connect' = 'hostA:2181',
  'connector.properties.bootstrap.servers' = 'hostA:6667',  
  'connector.properties.group.id' = 'tg1',   
  'connector.startup-mode' = 'latest-offset', 
  'format.type' = 'csv',   
  'format.field-delimiter' = ','
  )

CREATE TABLE RightTable (
  r_id STRING, 
  r_location STRING, 
  r_time TIMESTAMP(3), 
  WATERMARK FOR r_time AS r_time - INTERVAL '2' SECOND 
) WITH (  
  'connector.type' = 'kafka',   
  'connector.version' = 'universal',  
  'connector.topic' = 'bar',  
  'connector.properties.zookeeper.connect' = 'hostA:2181',
  'connector.properties.bootstrap.servers' = 'hostA:6667',
  'connector.properties.group.id' = 'tg1',   
  'connector.startup-mode' = 'latest-offset', 
  'format.type' = 'csv',   
  'format.field-delimiter' = ',' 
)

執(zhí)行SQL

SELECT l_id, l_imsi, r_location 
FROM LeftTable 
LEFT JOIN RightTable 
on l_id = r_id 
and r_time >= l_time - INTERVAL '4' SECOND AND r_time <= l_time + INTERVAL '6' SECOND

即 upperBound = 6和lowerNound = -4

那么

leftRelativeSize = -leftLowerBound = -(-upperBound) = upperBound = 6

rightRelativeSize = leftUpperBound = -lowerBound = 4

leftExpirationTime = wm - upperBound - 0.001 
                   = wm - 6 - 0.001 
                   = wm - 6.001

rightExpirationTime = wm + lowerBound - 0.001 
                    = wm - 4 - 0.001 
                    = wm - 4.001
  
leftRowCleanUpTime = rowTime + leftRelativeSize + minCleanUpInterval + allowedLateness + 0.001 
                   = rowTime + 6 + (6+4)/2 + 0.001 
                   = rowTime + 11.001
                   
rightRowCleanUpTime = rowTime + rightRelativeSize + minCleanUpInterval + allowedLateness + 0.001 
                    = rowTime + 4 + (6+4)/2 + 0.001 
                    = rowTime + 9.001

rightOperatorTime = leftOperatorTime = wm = min(leftWatermark, rightWatermark)
  
allowedLateness = 0 //忽略不計

根據(jù)上面的公式進(jìn)行模擬計算

總結(jié)

每次計算ExpirationTime的時候用的是上一次的wm；
6-1，6-2，6-3為三種獨立可能觸發(fā)刪除過期數(shù)據(jù)的場景；
通過上面的測試可以發(fā)現(xiàn)，數(shù)據(jù)即便過期了，但是沒有到清理時間，如果這時候有符合關(guān)聯(lián)條件的數(shù)據(jù)還是可以關(guān)聯(lián)上的，例如6-2場景。

以上數(shù)據(jù)基于flink 1.10.0版本blink planner進(jìn)行測試。