Flink Time-windowed Joins过期数据清理机
2020-05-12 本文已影响0人
Woople
在flink双流Time-windowed Joins的主要实现是在TimeBoundedStreamJoin中，这个类里面的变量非常的多，所以首先要清楚，这些重要变量或者概念的计算过程。简单的说整个join过程就是把左流的数据和右流的数据都通过state保存起来，左流有新的数据到，就会根据key去遍历右流state中的数据，符合关联条件就输出，关联不上的就保存在左流的state中等待右流数据的遍历，反之亦然。另外会对每个流计算过期时间，以及每个数据的清理时间。本文主要根据代码的实现过程对清理机制做一个分步的演算。
基本公式
建表语句
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' = ','
)
执行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 //忽略不计
根据上面的公式进行模拟计算
| 顺序 | 来源 | 数据 | wm | ExpirationTime | RowCleanUpTime | 结果 |
|---|---|---|---|---|---|---|
| 1 | Left | 1,111,2020-01-01 10:10:16 | 0 | left=0, right=-4001 | 10:10:27.001 | |
| 2 | Right | 2,B,2020-01-01 10:10:20 | 10:10:11 | left=-6001, right=-4001 | 10:10:29.001 | |
| 3 | Left | 2,222,2020-01-01 10:10:22 | 10:10:17 | left=-6001, right=10:10:06.999 | 10:10:33.001 | join输出2,222,B |
| 4 | Left | 4,4444,2020-01-01 10:10:35 | 10:10:18 | left=-6001, right=10:10:12.999 | 10:10:46.001 | |
| 5 | Right | 4,D,2020-01-01 10:10:29 | 10:10:27 | left=10:10:11.999, right=10:10:12.999 | 10:10:38.001 | |
| 6-1 | Right | 5,E,2020-01-01 10:10:30 | 10:10:28 | left=10:10:20.999, right=10:10:12.999 | 10:10:39.001 | |
| wm超过第一条数据的CleanUpTime，触发定时器 | 10:10:28 | left=10:10:21.999, right=10:10:12.999 | 删除第一条数据，因为是left join所以输出1,111, | |||
| 6-2 | Right | 1,A,2020-01-01 10:10:17 | 10:10:27 | left=10:10:20.999, right=10:10:12.999 | 可以和第一条数据join上，所以输出1,111,A但是第一条数据的时间戳已经小于leftExpirationTime说明已经过期，同时在缓存中删除第一条数据，但是这时wm并没有超过第一条数据的CleanUpTime，不会触发清理的定时器 | |
| 6-3 | Right | 1,A,2020-01-01 10:10:30 | 10:10:28 | left=10:10:20.999, right=10:10:12.999 | 不能和第一条数据join上，所以输出1,111,但是第一条数据的时间戳已经小于leftExpirationTime说明已经过期，同时在缓存中删除第一条数据 | |
| wm超过第一条数据的CleanUpTime，触发定时器 | 10:10:28 | left=10:10:21.999, right=10:10:12.999 | 第一条数据已经删除，定时器不需要做其他操作 |
总结
每次计算ExpirationTime的时候用的是上一次的wm；
6-1，6-2，6-3为三种独立可能触发删除过期数据的场景；
通过上面的测试可以发现，数据即便过期了，但是没有到清理时间，如果这时候有符合关联条件的数据还是可以关联上的，例如6-2场景。
以上数据基于flink 1.10.0版本blink planner进行测试。
