创建Paimon表并进行数据读写与更新-数据湖 OpenLake-阿里云

本文通过简单的示例，在实时计算开发控制台带您快速体验Paimon的基本功能，包括创建和删除Paimon Catalog，创建和删除Paimon Catalog表，向Paimon表写入、更新以及消费数据。

前提条件

如果您使用RAM用户或RAM角色等身份访问，需要确认已具有实时计算开发控制台相关权限，详情请参见权限管理。
已创建Flink工作空间，详情请参见开通实时计算Flink版。

已开通对象存储OSS且Bucket的存储类型为标准存储，详情请参见OSS控制台快速入门。OSS用于存储Paimon表的相关文件，包括数据文件与元数据文件等。
仅实时计算引擎VVR 8.0.5及以上版本支持Paimon表。

创建Paimon DLF Catalog

在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

步骤二：创建Paimon Catalog表

在查询脚本页面，输入如下代码创建名为my_db的Paimon数据库以及名为my_tbl的Paimon表。
```
CREATE DATABASE `my-catalog`.`my_db`;
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  dt STRING,
  id BIGINT,
  content STRING,
  PRIMARY KEY (dt, id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'changelog-producer' = 'lookup'  
);
```
说明
为了后续能够流式消费Paimon表，本示例在WITH中指定了changelog-producer参数为lookup，表示采用查找策略生成变更日志，详情请参见变更数据产生机制。
选中创建Paimon数据库以及Paimon表的代码，单击左侧的运行。
返回The following statement has been executed successfully!信息表示名为my_db的Paimon数据库以及名为my_tbl的Paimon表创建成功。

步骤三：向Paimon表写入数据

在数据开发 > ETL页面的作业草稿页签，新建空白流作业草稿，详情请参见作业开发地图。将如下插入语句复制到SQL编辑器。

-- Paimon结果表在每次检查点完成之后才会正式提交数据。
-- 此处将检查点间隔缩短为10s，是为了更快地提交数据。
-- 在生产环境下，系统检查点的间隔与两次系统检查点之间的最短时间间隔根据业务对延时要求的不同，一般设置为1分钟到10分钟。
SET 'execution.checkpointing.interval'='10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

在SQL编辑区域右上方，单击部署，在部署新版本对话框中，根据需要填写或选中相关内容，单击确定。
在运维中心 > 作业运维页面，单击目标作业名称操作列下的启动，选择无状态启动，单击启动。
当作业状态变为已完成，表示已完成数据的写入。

步骤四：从Paimon表中流式消费数据

新建空白流作业草稿，将如下SQL代码复制到SQL编辑器，使用print连接器，输出my_tbl表中所有数据到日志中。

CREATE TEMPORARY TABLE Print (
  dt STRING,
  id BIGINT,
  content STRING
) WITH (
  'connector' = 'print'
);
INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;

在SQL编辑区域右上方，单击部署，在部署新版本对话框中，根据需要填写或选中相关内容，单击确定。
在运维中心 > 作业运维页面，单击目标作业操作列下的启动，选择无状态启动，单击启动。
在作业运维详情页面，查看Flink计算结果。
1. 在运维中心 > 作业运维页面，单击目标作业名称。
2. 在作业日志页签下的运行日志页面，单击运行Task Managers页签下的Path, ID。
3. 单击Stdout后，查看消费的Paimon数据。

步骤五：向Paimon表更新数据

新建空白流作业草稿，将如下SQL代码复制到SQL编辑器。

SET 'execution.checkpointing.interval' = '10s';
INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

在SQL编辑区域右上方，单击部署，在部署新版本对话框中，根据需要填写或选中相关内容，单击确定。
在运维中心 > 作业运维页面，单击目标作业操作列下的启动，选择无状态启动，单击启动。
当作业状态变为已完成，表示已完成数据的写入。
在步骤四作业的作业运维页面的Stdout页签下，查看向Paimon表中更新的数据。

步骤六（可选）：停止流式消费作业并清理资源

测试完毕后，如果您需要停止流式消费作业并清理资源，可参考以下步骤：

在运维中心 > 作业运维页面，单击目标作业操作列的停止，停止作业运行。

在查询脚本页签的文本编辑区域，输入如下代码，删除Paimon数据文件以及Paimon Catalog。

DROP DATABASE `my-catalog`.`my_db` CASCADE; --将删除存储在OSS上的整个Paimon数据库的数据文件。
DROP CATALOG `my-catalog`; --将Paimon Catalog从实时计算开发控制台元数据中删除，但对OSS上的数据文件没有影响。

返回The following statement has been executed successfully!信息表示Paimon数据文件以及Paimon Catalog删除成功。

Paimon快速开始：基本功能