本文共 1817 字，大约阅读时间需要 6 分钟。

Flink SQL 在动态表上的使用：版本表与普通表的定义与应用

一、定义

在 Flink SQL 中，动态表可以是仅追加型或更新型的表。版本表是一种特殊的更新表，它记录每个键的之前值。时态表可以是一个包含一个或多个版本快照的表，或者是一个跟踪所有变更记录的表（如数据库表的 changelog），也可以是物化所有变更后的表（如数据库表，只有最新快照）。

1.1 版本的概念

时态表可以划分为一系列带版本的快照集合。每个快照的版本表示记录的有效区间，有效时间范围可以由用户自定义。根据是否可以追溯自身的历史版本，时态表分为两种类型：

• 版本表：记录可以追溯和访问其历史版本。
• 普通表：记录仅能追溯和访问其最新版本。

二、设计初衷

2.1 关联版本表：订单与产品的关系

以订单流和产品表为例，订单表（order）从 Kafka 中实时获取订单流，产品 changelog 表（product_changelog）从数据库表 products 中获取。产品价格随时间动态变化。

2.1.1 产品 changelog 表的结构

• product_id：产品唯一标识符。
• product_name：产品名称。
• product_price：产品价格。
• update_time：事件时间。

2.1.2 产品 changelog 表的版本查询

在特定时间点查询产品 changelog 表的版本：

• 10:00:00 版本：初始价格。
• 13:00:00 版本：最新价格。

2.2 关联普通表：最新汇率的关联

某些场景需要连接变化的维表。LatestRates 是一个物化的最新汇率表（如 HBase 表）。它总是表示最新汇率数据。

2.2.1 LatestRates 表的特点

• currency：货币代码。
• rate：汇率值。
• currency_time：事件时间。

2.2.2 LatestRates 表的版本查询

• 10:15:00 查询：初始汇率。
• 11:00:00 查询：最新汇率。

三、使用

在 Flink 中，定义主键约束和事件时间属性的表即为版本表。Flink 支持通过去重查询定义版本视图。

3.1 声明版本表

以下是定义版本表的示例：

-- 定义版本表CREATE TABLE product_changelog (    product_id STRING,    product_name STRING,    product_price DECIMAL(10, 4),    update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,    PRIMARY KEY (product_id) NOT ENFORCED,    WATERMARK FOR update_time AS update_time)WITH (    'connector' = 'kafka',    'topic' = 'products',    'scan.startup.mode' = 'earliest-offset',    'properties.bootstrap.servers' = 'localhost:9092',    'value.format' = 'debezium-json');

3.2 声明版本视图

以下是定义版本视图的示例：

-- 定义版本视图CREATE VIEW versioned_rates ASSELECT currency, rate, currency_timeFROM (    SELECT *,     ROW_NUMBER() OVER (PARTITION BY currency ORDER BY currency_time DESC) AS rowNum    FROM RatesHistory) WHERE rowNum = 1;

3.2.1 版本视图的特点

• currency：去重查询的唯一标识符。
• currency_time：事件时间属性保留。

3.2.2 版本视图的输出

• 09:00:00：初始汇率。
• 10:45:00：最新汇率更新。
• 11:15:00：进一步更新。
• 11:49:00：新增汇率。

通过上述方法，可以在 Flink 中高效管理和查询动态数据表，充分发挥其时态数据处理能力。

转载地址：http://gtefk.baihongyu.com/