Apache Flink 漫谈系列(11) - Temporal Table JOIN

Temporal Table的概念旨在简化此类查询，加速它们的执行。Temporal Table是Append Only表上的参数化视图，它把Append Only的表变化解释为表的Changelog，并在特定时间点提供该表的版本(时间版本)。将Applend Only表解释为changelog需要指定主键属性和时间戳属性。主键确定覆盖哪些行，，时间戳确定行有效的时间，也就是数据版本，与上面SQL Server示例的有效期的概念一致。

在上面的示例中，currency是RatesHistory表的主键，而rowtime是timestamp属性。

2. 如何定义Temporal Table

在Apache Flink中扩展了TableFunction的接口，在TableFunction接口的基础上添加了时间属性和pk属性。

(1) 内部TemporalTableFunction定义如下：

class TemporalTableFunction private( 
@transient private val underlyingHistoryTable: Table, 
// 时间属性，相当于版本信息 
private val timeAttribute: Expression, 
// 主键定义 
private val primaryKey: String, 
private val resultType: RowTypeInfo) 
extends TableFunction[Row] { 
...} 

(2) 用户创建TemporalTableFunction方式

在Table中添加了createTemporalTableFunction方法，该方法需要传入时间属性和主键，接口定义如下：

// Creates TemporalTableFunction backed up by this table as a history table. 
 
def createTemporalTableFunction( 
timeAttribute: Expression, 
primaryKey: Expression): TemporalTableFunction = { 
...} 

用户通过如下方式调用就可以得到一个TemporalTableFunction的实例，代码如下：

val tab = ... 
val temporalTableFunction = tab.createTemporalTableFunction('time, 'pk) 
... 

3. 案例代码

(1) 需求描述

假设我们有一张订单表Orders和一张汇率表Rates，那么订单来自于不同的地区，所以支付的币种各不一样，那么假设需要统计每个订单在下单时候Yen币种对应的金额。

（编辑：惠州站长网）

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