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

基本公式

建表語(yǔ)句

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ù)上面的公式進(jìn)行模擬計(jì)算

總結(jié)

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

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