创建及使用物化表 - 实时计算Flink版

本文为您介绍如何创建物化表，以及进行历史数据回刷、修改新鲜度、查看物化表血缘关系。

使用限制

仅实时计算引擎VVR 8.0.10及以上版本支持。
目前仅支持元数据存储类型为Filesystem的Paimon Catalog，自定义的Paimon Catalog不支持创建物化表。
需要具备开发及部署作业权限，详情请参见角色类型及权限列表。
物化表创建后不支持表结构修改，不支持引用临时对象，如temporary table/temporay function/temporary view。

创建物化表

语法结构

CREATE MATERIALIZED TABLE [catalog_name.][db_name.]table_name
-- 主键约束
[([CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED)]

[COMMENT table_comment]
-- 分区键
[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
-- With选项
[WITH (key1=val1, key2=val2, ...)]
-- 数据新鲜度
FRESHNESS = INTERVAL '<num>' { SECOND | MINUTE | HOUR | DAY }
-- 刷新模式
[REFRESH_MODE = { CONTINUOUS | FULL }]
AS  <select_statement>

参数说明

参数	是否必填	说明
FRESHNESS	是	用于指定物化表的数据新鲜度。定义了物化表数据相对于源表更新的最大延迟时间。说明当基础表为物化表时，需确保下游的数据新鲜度是上游的1~N倍（N为正整数）。数据新鲜度最大不能超过1天。
AS <select_statement>	是	该子句用于定义填充物化表数据的查询。上游表可以是物化表、表或视图。SELECT语句支持所有Flink SQL查询。
PRIMARY KEY	否	定义了一组可选的列，用于唯一标识表中的每一行。被标识字段列必须非空。
PARTITIONED BY	否	定义了一组可选的列，用于对物化表进行分区。
WITH Options	否	可以定义创建物化表所需的表属性和分区字段的时间格式参数。例如，分区字段的时间格式参数`WITH ('partition.fields.#.date-formatter' = 'yyyyMMdd')`。参数使用详情请参考下列操作示例。
REFRESH_MODE	否	用于指定物化表的刷新模式。指定的刷新模式比框架根据新鲜度自动推导的模式具有更高的优先级，以满足特定场景的需求。 CONTINUOUS：流作业会增量更新物化表数据，下游数据会立即可见，或者等Checkpoint完成时才可见。 FULL：工作流定期触发对物化表数据的全量覆盖，其数据刷新周期与数据新鲜度相匹配。默认覆盖是表级别，如果有分区字段存在，可选择是否重新刷新最新分区或更新全部分区。

操作步骤

登录实时计算管理控制台。
单击目标工作空间操作列下的控制台。
在左侧导航栏选择数据管理，单击目标Paimon Catalog。
单击目标数据库后，单击创建物化表。
假设有一张基础表orders（主键order_id，类目名称order_name，日期ds）。下面将展示基于该表创建物化表：
- 基于orders表构建物化表mt_order，查询所有字段结果为表字段，数据新鲜度为5秒。
```
CREATE MATERIALIZED TABLE mt_order
FRESHNESS = INTERVAL '5' SECOND
AS
SELECT * FROM `paimon`.`db`.`orders`
;
```
- 基于物化表mt_order创建物化表mt_id，查询order_id、ds为表字段，且设置order_id为主键，ds为分区字段，数据新鲜度为30分钟。
```
CREATE MATERIALIZED TABLE mt_id (
 PRIMARY KEY (order_id) NOT ENFORCED
)
PARTITIONED BY(ds)
FRESHNESS = INTERVAL '30' MINUTE
AS
SELECT order_id,ds FROM mt_order
;
```
- 创建物化表mt_ds，基础表为物化表mt_order，为ds分区字段列指定了date-formatter（时间格式）。每次调度时，调度时间减去新鲜度将转换为相应的ds分区值。例如，设定数据新鲜度为1小时的情况下，在2024-01-01 00:00:00的调度时间，则计算出来的ds=20231231，只有分区ds = '20231231'的数据会被刷新。如果定时调度时间为2024-01-01 01:00:00，计算出来的ds=20240101，则分区ds = '20240101'的数据会被刷新。
```
CREATE MATERIALIZED TABLE mt_ds
PARTITIONED BY(ds)
WITH (
    'partition.fields.ds.date-formatter' = 'yyyyMMdd'
)
FRESHNESS = INTERVAL '1' HOUR
AS
SELECT order_id,order_name,ds FROM mt_order
;
```
  说明
  - partition.fields.#.date-formatter中的'#'字段必须是有效的字符串类型分区字段。
  - partition.fields.#.date-formatter指定物化表的时间分区格式化，其中'#'为字符串类型的分区字段名称，可以提示系统在刷新数据时，刷新哪个分区的数据。
开始/停止更新物化表。
1. 单击对应Catalog下的materialized table库，单击目标物化表。
2. 单击右侧开始更新/停止更新。
  说明
  在停止更新时，如果最近一次更新正在进行，将在此次数据更新完成后停止。
查询物化表作业详情。
在表结构详情页签中，查看基本信息，单击数据更新作业或工作流对应的作业ID，即可查看详情。

历史数据回刷

以往在使用流作业后，如需利用前一天的全量数据来订正流处理的结果，则必须单独开发一个批作业进行处理。而使用物化表后，可以直接选择物化表的历史数据分区进行数据回刷。这一改变降低了开发和运维成本，实现了流批一体化的能力。

单击目标Catalog下的物化表。
在数据信息页签，进行数据回刷。
在创建物化表中，如果已声明分区字段，则为分区表；否则为非分区表。
分区表
查看数据分区，如果为首次回刷或没有所需要的分区，单击手动更新。如果已有分区，可以选择对应的分区回刷，单击回刷。
参数说明：
- 分区字段：该参数为表分区字段。例如填写20241201，将会回刷所有ds=20241201的数据。
- 任务名称：数据回刷任务名称。
- 更新范围（可选）：是否级联更新下游关联物化表。以该表为起点，更新链路上的所有物化表。（下游层数最大为6）。
  说明
  分区表更新，下游的物化表在分区字段上需与起始表完全一致，否则更新操作将失败。
  链路中某个物化表更新失败时，下游节点将会全部失败。
- 部署目标：支持选择queue和Session队列。默认选择default-queue。
非分区表
查看数据情况，单击回刷。
参数说明：
- 任务名称：数据回刷任务名称。
- 更新范围：非分区表不可选。
  说明
  更新时下游数据将进行全量刷新。
  链路中某个物化表更新失败时，下游节点将会全部失败。
  起始表的新鲜度被系统判定为流任务且为非分区表时，不支持级联更新。
- 部署目标：支持选择queue和Session队列。默认选择default-queue。

修改数据新鲜度

单击对应Catalog下的materialized table库，单击对应的物化表。
单击右侧修改数据新鲜度。
- 当物化表为非主键表时，不允许更改任务的流批属性。例如，若将数据新鲜度从2秒修改为1小时，此时Flink会把流作业转变为批作业；反之亦然，不可进行此类操作。（小于30分钟为流作业，大于等于30分钟为批作业）
- 当基础表为物化表时，需确保下游的数据新鲜度是上游的1~N倍（N为正整数）。
- 数据新鲜度最大不能超过1天。

查看数据血缘

数据血缘页面可以查询所有物化表之间的血缘关系，并支持在该页面直接对物化表进行开始/停止更新，修改数据新鲜度等操作，单击详情，即可跳转到对应的物化表进行查看。

实时计算Flink版：创建及使用物化表