Flink SQL之时态表（Temporal Tables）

ゝ一纸荒年。 2022-11-17 11:22 271阅读 0赞

### 目录 ###

时态表（Temporal Table）是一张随时间变化的表 – 在 Flink 中称为动态表，时态表中的每条记录都关联了一个或多个时间段，所有的 Flink 表都是时态的（动态的）。

时态表包含表的一个或多个有版本的表快照，时态表可以是一张跟踪所有变更记录的表（例如数据库表的 changelog，包含多个表快照），也可以是物化所有变更之后的表（例如数据库表，只有最新表快照）。

**版本**: 时态表可以划分成一系列带版本的表快照集合，表快照中的版本代表了快照中所有记录的有效区间，有效区间的开始时间和结束时间可以通过用户指定，根据时态表是否可以追踪自身的历史版本与否，时态表可以分为 版本表 和 普通表。

**版本表**: 如果时态表中的记录可以追踪和并访问它的历史版本，这种表我们称之为版本表，来自数据库的 changelog 可以定义成版本表。

**普通表**: 如果时态表中的记录仅仅可以追踪并和它的最新版本，这种表我们称之为普通表，来自数据库 或 HBase 的表可以定义成普通表。

注意 仅 Blink planner 支持此功能。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 3]  
如果流延迟过来的数据要跟之前的维表数据做关联，即根据流的事件时间，查找某个时间点的维度数据而不是当前维度表数据。

**比如这样一个场景:用户的订单表和和商品维度表，将维度表设置成时态表，这样用户就可以根据订单表中的下单时间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) */
    
        }
    }

关于时态表的官方文档：

*  [https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/streaming/versioned\_tables.html][https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_streaming_versioned_tables.html]

关于流上join的官方文档：

*  [https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/streaming/joins.html][https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_streaming_joins.html]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]: /images/20221022/3957071a882745d9b219dc961d756b33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]: /images/20221022/ad69bbdf85754e11b945876afed618a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]: /images/20221022/b4067574a8de423795d52b9f0bf9c0b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 3]: /images/20221022/3f4009e49bcf4e79a446d958675cc976.png
[https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_streaming_versioned_tables.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/streaming/versioned_tables.html
[https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_table_streaming_joins.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/streaming/joins.html