目录
时态表（Temporal Table）是一张随时间变化的表 – 在 Flink 中称为动态表，时态表中的每条记录都关联了一个或多个时间段，所有的 Flink 表都是时态的（动态的）。
时态表包含表的一个或多个有版本的表快照，时态表可以是一张跟踪所有变更记录的表（例如数据库表的 changelog，包含多个表快照），也可以是物化所有变更之后的表（例如数据库表，只有最新表快照）。
版本: 时态表可以划分成一系列带版本的表快照集合，表快照中的版本代表了快照中所有记录的有效区间，有效区间的开始时间和结束时间可以通过用户指定，根据时态表是否可以追踪自身的历史版本与否，时态表可以分为 版本表 和 普通表。
版本表: 如果时态表中的记录可以追踪和并访问它的历史版本，这种表我们称之为版本表，来自数据库的 changelog 可以定义成版本表。
普通表: 如果时态表中的记录仅仅可以追踪并和它的最新版本，这种表我们称之为普通表，来自数据库 或 HBase 的表可以定义成普通表。
注意 仅 Blink planner 支持此功能。
如果流延迟过来的数据要跟之前的维表数据做关联，即根据流的事件时间，查找某个时间点的维度数据而不是当前维度表数据。
比如这样一个场景:用户的订单表和和商品维度表，将维度表设置成时态表，这样用户就可以根据订单表中的下单时间Join下单时的商品当时最新的维度数据
Flink引入了temporal table， Temporal Table是Append Only表（只允许插入记录，而不会修改、删除记录）上的物化视图，它把Append Only的表变化解释为表的Changelog，并提供特定时间版本的表数据、将每个时间点映射到静态关系中，表的版本数据存储在状态里。在Flink中，通过TemporalTableFunction实现对temporal table的访问，必须定义主键和时间戳属性，主键确定覆盖哪些行，时间戳确定有效的时间，也就是表的数据版本。
时态表代码开发：
数据源：ratesHistory.csv
RMB,114,2015-01-01 00:00:00 RMB,115,2015-01-03 00:00:00 RMB,116,2015-01-19 00:00:00 Euro,119,2015-01-03 00:00:00 USD,99,2015-01-03 00:00:00 USD,100,2015-01-03 00:00:00 Euro,118,2015-01-03 00:00:00
数据源：rateOrder.csv
1,29,RMB,2015-01-02 00:00:00 2,19,RMB,2015-01-03 00:00:00 3,33,RMB,2015-01-11 00:00:00 4,55,RMB,2015-01-21 00:00:00
package com.aikfk.flink.sql; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; import org.apache.flink.types.Row; /** * @author ：caizhengjie * @description：TODO * @date ：2021/4/5 10:32 下午 */ public class TemporalTableJoinEventTime { public static void main(String[] args) throws Exception { // 1.准备环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2.创建TableEnvironment(Blink planner) EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build(); StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env , settings); // 3.文件path String ratesHistoryPath = "/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/ratesHistory.csv"; String ratesOrderPath = "/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/rateOrder.csv"; // 4.DDL String ratesHistory_ddl = "create table currency_rates (\n" + " currency STRING,\n" + " conversion_rate DECIMAL(32, 2),\n" + " update_time TIMESTAMP(3),\n" + " PRIMARY KEY (currency) NOT ENFORCED,\n" + " WATERMARK FOR update_time AS update_time \n" + ") WITH (\n" + " 'connector.type' = 'filesystem',\n" + " 'connector.path' = '"+ratesHistoryPath+"',\n" + " 'format.type' = 'csv'\n" + ")"; tableEnvironment.executeSql(ratesHistory_ddl); String ratesOrder_DDL = "create table orders (\n" + " order_id STRING,\n" + " price DECIMAL(32,2),\n" + " currency STRING,\n" + " order_time TIMESTAMP(3),\n" + " WATERMARK FOR order_time AS order_time \n" + ") WITH (\n" + " 'connector.type' = 'filesystem',\n" + " 'connector.path' = '"+ratesOrderPath+"',\n" + " 'format.type' = 'csv'\n" + ")"; tableEnvironment.executeSql(ratesOrder_DDL); // 6.通过SQL对表的查询，生成结果表 String sql = "SELECT \n" + " order_id,\n" + " price,\n" + " orders.currency,\n" + " conversion_rate,\n" + " order_time\n" + " FROM orders\n" + " LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time\n" + " ON orders.currency = currency_rates.currency"; Table table = tableEnvironment.sqlQuery(sql); // 7.将table表转换为DataStream DataStream<Tuple2<Boolean, Row>> retractStream = tableEnvironment.toRetractStream(table, Row.class); retractStream.print(); env.execute(); /** * 2> (true,1,29.00,RMB,114.00,2015-01-02T00:00) * 2> (true,4,55.00,RMB,116.00,2015-01-21T00:00) * 2> (true,2,19.00,RMB,115.00,2015-01-03T00:00) * 2> (true,3,33.00,RMB,115.00,2015-01-11T00:00) */ } }
