Apache Flink 漫谈系列(12) - Time Interval(Time-windowed) JOIN

一、说什么

JOIN 算子是数据处理赏罚的焦点算子，前面我们在《Apache Flink 漫谈系列(09) - JOIN 算子》先容了UnBounded的双流JOIN，在《Apache Flink 漫谈系列(10) - JOIN LATERAL》先容了单流与UDTF的JOIN操纵，在《Apache Flink 漫谈系列(11) - Temporal Table JOIN》又先容了单流与版本表的JOIN，本篇将先容在UnBounded数据流上定时刻维度举办数据分别举办JOIN操纵 - Time Interval(Time-windowed)JOIN, 后头我们叫做Interval JOIN。

二、现实题目

前面章节我们先容了Flink中对各类JOIN的支持，那么想想下面的查询需求之前先容的JOIN可否满意?需求描写如下：

好比有一个订单表Orders(orderId, productName, orderTime)和付款表Payment(orderId, payType, payTime)。 假设我们要统计下单一小时内付款的订单信息。

1. 传统数据库办理方法

在传统刘数据库中完成上面的需求很是简朴，查询sql如下:：

SELECT 
o.orderId, 
o.productName, 
p.payType, 
o.orderTime， 
payTime 
FROM 
Orders AS o JOIN Payment AS p ON 
o.orderId = p.orderId AND p.payTime >= orderTime AND p.payTime < orderTime + 3600 // 秒 

上面查询可以美满的完成查询需求，那么在Apache Flink内里应该怎样完成上面的需求呢?

2. Apache Flink办理方法

(1) UnBounded 双流 JOIN

上面查询需求我们很轻易想到操作《Apache Flink 漫谈系列(09) - JOIN 算子》先容了UnBounded的双流JOIN，SQL语句如下：

SELECT 
o.orderId, 
o.productName, 
p.payType, 
o.orderTime， 
payTime 
FROM 
Orders AS o JOIN Payment AS p ON 
o.orderId = p.orderId AND p.payTime >= orderTime AND p.payTime as timestamp < TIMESTAMPADD(SECOND, 3600, orderTime) 

UnBounded双流JOIN可以办理上面题目，这个示例和本篇要先容的Interval JOIN有什么相关呢?

(2) 机能题目

固然我们操作UnBounded的JOIN能办理上面的题目，可是细心说明用户需求，会发明这个需求场景订单信息和付款信息并不必要恒久存储，好比2018-12-27 14:22:22的订单只必要保持1小时，由于高出1个小时的订单假如没有被付款就是无效订单了。同样付款信息也不必要恒久保持，2018-12-27 14:22:22的订单付款信息假如是2018-12-27 15:22:22往后达到的那么我们也没有须要生涯到State中。 而对付UnBounded的双流JOIN我们会一向将数据生涯到State中，如下表示图：

这样的底层实现，对付当前需求有不须要的机能丧失。以是我们有须要开拓一种新的可以破除State的JOIN方法(Interval JOIN)来高机能的完成上面的查询需求。

(3) 成果扩展

今朝的UnBounded的双流JOIN是后头是没有步伐再举办Event-Time的Window Aggregate的。也就是下面的语句在Apache Flink上面是无法支持的：

SELECT COUNT(*) FROM ( 
SELECT 
..., 
payTime 
FROM Orders AS o JOIN Payment AS p ON 
o.orderId = p.orderId 
) GROUP BY TUMBLE(payTime, INTERVAL '15' MINUTE) 

由于在UnBounded的双流JOIN中无法担保payTime的值必然大于WaterMark(WaterMark相干可以查阅<>). Apache Flink的Interval JOIN之后可以举办Event-Time的Window Aggregate。

3. Interval JOIN

为了完成上面需求，而且办理机能和成果扩展的题目，Apache Flink在1.4开始开拓了Time-windowed Join，也就是本文所说的Interval JOIN。接下来我们具体先容Interval JOIN的语法，语义和实现道理。

三、什么是Interval JOIN

Interval JOIN 相对付UnBounded的双流JOIN来说是Bounded JOIN。就是每条流的每一条数据会与另一条流上的差异时刻地区的数据举办JOIN。对应Apache Flink官方文档的 Time-windowed JOIN(release-1.7之前都叫Time-Windowed JOIN)。

1. Interval JOIN 语法

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

TIMEBOUND_EXPRESSION 有两种写法，如下：

• L.time between LowerBound(R.time) and UpperBound(R.time)
• R.time between LowerBound(L.time) and UpperBound(L.time)
• 带偶然刻属性(L.time/R.time)的较量表达式。

2. Interval JOIN 语义

Interval JOIN 的语义就是每条数据对应一个 Interval 的数据区间，好比有一个订单表Orders(orderId, productName, orderTime)和付款表Payment(orderId, payType, payTime)。 假设我们要统计在下单一小时内付款的订单信息。SQL查询如下:

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付款数据

切合语义的预期功效是 订单id为003的信息不呈此刻功效表中，由于下单时刻2018-12-26 04:53:24.0, 付款时刻是 2018-12-26 05:53:30.0高出了1小时付款。

（编辑：湖南网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!