【翻译】Flink Table Api & SQL —Streaming 概念 —— 时态表
本文翻译自官网: Temporal Tables https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/temporal_tables.html
时态表(注:Temporal Table , 我翻译为时态表,可以访问表在不同时间的内容)表示一直在修改的表上的(参数化)视图的概念,该视图返回表在特定时间点的内容。
更改表可以是跟踪表的修改历史(例如,数据库更改日志),也可以是维表的具体修改(例如,数据库表)。
对于表的历史修改,Flink可以跟踪修改,并允许在查询中访问表的特定时间点的内容。 在Flink中,这种表由Temporal Table Function表示。
对于变化的维表,Flink允许在查询中的处理时访问表的内容。在Flink中,这种表由Temporal Table 表示。
设计初衷
与表的修改历史相关
假设我们有下表 RatesHistory
。
SELECT * FROM RatesHistory; rowtime currency rate
======= ======== ======
09:00 US Dollar 102
09:00 Euro 114
09:00 Yen 1
10:45 Euro 116
11:15 Euro 119
11:49 Pounds 108
RatesHistory
表示一个不断增长的关于日元的货币汇率的附加表(汇率为
1
)。例如,汇率期间从 09:00
到10:45
的欧元到
日元
的汇率为 114
。从 10:45
到 11:15
是 116
。
假设我们要在10:58的时间输出所有当前汇率,则需要以下SQL查询来计算结果表:
SELECT *
FROM RatesHistory AS r
WHERE r.rowtime = (
SELECT MAX(rowtime)
FROM RatesHistory AS r2
WHERE r2.currency = r.currency
AND r2.rowtime <= TIME '10:58');
子查询确定对应货币的最大时间小于或等于所需时间。外部查询列出具有最大时间戳的汇率。
下表显示了这种计算的结果。 在我们的示例中,考虑了10:45 时欧元的更新,但是 10:58 时表的版本中未考虑 11:15 时欧元的更新和新的英镑输入。
rowtime currency rate
======= ======== ======
09:00 US Dollar 102
09:00 Yen 1
10:45 Euro 116
时态表的概念旨在简化此类查询,加快其执行速度,并减少Flink的状态使用率。时态表是 append-only 表上的参数化视图,该视图将 append-only 表的行解释为表的变更日志,并在特定时间点提供该表的版本。将 append-only 表解释为变更日志需要指定主键属性和时间戳属性。主键确定覆盖哪些行,时间戳确定行有效的时间。
在上面的示例中,currency
是RatesHistory
表的主键,并且rowtime
是timestamp属性。
在Flink中,这由时态表函数表示。
与维表变化相关
另一方面,某些用例需要连接变化的维表,该表是外部数据库表。
假设 LatestRates
是一个以最新汇率实现的表(例如,存储在其中)。LatestRates
是物化的 RatesHistory 历史。那么时间 10:58 的 LatestRates 表的内容将是:
10:58> SELECT * FROM LatestRates;
currency rate
======== ======
US Dollar 102
Yen 1
Euro 116
12:00 时 LatestRates
表的内容将是:
12:00> SELECT * FROM LatestRates;
currency rate
======== ======
US Dollar 102
Yen 1
Euro 119
Pounds 108
在Flink中,这由Temporal Table表示。
时态表函数
为了访问时态表中的数据,必须传递一个时间属性,该属性确定将要返回的表的版本。Flink使用表函数的SQL语法提供一种表达它的方法。
定义后,时态表函数将使用单个时间参数timeAttribute
并返回一组行。该集合包含相对于给定时间属性的所有现有主键的行的最新版本。
假设我们Rates(timeAttribute)
基于RatesHistory
表定义了一个时态表函数,我们可以通过以下方式查询该函数:
SELECT * FROM Rates('10:15'); rowtime currency rate
======= ======== ======
09:00 US Dollar 102
09:00 Euro 114
09:00 Yen 1 SELECT * FROM Rates('11:00'); rowtime currency rate
======= ======== ======
09:00 US Dollar 102
10:45 Euro 116
09:00 Yen 1
对Rates(timeAttribute)的每个查询都将返回给定timeAttribute的Rates状态。
注意:当前 Flink 不支持使用常量时间属性参数直接查询时态表函数。目前,时态表函数只能在 join 中使用。上面的示例用于提供有关函数 Rates(timeAttribute)
返回内容的直观信息。
另请参阅有关用于持续查询的 join的页面,以获取有关如何与时态表 join 的更多信息。
定义时态表函数
以下代码段说明了如何从 append-only 表中创建时态表函数。
// Get the stream and table environments.
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = StreamTableEnvironment.create(env) // Provide a static data set of the rates history table.
val ratesHistoryData = new mutable.MutableList[(String, Long)]
ratesHistoryData.+=(("US Dollar", 102L))
ratesHistoryData.+=(("Euro", 114L))
ratesHistoryData.+=(("Yen", 1L))
ratesHistoryData.+=(("Euro", 116L))
ratesHistoryData.+=(("Euro", 119L)) // Create and register an example table using above data set.
// In the real setup, you should replace this with your own table.
val ratesHistory = env
.fromCollection(ratesHistoryData)
.toTable(tEnv, 'r_currency, 'r_rate, 'r_proctime.proctime) tEnv.registerTable("RatesHistory", ratesHistory) // Create and register TemporalTableFunction.
// Define "r_proctime" as the time attribute and "r_currency" as the primary key.
val rates = ratesHistory.createTemporalTableFunction('r_proctime, 'r_currency) // <==== (1)
tEnv.registerFunction("Rates", rates) // <==== (2)
Line (1)
创建了一个 时态表函数 rates
,使我们可以在 Table API 中使用 rates 函数 。
Line (2) 在表环境中以Rates名称注册此函数,这使我们可以在SQL中使用Rates函数。
时态表
注意:仅 Blink planner 支持此功能。
为了访问时态表中的数据,当前必须使用LookupableTableSource定义一个TableSource。 Flink 使用FOR SYSTEM_TIME AS OF 的SQL语法查询时态表,这在SQL:2011中提出。
假设我们定义了一个时态表 LatestRates
,我们可以通过以下方式查询此类表:
SELECT * FROM LatestRates FOR SYSTEM_TIME AS OF TIME '10:15'; currency rate
======== ======
US Dollar 102
Euro 114
Yen 1 SELECT * FROM LatestRates FOR SYSTEM_TIME AS OF TIME '11:00'; currency rate
======== ======
US Dollar 102
Euro 116
Yen 1
注意:当前,Flink不支持以固定时间直接查询时态表。目前,时态表只能在 join 中使用。上面的示例用于提供有关时态表LatestRates
返回内容的直觉。
另请参阅有关用于持续查询的 join的页面,以获取有关如何与时态表 join 的更多信息。
定义时态表
// Get the stream and table environments.
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = TableEnvironment.getTableEnvironment(env) // Create an HBaseTableSource as a temporal table which implements LookableTableSource
// In the real setup, you should replace this with your own table.
val rates = new HBaseTableSource(conf, "Rates")
rates.setRowKey("currency", String.class) // currency as the primary key
rates.addColumn("fam1", "rate", Double.class) // register the temporal table into environment, then we can query it in sql
tEnv.registerTableSource("Rates", rates)
另请参阅有关如何定义LookupableTableSource的页面。
欢迎关注Flink菜鸟公众号,会不定期更新Flink(开发技术)相关的推文
最新文章
- 【Hadoop】Hive HSQ 使用 &;&; 自定义HQL函数
- BZOJ3925: [Zjoi2015]地震后的幻想乡
- 【转】使用Xcode中的iOS SDK给iphone开发出第一个App程序
- StackOverflowError 和 OutOfMemoryError
- 用css做类似表格的布局
- HDU 5648 DZY Loves Math 暴力打表
- Sql中Rank排名函数
- Heritrix源码分析(十五)
- java通过jsp的Excel导出
- 开启apache服务
- 在Centos7 更改Docker默认镜像和容器的位置
- JDBC&;&;c3p0、事务、批处理、多线程 于一体的经典秘方QueryRunner
- Redis 常用数据结构及其控制命令整合
- win10 uwp 活动磁贴
- USACO比赛题泛刷
- Javascript高级编程学习笔记(25)—— 函数表达式(3)模仿块级作用域
- 上线啦,PP.io!
- shell基础:多命令顺序执行与管道符
- 第一节 Spring的环境搭建
- js判断当前浏览器是pc端还是移动端
热门文章
- 洛谷P3629 [APIO2010]巡逻(树的直径)
- 日期对象|Date构造函数|
- FFT版题 [51 Nod 1028] 大数乘法
- 牛客小白月赛12 H 华华和月月种树
- CentOS7中使用yum安装nginx和php7.2的方法
- HTML | CSS | JavaScript 常见错误
- 转载:scala中的:+\+:\::\:::
- GoCN每日新闻(2019-10-22)
- springboot2.0整合redis作为缓存以json格式存储对象
- Alibaba Nacos:搭建Nacos平台