FlinkSql系列6之 Interval Join

前言

本次我们来学习flinksql的Interval Join,这个方式主要是针对两个流在一定时间区间内的join,支持事件时间和处理时间，而且这个流每次只会产生一条数据，是一个完全的追加流。

一、Interval Join是什么？

Interval Join 可以让⼀条流去 Join 另⼀条流中前后⼀段时间内的数据。

1.1 Inner Interval Join

只有两条流中的满足时间区间符合，并且满足其他的join条件，这时候才会匹配的上并且输出对应结果 +[L,R]

1.2 Left Interval Join

这里以左流为主，在指定的时间区间内，如果右流还没到，那么会先存入state中，当右流到了之后，并且在时间区间内那么就输出结果+ [L,R],否则如果右流在时间区间内没有到，并且时间到了，那么就会删除state，输出+[L,null].

1.3 Right Interval Join

如上，反着来

1.4 Full Interval Join

左流或者右流的数据到达之后，如果没有 Join 到另外⼀条流的数据，就会等待（左流放在左流对应的 State 中等，右流放在右流对应的 State 中等），如果之后另⼀条流数据到达之后，发现能和刚刚那条数据 Join 到，则会输出 +[L, R] 。事件时间中随着 Watermark 的推进（也⽀持处理时间），发现 State 中的数据能够过期了，就将这些数据从 State 中删除并且输出（左流过期输出
+[L, null] ，右流过期输出 -[null, R] ）

二、实际测试效果

准备步骤

CREATE TABLE show_log_table(
--曝光日志id
`log_id` BIGINT,
--曝光日志参数
`show_params` STRING,
--事件时间
`row_time` as CAST(CURRENT_TIMESTAMP AS TIMESTAMP(3)),
--watermark设置
WATERMARK FOR row_time AS row_time - INTERVAL '2' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '2','fields.show_params.length' = '1','fields.log_id.min' = '1','fields.log_id.max' = '100'
)CREATE TABLE click_log_table(
--点击日志id
`log_id` BIGINT,
--点击日志参数
`click_params` STRING,
--事件时间
`row_time` as CAST(CURRENT_TIMESTAMP AS TIMESTAMP(3)),
--watermark设置
WATERMARK FOR row_time AS row_time - INTERVAL '2' SECOND
) WITH ('connector' = 'datagen','rows-per-second' = '2','fields.click_params.length' = '1','fields.log_id.min' = '1','fields.log_id.max' = '10'
)CREATE TABLE sink_table6(
--曝光日志id
`s_id` BIGINT,
--点击日志id
`c_id` BIGINT,
--曝光日志参数
`show_params` STRING,
--点击日志参数
`click_params` STRING
) WITH (
'connector' = 'print'
)

2.1 Inner Interval Join

INSERT INTO sink_table6
SELECT 
show_log_table.log_id as s_id,
click_log_table.log_id as c_id,
show_log_table.show_params,
click_log_table.click_params
FROM show_log_table INNER JOIN click_log_table on show_log_table.log_id = click_log_table.log_id
AND show_log_table.row_time BETWEEN click_log_table.row_time - INTERVAL '3' SECOND AND click_log_table.row_time + INTERVAL '2' SECOND;

在这里插入图片描述

2.2 Left Interval Join

INSERT INTO sink_table6
SELECT 
show_log_table.log_id as s_id,
click_log_table.log_id as c_id,
show_log_table.show_params,
click_log_table.click_params
FROM show_log_table LEFT JOIN click_log_table on show_log_table.log_id = click_log_table.log_id
AND show_log_table.row_time BETWEEN click_log_table.row_time - INTERVAL '3' SECOND AND click_log_table.row_time + INTERVAL '2' SECOND;

在这里插入图片描述

2.3 Right Interval Join

INSERT INTO sink_table6
SELECT 
show_log_table.log_id as s_id,
click_log_table.log_id as c_id,
show_log_table.show_params,
click_log_table.click_params
FROM show_log_table RIGHT JOIN click_log_table on show_log_table.log_id = click_log_table.log_id
AND show_log_table.row_time BETWEEN click_log_table.row_time - INTERVAL '3' SECOND AND click_log_table.row_time + INTERVAL '2' SECOND;

在这里插入图片描述

2.4 Full Interval Join

INSERT INTO sink_table6
SELECT 
show_log_table.log_id as s_id,
click_log_table.log_id as c_id,
show_log_table.show_params,
click_log_table.click_params
FROM show_log_table FULL JOIN click_log_table on show_log_table.log_id = click_log_table.log_id
AND show_log_table.row_time BETWEEN click_log_table.row_time - INTERVAL '3' SECOND AND click_log_table.row_time + INTERVAL '2' SECOND;