品牌网站建设小蝌蚪,成都网络推广服务,saas小程序,网络公司名字免费起名大全Apache Paimon是一种流批统一的湖存储格式#xff0c;支持高吞吐的写入和低延迟的查询。本文通过Paimon Catalog和MySQL连接器#xff0c;将云数据库RDS中的订单数据和表结构变更导入Paimon表中#xff0c;并使用Flink对Paimon表进行简单分析。
背景信息
Apache Paimon是一…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版。
步骤一准备数据源 快速创建RDS MySQL实例。 说明 RDS MySQL版实例需要与Flink工作空间处于同一VPC。不在同一VPC下时请参见网络连通性。 创建数据库和账号。 创建名称为orders的数据库并创建高权限账号或具有数据库orders读写权限的普通账号。 通过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)
); 插入如下测试数据。 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 进入元数据管理页面。 登录实时计算控制台。 单击目标工作空间操作列下的控制台。 单击左侧的元数据管理。 创建Paimon Catalog。 单击创建Catalog在内置Catalog页签选择Apache Paimon后单击下一步。 填写配置信息。 配置项 说明 备注 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名称。 单击确定。
步骤三创建Flink作业 在数据开发 ETL页面单击新建。 选择空白的流作业草稿单击下一步。 在新建作业草稿对话框填写作业配置信息。 作业参数 说明 文件名称 作业的名称。 说明 作业名称在当前项目中必须保持唯一。 存储位置 指定该作业的存储位置。 您还可以在现有文件夹右侧单击 图标新建子文件夹。 引擎版本 当前作业使用的Flink的引擎版本。引擎版本号含义、版本对应关系和生命周期重要时间点详情请参见引擎版本介绍。 单击创建。 输入以下语句实时捕获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数据库服务的端口号。 无。 可选单击右上方的深度检查确认作业Flink SQL语句中是否存在语法错误。 单击右上方的部署单击确定。 在左侧导航栏单击运维中心 作业运维单击目标作业名称进入作业部署详情页面。 单击运行参数配置区域右侧的编辑。 本文为了更快观察到任务运行的结果将系统检查点间隔和两次系统检查点之间的最短时间间隔均改为10s单击保存。 在目标作业部署详情页顶部单击启动选择无状态启动。 查询Paimon数据。 在数据开发 数据查询页面的查询脚本页签将如下代码拷贝到查询脚本后选中目标片段后单击左侧代码行上的运行。 select custkey, sum(total_price) from test.default.orders group by custkey; 结果浏览完成后单击左侧的 停止调试。
步骤四更新MySQL表结构
本部分将演示MySQL表结构变更同步到Paimon表的功能。 登录云数据库RDS控制台。 在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; 在实时计算控制台数据开发 数据查询页面的查询脚本页签将如下代码拷贝到查询脚本后选中目标片段后单击左侧代码行上的运行。 select * from test.default.orders where quantity is not null; 结果如下浏览完成后可单击左侧的 停止调试。
