Apache Paimon是一种流批统一的湖存储格式，支持高吞吐的写入和低延迟的查询。本文通过Paimon Catalog和MySQL连接器，将云数据库RDS中的订单数据和表结构变更导入Paimon表中，并使用Flink对Paimon表进行简单分析。

背景信息

Apache Paimon是一种流批统一的湖存储格式，支持高吞吐的写入和低延迟的查询。目前阿里云实时计算Flink版，以及开源大数据平台E-MapReduce上常见的计算引擎（例如Spark、Hive或Trino）都与Paimon有着较为完善的集成度。您可以借助Apache Paimon快速地在HDFS或者OSS上构建自己的数据湖存储服务，并接入计算引擎实现数据湖的分析。

前提条件

• 如果您使用RAM用户或RAM角色等身份访问，需要确认已具有Flink控制台相关权限，详情请参见权限管理。

• 已创建Flink工作空间，详情请参见开通实时计算Flink版。

步骤一：准备数据源

1. 快速创建RDS MySQL实例。

   说明

   RDS MySQL版实例需要与Flink工作空间处于同一VPC。不在同一VPC下时请参见网络连通性。

2. 创建数据库和账号。

   创建名称为orders的数据库，并创建高权限账号或具有数据库orders读写权限的普通账号。

3. 通过DMS登录RDS MySQL，在orders数据库中创建表orders_1和orders_2。

   CREATE TABLE `orders_1` (
       orderkey BIGINT NOT NULL,
       custkey BIGINT,
       order_status VARCHAR(100),
       total_price DOUBLE,
       order_date DATE,
       order_priority VARCHAR(100),
       clerk VARCHAR(100),
       ship_priority INT,
       comment VARCHAR(100),
       PRIMARY KEY (orderkey)
   );
   
   CREATE TABLE `orders_2` (
       orderkey BIGINT NOT NULL,
       custkey BIGINT,
       order_status VARCHAR(100),
       total_price DOUBLE,
       order_date DATE,
       order_priority VARCHAR(100),
       clerk VARCHAR(100),
       ship_priority INT,
       comment VARCHAR(100),
       PRIMARY KEY (orderkey)
   );

4. 插入如下测试数据。

   INSERT INTO `orders_1` VALUES (1, 1, 'O', 131251.81, '1996-01-02', '5-LOW', 'Clerk#000000951', 0, 'nstructions sleep furiously among ');
   INSERT INTO `orders_1` VALUES (2, 3, 'O', 40183.29, '1996-12-01', '1-URGENT', 'Clerk#000000880', 0, ' foxes. pending accounts at the pending, silent asymptot');
   INSERT INTO `orders_1` VALUES (3, 6, 'F', 160882.76, '1993-10-14', '5-LOW', 'Clerk#000000955', 0, 'sly final accounts boost. carefully regular ideas cajole carefully. depos');
   INSERT INTO `orders_1` VALUES (4, 6, 'O', 31084.79, '1995-10-11', '5-LOW', 'Clerk#000000124', 0, 'sits. slyly regular warthogs cajole. regular, regular theodolites acro');
   INSERT INTO `orders_1` VALUES (5, 2, 'F', 86615.25, '1994-07-30', '5-LOW', 'Clerk#000000925', 0, 'quickly. bold deposits sleep slyly. packages use slyly');
   INSERT INTO `orders_1` VALUES (6, 2, 'F', 36468.55, '1992-02-21', '4-NOT SPECIFIED', 'Clerk#000000058', 0, 'ggle. special, final requests are against the furiously specia');
   INSERT INTO `orders_1` VALUES (7, 2, 'O', 171488.73, '1996-01-10', '2-HIGH', 'Clerk#000000470', 0, 'ly special requests ');
   INSERT INTO `orders_1` VALUES (8, 6, 'O', 116923.00, '1995-07-16', '2-HIGH', 'Clerk#000000616', 0, 'ise blithely bold, regular requests. quickly unusual dep');
   INSERT INTO `orders_1` VALUES (9, 3, 'F', 99798.76, '1993-10-27', '3-MEDIUM', 'Clerk#000000409', 0, 'uriously. furiously final request');
   INSERT INTO `orders_1` VALUES (10, 3, 'O', 41670.02, '1998-07-21', '3-MEDIUM', 'Clerk#000000223', 0, 'ly final packages. fluffily final deposits wake blithely ideas. spe');
   INSERT INTO `orders_2` VALUES (11, 6, 'O', 148789.52, '1995-10-23', '4-NOT SPECIFIED', 'Clerk#000000259', 0, 'zzle. carefully enticing deposits nag furio');
   INSERT INTO `orders_2` VALUES (12, 5, 'O', 38988.98, '1995-11-03', '1-URGENT', 'Clerk#000000358', 0, ' quick packages are blithely. slyly silent accounts wake qu');
   INSERT INTO `orders_2` VALUES (13, 4, 'F', 113701.89, '1992-06-03', '3-MEDIUM', 'Clerk#000000456', 0, 'kly regular pinto beans. carefully unusual waters cajole never');
   INSERT INTO `orders_2` VALUES (14, 6, 'O', 46366.56, '1996-08-21', '4-NOT SPECIFIED', 'Clerk#000000604', 0, 'haggle blithely. furiously express ideas haggle blithely furiously regular re');
   INSERT INTO `orders_2` VALUES (15, 4, 'O', 219707.84, '1996-09-20', '3-MEDIUM', 'Clerk#000000659', 0, 'ole express, ironic requests: ir');
   INSERT INTO `orders_2` VALUES (16, 1, 'F', 20065.73, '1994-07-16', '3-MEDIUM', 'Clerk#000000661', 0, 'wake fluffily. sometimes ironic pinto beans about the dolphin');
   INSERT INTO `orders_2` VALUES (17, 0, 'P', 65883.92, '1995-03-18', '1-URGENT', 'Clerk#000000632', 0, 'ular requests are blithely pending orbits-- even requests against the deposit');
   INSERT INTO `orders_2` VALUES (18, 6, 'F', 79258.24, '1994-01-20', '5-LOW', 'Clerk#000000743', 0, 'y pending requests integrate');
   INSERT INTO `orders_2` VALUES (19, 2, 'O', 116227.05, '1996-12-19', '4-NOT SPECIFIED', 'Clerk#000000547', 0, 'symptotes haggle slyly around the furiously iron');
   INSERT INTO `orders_2` VALUES (20, 1, 'O', 215135.72, '1998-04-18', '3-MEDIUM', 'Clerk#000000440', 0, ' pinto beans sleep carefully. blithely ironic deposits haggle furiously acro');

步骤二：创建Catalog

1. 进入元数据管理页面。

   1. 登录实时计算控制台。

   2. 单击目标工作空间操作列下的控制台。

   3. 单击左侧的元数据管理。

2. 创建Paimon Catalog。

   1. 单击创建Catalog，在内置Catalog页签，选择Apache Paimon后，单击下一步。

   2. 填写配置信息。

      

      配置项	说明	备注
      catalog name	您自定义的Paimon Catalog名称。	填写为自定义的英文名。
      metastore	Paimon表的元数据存储类型： filesystem：仅将元数据存储在OSS中。 dlf：除了将元数据存储在OSS上外，还会将元数据同步到阿里云数据湖构建服务DLF中。	本文选择filesystem。
      warehouse	Paimon Catalog的存储根目录，是一个OSS目录。可以选择创建实时计算Flink版时使用的OSS Bucket，也可以使用同一账号同一地域下的其他OSS Bucket。	格式为oss://<bucket>/<object>。其中： bucket：表示您创建的OSS Bucket名称。 object：表示您存放数据的路径。 您可以在OSS管理控制台上查看您的bucket和object名称。

   3. 单击确定。

步骤三：创建Flink作业

1. 在数据开发 > ETL页面，单击新建。

2. 选择空白的流作业草稿，单击下一步。

3. 在新建作业草稿对话框，填写作业配置信息。

   作业参数	说明
   文件名称	作业的名称。 说明 作业名称在当前项目中必须保持唯一。
   存储位置	指定该作业的存储位置。 您还可以在现有文件夹右侧，单击 图标，新建子文件夹。
   引擎版本	当前作业使用的Flink的引擎版本。引擎版本号含义、版本对应关系和生命周期重要时间点详情请参见引擎版本介绍。

4. 单击创建。

5. 输入以下语句，实时捕获orders数据库中相关表的变化，并同步到Paimon表中。

   -- 使用刚刚创建的Paimon Catalog
   USE CATALOG `test`;
   
   -- 创建一张MySQL临时表，捕获表名符合正则表达式orders_\d+的MySQL表的变化
   CREATE TEMPORARY TABLE mysql_orders (
       orderkey BIGINT,
       custkey BIGINT,
       order_status VARCHAR(100),
       total_price DOUBLE,
       order_date DATE,
       order_priority VARCHAR(100),
       clerk VARCHAR(100),
       ship_priority INT,
       `comment` VARCHAR(100),
       PRIMARY KEY (orderkey) NOT ENFORCED
   ) WITH (
       'connector' = 'mysql',
       'hostname' = 'rm-bp1s1xgll21ey****.mysql.rds.aliyuncs.com',
       'port' = '3306',
       'username' = 'your_username',
       'password' = '${secret_values.mysql_pw}',
       'database-name' = 'orders',
       'table-name' = 'orders_\d+',
       'server-time-zone' = 'Asia/Shanghai'
   );
   
   -- 将MySQL表的变化同步到Paimon表中
   CREATE TABLE IF NOT EXISTS orders AS TABLE mysql_orders;

   ​参数说明如下，您可以根据实际情况进行修改。MySQL连接器更多参数详情请参见MySQL。

   参数	说明	备注
   connector	连接器类型。	本示例固定值为mysql。
   hostname	MySQL数据库的IP地址或者Hostname。	本文填写为RDS实例的内网地址。
   username	MySQL数据库服务的用户名。	无。
   password	MySQL数据库服务的密码。	本示例通过使用名为mysql_pw密钥的方式填写密码值，避免信息泄露，详情请参见变量管理。
   database-name	MySQL数据库名称。	本示例填写为步骤一：准备数据源中创建的数据库。
   table-name	MySQL表名。	作为源表时，表名支持正则表达式以读取多个表的数据。
   port	MySQL数据库服务的端口号。	无。

6. （可选）单击右上方的深度检查，确认作业Flink SQL语句中是否存在语法错误。

7. 单击右上方的部署，单击确定。

8. 在左侧导航栏，单击运维中心 > 作业运维，单击目标作业名称，进入作业部署详情页面。

9. 单击运行参数配置区域右侧的编辑。

   本文为了更快观察到任务运行的结果，将系统检查点间隔和两次系统检查点之间的最短时间间隔均改为10s，单击保存。

   

10. 在目标作业部署详情页顶部，单击启动，选择无状态启动。

    

11. 查询Paimon数据。

    1. 在数据开发 > 数据查询页面的查询脚本页签，将如下代码拷贝到查询脚本后，选中目标片段后单击左侧代码行上的运行。

       select custkey, sum(total_price) from `test`.`default`.`orders` group by custkey;

    2. 结果浏览完成后，单击左侧的

       

       停止调试。

       

步骤四：更新MySQL表结构

本部分将演示MySQL表结构变更同步到Paimon表的功能。

1. 登录云数据库RDS控制台。

2. 在orders数据库，输入如下SQL语句，然后单击执行，为两张数据表添加一列，并填充一些数据。

   ALTER TABLE `orders_1` ADD COLUMN quantity BIGINT; 
   ALTER TABLE `orders_2` ADD COLUMN quantity BIGINT; 
   UPDATE `orders_1` SET `quantity` = 100 WHERE `orderkey` < 5;
   UPDATE `orders_2` SET `quantity` = 100 WHERE `orderkey` > 15;

3. 在实时计算控制台数据开发 > 数据查询页面的查询脚本页签，将如下代码拷贝到查询脚本后，选中目标片段后单击左侧代码行上的运行。

   select * from `test`.`default`.`orders` where `quantity` is not null;

   结果如下，浏览完成后，可单击左侧的

   

   停止调试。