管理Paimon Catalog - 实时计算Flink版

Paimon Catalog 实现了 Flink 对数据湖构建（DLF）数据的直接访问。本文介绍 Catalog 的创建与管理，以及数据库与表的操作流程。

注意事项

引擎版本要求
- 基础功能：创建 Paimon Catalog 和表，需使用实时计算引擎 VVR 8.0.5 及以上版本。
- DLF 集成：若将元数据存储类型设置为 DLF，需使用 VVR 11.1 及以上版本。
- OSS-HDFS 支持：使用 OSS-HDFS 作为 Paimon Filesystem Catalog 的存储时，需使用 VVR 11.8 及以上版本。

OSS 存储配置
- 存储类型：Paimon 数据文件与元数据文件存储于 OSS。必须使用标准存储类型的 Bucket。详情请参见存储类型。
- 部署建议：建议创建一个位于同一地域的独立 Bucket。避免复用系统默认 Bucket，以防止数据混淆及误操作。
权限管理
配置 Catalog 时使用的 AccessKey，必须拥有目标 OSS Bucket 或 DLF 目录的读写权限。
格式兼容性

Paimon Catalog 专用于管理 Paimon 格式数据。它不支持访问 Lance、Iceberg 等其他数据湖格式的表。
元数据同步
通过 SQL 语句创建或删除 Catalog、数据库及表后，控制台视图可能不会立即变更。请点击页面上的刷新按钮以获取最新元数据状态。

Paimon对应版本

Paimon社区版本
实时计算Flink版引擎版本（VVR ）
1.3.1
11.5、11.6、11.7、11.8
1.3
11.4
1.2
11.2、11.3
1.1
11.1
1.0
8.0.11

创建Paimon Catalog

创建Paimon Filesystem Catalog

UI方式

进入数据管理页面。
1. 登录实时计算管理控制台，单击目标工作空间操作列下的控制台。
2. 单击数据管理。
单击创建Catalog，选择Apache Paimon，单击下一步。
填写参数配置信息。

SQL方式

在数据查询文本编辑区域，输入以下命令。

CREATE CATALOG `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = '<warehouse>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

配置项参数说明如下。

通用

配置项	说明	是否必填	备注
my-catalog	Paimon Catalog名称。	是	请填写为自定义的英文名。
type	Catalog类型。	是	固定值为paimon。
metastore	元数据存储类型。	是	参数取值如下： filesystem：配置为Paimon Filesystem Catalog类型时填写。 dlf：配置为Paimon DLF Catalog类型时填写。

OSS

配置项	说明	是否必填	备注
warehouse	OSS服务中所指定的数仓目录。	是	格式为oss://<bucket>/<object>。其中： bucket：表示您创建的OSS Bucket名称。 object：表示您存放数据的路径。请在OSS管理控制台上查看您的bucket和object名称。
fs.oss.endpoint	OSS服务的连接地址。	是	如果Flink与DLF位于同一地域，则使用VPC网络Endpoint，否则使用公网Endpoint。当warehouse指定的OSS Bucket与Flink工作空间不在同一地域，或使用其它账号下的OSS Bucket时需要填写。获取方法请参见地域和Endpoint和创建AccessKey。
fs.oss.accessKeyId	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey ID。	是
fs.oss.accessKeySecret	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey secret。	是

OSS-HDFS（VVR 11.8及以上版本）

使用 OSS-HDFS 作为 Paimon Filesystem Catalog 存储时，SQL 示例如下：

CREATE CATALOG `my_catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'filesystem',
  'warehouse' = 'dls://<bucket>.<dls-endpoint>/<warehouse-path>',
  'fs.oss.endpoint' = '<dls-endpoint>',
  'fs.oss.accessKeyId' = '<access-key-id>',
  'fs.oss.accessKeySecret' = '<access-key-secret>',
  'fs.dls.client.mode.enabled' = 'false',
  'fs.dls.data.endpoint' = '<oss-data-endpoint>'
);

配置项参数说明如下。

配置项	说明	是否必填	备注
warehouse	OSS-HDFS服务中所指定的数仓目录。	是	格式为dls://<bucket>.<dls-endpoint>/<warehouse-path>。其中<dls-endpoint>同`fs.oss.endpoint`。
fs.oss.endpoint	OSS-HDFS服务的连接地址。	是	在OSS管理控制台目标 Bucket 的"概览 > 访问端口"区域，查看 HDFS 服务 Endpoint。
fs.oss.accessKeyId	拥有读写OSS-HDFS权限的阿里云账号或RAM账号的AccessKey ID。	是	—
fs.oss.accessKeySecret	AccessKey ID对应的AccessKey Secret。	是	—
fs.dls.client.mode.enabled	是否启用DLS Client Mode。	是	固定设置为`false`。
fs.dls.data.endpoint	DLS协议访问底层OSS数据时使用的数据Endpoint。	是	在OSS管理控制台目标 Bucket 的"概览 > 访问端口"区域，查看 OSS 内网 Endpoint。也可查询地域和Endpoint文档。

创建Paimon DLF Catalog

DLF

在DLF上创建Paimon Catalog，详情请参见快速使用DLF。
1. DLF Catalog需要和Flink工作空间位于同一地域，否则无法在后续流程中完成关联。

在实时计算开发控制台上，创建Paimon Catalog。

说明

这里仅用于建立与DLF Catalog的映射连接，创建或删除Catalog仅影响映射关系，不会对DLF中的实际数据产生影响。

登录实时计算控制台。
单击Flink全托管工作空间列表中的工作空间名称，进入对应的工作空间。

可以通过UI或者SQL方式来添加Catalog。

UI方式

在左侧导航栏，单击数据管理。
在Catalog列表页，单击创建Catalog。
选择Apache Paimon，单击下一步。
metastore 选择 dlf，catalog name 选择需要关联版本的DLF Catalog，单击确定，即可创建Catalog。

SQL命令

在数据查询文本编辑区域，输入以下SQL并执行，即可创建Catalog。

CREATE CATALOG `flink_catalog_name` 
WITH (
  'type' = 'paimon',
  'metastore' = 'rest',
  'token.provider' = 'dlf',
  'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com',
  'warehouse' = 'dlf_test'
);

参数说明如下：

参数	描述	是否必填	示例
type	Catalog类型，固定为paimon。	是	paimon
metastore	metastore类型，固定为rest。	是	rest
token.provider	Token提供方，固定为dlf。	是	dlf
uri	访问DLF Rest Catalog Server的URI，格式为`http://[region-id]-vpc.dlf.aliyuncs.com`。详见地域与服务接入点中的Region ID。	是	http://cn-hangzhou-vpc.dlf.aliyuncs.com
warehouse	DLF Catalog名称。	是	dlf_test

DLF-Legacy

在DLF上创建Paimon Catalog，详情请参见快速入门。
1. DLF Catalog需要和Flink工作空间位于同一地域，否则无法在后续流程中完成关联。

在实时计算开发控制台上，创建Paimon Catalog。

UI方式

进入数据管理页面。
1. 登录实时计算控制台，单击目标工作空间操作列下的控制台。
2. 单击数据管理。
单击创建Catalog，选择Apache Paimon，单击下一步。
metastore 选择 dlf，catalog name 选择需要关联的V1.0版本的DLF Catalog，然后填写参数配置（所需参数与SQL方式一致）。

SQL命令

在数据查询文本编辑区域，输入以下命令。

CREATE CATALOG `my-catalog` WITH (
  'type' = 'paimon',
  'metastore' = 'dlf',
  'warehouse' = '<warehouse>',
  'dlf.catalog.id' = '<dlf.catalog.id>',
  'dlf.catalog.accessKeyId' = '<dlf.catalog.accessKeyId>',
  'dlf.catalog.accessKeySecret' = '<dlf.catalog.accessKeySecret>',
  'dlf.catalog.endpoint' = '<dlf.catalog.endpoint>',
  'dlf.catalog.region' = '<dlf.catalog.region>',
  'fs.oss.endpoint' = '<fs.oss.endpoint>',
  'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
  'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
);

配置项参数说明如下。

通用

配置项	说明	是否必填	备注
my-catalog	Paimon Catalog名称。	是	请填写为自定义的英文名。
type	Catalog类型。	是	固定值为paimon。
metastore	元数据存储类型。	是	固定值为dlf。

OSS

配置项	说明	是否必填	备注
warehouse	OSS服务中所指定的数仓目录。	是	格式为oss://<bucket>/<object>。其中： bucket：表示您创建的OSS Bucket名称。 object：表示您存放数据的路径。请在OSS管理控制台上查看您的bucket和object名称。
fs.oss.endpoint	OSS服务的连接地址。	是	如果Flink与DLF位于同一地域，则使用VPC网络Endpoint，否则使用公网Endpoint。如果需要将Paimon表存储在OSS-HDFS中，则fs.oss.endpoint的值为`cn-<region>.oss-dls.aliyuncs.com`，例如`cn-hangzhou.oss-dls.aliyuncs.com`。
fs.oss.accessKeyId	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey ID。	是	获取方法请参见地域和Endpoint和创建AccessKey。
fs.oss.accessKeySecret	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey secret。	是	获取方法请参见地域和Endpoint和创建AccessKey。

DLF

配置项	说明	是否必填	备注
dlf.catalog.id	DLF数据目录ID。	是	请在数据湖构建控制台上查看数据目录对应的ID。
dlf.catalog.accessKeyId	访问DLF服务所需的Access Key ID。	是	获取方法请参见创建AccessKey。
dlf.catalog.accessKeySecret	访问DLF服务所需的Access Key Secret。	是	获取方法请参见创建AccessKey。
dlf.catalog.endpoint	DLF服务的Endpoint。	是	详情请参见已开通的地域和访问域名。说明如果Flink与DLF位于同一地域，则使用VPC网络Endpoint，否则使用公网Endpoint。
dlf.catalog.region	DLF所在区域。	是	详情请参见已开通的地域和访问域名。说明请和dlf.catalog.endpoint选择的地域保持一致

管理Paimon数据库

在数据查询文本编辑区域输入如下命令，选中代码后单击运行。

创建数据库

创建Paimon Catalog后，Catalog中会自动创建名为default的数据库。

--my-catalog需要替换为您创建的Paimon Catalog名称。
USE CATALOG `my-catalog`;

--my_db需要替换为自定义的database英文名称。
CREATE DATABASE `my_db`;

删除数据库

重要

无法删除DLF类型Catalog下的default数据库，允许删除Filesystem类型Catalog下的default数据库。

--my-catalog需要替换为您创建的Paimon Catalog名称。
USE CATALOG `my-catalog`;

--my_db需要替换为您想要删除的数据库名称。
DROP DATABASE `my_db`; --只需要删除内部没有任何表的数据库。
DROP DATABASE `my_db` CASCADE; --删除数据库且同时删除数据库下的所有表。

管理Paimon表

创建表

说明

Paimon Catalog配置完成后，您就可以在作业中引用Paimon Catalog表信息，作为源表、结果表和维表时，无需声明表的DDL。在SQL命令方式中，您可以直接使用Paimon Catalog表名称的完整格式${Paimon-catalog-name}.${Paimon-db-name}.${Paimon-table-name}，也可以用use catalog ${Paimon-catalog-name}和use ${Paimon-db-name}语句先声明Catalog名称和数据库名称，后面的SQL里只写表名称${Paimon-table-name}即可。

通过CREATE TABLE语句创建表

在数据查询文本编辑区域输入如下命令后，选中代码后单击运行。

在名为my-catalog的Catalog下的my_db的数据库中，创建一张分区表，其中分区键为dt，主键为dt、shop_id和user_id，分桶数固定为4。代码示例如下。
```
--my-catalog需要替换为您创建的Paimon Catalog名称。
--my_db需要替换为您想要使用的database名称。
--my_tbl也可以改为自定义的英文名称。
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket' = '4'
);
```
关于Paimon表的更多参数与使用方式，详情请参见流式数据湖仓Paimon和Paimon主键表和Append Only表。
通过CREATE TABLE AS（CTAS）语句（退役中）或CREATE DATABASE AS（CDAS）语句创建表

CTAS与CDAS语句不仅会自动同步数据，还会同步表结构的变更。通过CTAS与CDAS语句，您可以方便地将MySQL、Kafka等数据源中的表同步至Paimon Catalog中。

由于CTAS与CDAS语句涉及数据的同步，您需要部署并启动作业，具体操作请参见作业开发地图和作业启动。
说明
- 通过CTAS与CDAS语句创建的Paimon表时，暂不支持设置'bucket' = '-1'以使用动态分桶的Paimon主键表或Paimon Append Only表（非主键表）。
- CTAS与CDAS支持在WITH参数中设置物理表属性，在创建目标表时，同时在表上设置对应的属性。当作业启动时，这些参数将被应用到需要同步的下游表中。具体支持的表属性参数详情，请参见流式数据湖仓Paimon。
- 创建指定表并同步表中的数据
  
  例如，根据mysql.tpcds.web_sales表的结构，自动创建`my-catalog`.`my_db`.`web_sales`Paimon表，并同步表中的数据。另外指定Paimon表的分桶数为4，以及使用input变更数据产生机制。
```
CREATE TABLE IF NOT EXISTS `<catalog name>`.`<db name>`.`<table name>`
WITH (
  'bucket' = '4',
  'changelog-producer' = 'input'
) AS TABLE mysql.tpcds.web_sales;
```
- 创建整个数据库的表
  
  例如，根据mysql.tpcds数据库中每张表的结构，自动在`my-catalog`.`my_db`中创建对应的Paimon表，并同步表中的数据。另外指定Paimon表使用input变更数据产生机制。
```
CREATE DATABASE IF NOT EXISTS `<catalog name>`.`<db name>`
WITH (
  'changelog-producer' = 'input'
) AS DATABASE mysql.tpcds INCLUDING ALL TABLES;
```
- 同步列类型变更
  
  通过CTAS/CDAS语句创建的Paimon表不仅支持列的增加，还支持特定的列类型变更。您可以根据需要，选择是否使用字段类型宽容模式。
  - 默认情况下
    
    通过CTAS/CDAS语句创建的Paimon表的列类型与源表的列类型一致，并支持以下列类型变更：
    - 支持整数类型TINYINT、SMALLINT、INT、BIGINT变更为精度相同或更高的整数类型，其中TINYINT精度最低，BIGINT精度最高。
    - 支持浮点类型FLOAT、DOUBLE变更为精度相同或更高的浮点类型，其中FLOAT精度最低，DOUBLE精度最高。
    - 支持字符串类型CHAR、VARCHAR和STRING变更为精度相同或更高的字符串类型。
  - 使用字段类型宽容模式
    
    在CTAS与CDAS场景中，Paimon Catalog支持在WITH参数中设置'enableTypeNormalization' = 'true'后，将使用字段类型宽容模式。在上游发生数据类型修改事件时，只要所修改的类型与原类型的归一化类型相同，都视作修改成功。目前类型归一化规则如下：
    - TINYINT、SMALLINT、INT和BIGINT归一化为BIGINT。
    - FLOAT和DOUBLE归一化为DOUBLE。
    - CHAR、VARCHAR和STRING归一化为STRING。
    - 其他数据类型不进行归一化。
    例如：
    - SMALLINT修改为INT，两者的归一化类型都是BIGINT，视为修改成功，作业正常运行。
    - 从FLOAT改为BIGINT，两者的归一化类型分别为DOUBLE和BIGINT，属于不兼容的情况，会抛出异常。
    Paimon表中存储的数据类型将统一为归一化后的类型。例如，MySQL中类型分别为SMALLINT和INT的两列，在Paimon表中均存储为BIGINT类型。

修改表结构

在数据查询文本编辑区域输入如下命令，选中代码后单击运行。

操作	代码示例
添加或修改表参数	将表的`write-buffer-size`参数值修改为256 MB，并将`write-buffer-spillable`参数值修改为true。 `ALTER TABLE my_table SET ( 'write-buffer-size' = '256 MB', 'write-buffer-spillable' = 'true' );`
临时修改表参数	您可以通过在表名后添加SQL Hint的方式，在写入表时临时修改表参数。临时修改的表参数仅对当前SQL作业生效。在写入my_table表时，临时将`write-buffer-size`设为256 MB，将`write-buffer-spillable`设为true。 `INSERT INTO my_table /+ OPTIONS('write-buffer-size' = '256 MB', 'write-buffer-spillable' = 'true') / SELECT ...;` 在消费my_table表时，临时将`scan.mode`设为latest，将`scan.parallelism`设为10。 `SELECT * FROM t /+ OPTIONS('scan.mode' = 'latest', 'scan.parallelism' = '10') /;`
重命名表	将表的名称从my_table修改为my_table_new。 `ALTER TABLE my_table RENAME TO my_table_new;` 重要由于对象存储的重命名操作不具有原子性，因此如果您使用对象存储OSS保存Paimon表文件，请谨慎进行重命名表操作。推荐您使用OSS-HDFS服务以保证文件操作的原子性。
添加新列	将类型为INT的c1列和类型为STRING的c2列添加在my_table表的末尾。 `ALTER TABLE my_table ADD (c1 INT, c2 STRING);` 将类型为STRING的c2列添加在my_table表中c1的列后面。 `ALTER TABLE my_table ADD c2 STRING AFTER c1;` 将类型为INT的c1列添加在my_table表的开头。 `ALTER TABLE my_table ADD c1 INT FIRST;`
重命名列	将my_table表中名称为c0的列重命名为c1。 `ALTER TABLE my_table RENAME c0 TO c1;`
删除列	将my_table表中c1和c2两列删除。 `ALTER TABLE my_table DROP (c1, c2);`
删除分区	将my_table表中`dt=20240108,hh=06`与`dt=20240109,hh=07`两个分区删除。 ALTER TABLE my_table DROP PARTITION (`dt` = '20240108', `hh` = '08'), PARTITION (`dt` = '20240109', `hh` = '07');
修改列注释	将my_table表中buy_count列的注释改为this is buy count。 `ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';`
修改列顺序	将my_table表中名称为col_a，类型为DOUBLE的列移动到开头。 `ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;` 将my_table表中名称为col_a，类型为DOUBLE的列移动到col_b列的后面。 `ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;`
修改列类型	将my_table表中的col_a列类型改为DOUBLE。 `ALTER TABLE my_table MODIFY col_a DOUBLE;` Paimon表支持的列类型修改情况请参见下表。下图中的〇表示支持类型转换，空单元格代表不支持类型转换。

删除表

在数据查询文本编辑区域输入如下命令，选中代码后单击运行。

--my-catalog需要替换为您创建的Paimon Catalog名称。
--my_db需要替换为您想要使用的database名称。
--my_tbl需要替换为您创建的Paimon Catalog表名称。
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;

出现The following statement has been executed successfully!的提示信息说明Paimon表删除成功。

查看或删除Paimon Catalog

在实时计算管理控制台，单击目标工作空间操作列的控制台。
在数据管理页面，查看或删除Paimon Catalog。
- 查看：在Catalog列表页面，查看Catalog名称和类型。如果您需要查看Catalog下的数据库和表，请单击查看。
- 删除：在Catalog列表页面，单击目标Catalog名称对应操作列的删除。
  
  说明
  删除Paimon Catalog只会删除Flink项目空间数据管理中的记录，不会影响Paimon表的数据文件。删除Catalog后，只需要重新执行创建Paimon Catalog，即可重新使用Catalog中的Paimon表。
  
  您也可以在数据查询文本编辑区域输入DROP CATALOG <catalog name>;，选中代码后单击运行。

实时计算Flink版：管理Paimon Catalog

注意事项

创建Paimon Catalog

创建Paimon Filesystem Catalog

UI方式

SQL方式

创建Paimon DLF Catalog

DLF

UI方式

SQL命令

DLF-Legacy

UI方式

SQL命令

管理Paimon数据库

管理Paimon表

创建表

修改表结构

删除表

查看或删除Paimon Catalog

相关文档

Paimon社区版本	实时计算Flink版引擎版本（VVR ）
1.3.1	11.5、11.6、11.7、11.8
1.3	11.4
1.2	11.2、11.3
1.1	11.1
1.0	8.0.11