基于Flink+Paimon+StarRocks搭建流式湖仓

准备MySQL CDC数据源

本文以云数据库RDS MySQL版为例，创建数据库名称为order_dw，并创建三张业务表及数据。

1. 快速创建RDS MySQL实例。

   重要

   RDS MySQL版实例需要与Flink工作空间处于同一VPC。不在同一VPC下时请参见如何访问跨VPC的其他服务？

2. 创建数据库和账号。

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

   创建三张表，并插入相应数据。

   CREATE TABLE `orders` (
     order_id bigint not null primary key,
     user_id varchar(50) not null,
     shop_id bigint not null,
     product_id bigint not null,
     buy_fee bigint not null,   
     create_time timestamp not null,
     update_time timestamp not null default now(),
     state int not null
   );
   
   CREATE TABLE `orders_pay` (
     pay_id bigint not null primary key,
     order_id bigint not null,
     pay_platform int not null, 
     create_time timestamp not null
   );
   
   CREATE TABLE `product_catalog` (
     product_id bigint not null primary key,
     catalog_name varchar(50) not null
   );
   
   -- 准备数据
   INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
   
   INSERT INTO orders VALUES
   (100001, 'user_001', 12345, 1, 5000, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
   (100002, 'user_002', 12346, 2, 4000, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
   (100003, 'user_003', 12347, 3, 3000, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
   (100004, 'user_001', 12347, 4, 2000, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
   (100005, 'user_002', 12348, 5, 1000, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
   (100006, 'user_001', 12348, 1, 1000, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
   (100007, 'user_003', 12347, 4, 2000, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
   
   INSERT INTO orders_pay VALUES
   (2001, 100001, 1, '2023-02-15 17:40:56'),
   (2002, 100002, 1, '2023-02-15 17:40:56'),
   (2003, 100003, 0, '2023-02-15 17:40:56'),
   (2004, 100004, 0, '2023-02-15 17:40:56'),
   (2005, 100005, 0, '2023-02-15 18:40:56'),
   (2006, 100006, 0, '2023-02-15 18:40:56'),
   (2007, 100007, 0, '2023-02-15 18:40:56');

管理元数据

创建Paimon Catalog

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

2. 在左侧导航栏，选择元数据管理页面，单击创建Catalog。

3. 在内置Catalog页签，单击Apache Paimon，单击下一步。

4. 填写以下参数，单击确定，新建名为paimoncatalog的Paimon Catalog。

   配置项	说明	是否必填	备注
   catalog name	Catalog名称。	是	填写为自定义的英文名。 本文以paimoncatalog为例。
   metastore	元数据存储类型。	是	此示例选择为filesystem存储类型。
   warehouse	OSS服务中所指定的数仓目录。	是	格式为oss://<bucket>/<object>。其中： bucket：表示您创建的OSS Bucket名称。 object：表示您存放数据的路径。 您可以在OSS管理控制台上查看您的bucket和object名称。
   fs.oss.endpoint	OSS服务的连接地址。	是	获取方法请参见OSS地域和访问域名和创建AccessKey。
   fs.oss.accessKeyId	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey ID。	是
   fs.oss.accessKeySecret	拥有读写OSS权限的阿里云账号或RAM账号的Accesskey secret。	是

创建MySQL Catalog

1. 在元数据管理页面，单击创建Catalog。

2. 在内置Catalog页签，单击MySQL，单击下一步。

3. 填写以下参数，单击确定，新建名为mysqlcatalog的MySQL Catalog。

   配置项	说明	是否必填	备注
   catalog name	Catalog名称。	是	填写为自定义的英文名。本文以mysqlcatalog为例。
   hostname	MySQL数据库的IP地址或者Hostname。	是	详情请参见查看和管理实例连接地址和端口。由于RDS MySQL版实例和Flink全托管处于相同VPC，此处应填写内网地址。
   port	MySQL数据库服务的端口号，默认值为3306。	否	详情请参见查看和管理实例连接地址和端口。
   default-database	默认的MySQL数据库名称。	是	本文填写需要同步的数据库名order_dw。
   username	MySQL数据库服务的用户名。	是	本文为准备MySQL CDC数据源中创建的账号。
   password	MySQL数据库服务的密码。	是	本文为准备MySQL CDC数据源中创建的密码。

构建ODS层：业务数据库实时入仓

基于CREATE DATABASE AS（CDAS）语句功能，可以一次性将ODS层建出来。SQL作业中通过SET语句指定的配置项也可以在运维中心 > 作业运维页面的作业部署详情页签指定，详见如何配置自定义的作业运行参数？。Paimon写入性能优化请参见Apache Paimon官方文档。

1. 创建CDAS同步作业。

   1. 在实时计算控制台的数据开发 > ETL页面，新建名为ods的SQL流作业，将如下代码复制到SQL编辑器。

      SET 'execution.checkpointing.max-concurrent-checkpoints' = '3'; -- 减轻检查点长尾的影响。
      SET 'table.exec.sink.upsert-materialize' = 'NONE'; -- 消除无用的SinkMaterialize算子。
      
      -- Paimon结果表在每次检查点完成之后才会正式提交数据。
      -- 此处将检查点间隔缩短为10s，是为了更快地看到结果。
      -- 在生产环境下，系统检查点的间隔与两次系统检查点之间的最短时间间隔根据业务对延时要求的不同，一般设置为1分钟到10分钟。
      SET 'execution.checkpointing.interval' = '10s';
      SET 'execution.checkpointing.min-pause' = '10s';
      
      CREATE DATABASE IF NOT EXISTS paimoncatalog.order_dw
      WITH (
        'changelog-producer' = 'input' -- 因为输入数据就是MySQL产生的binlog，已经是完整的变更数据，所以可以直接把输入数据作为变更数据。
      ) AS DATABASE mysqlcatalog.order_dw INCLUDING all tables; -- 也可以根据需要选择上游数据库需要入仓的表。

   2. 单击右上方的部署，进行作业部署。

   3. 在运维中心 > 作业运维，单击刚刚部署的ods作业操作列的启动，选择无状态启动启动作业。

2. 查看MySQL同步到Paimon的三张表的数据。

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

   SELECT * FROM paimoncatalog.order_dw.orders ORDER BY order_id;

   

构建DWD层：主题宽表

1. 创建DWD层Paimon宽表dwd_orders

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

   CREATE TABLE paimoncatalog.order_dw.dwd_orders (
       order_id BIGINT,
       order_user_id STRING,
       order_shop_id BIGINT,
       order_product_id BIGINT,
       order_product_catalog_name STRING,
       order_fee BIGINT,
       order_create_time TIMESTAMP,
       order_update_time TIMESTAMP,
       order_state INT,
       pay_id BIGINT,
       pay_platform INT COMMENT 'platform 0: phone, 1: pc',
       pay_create_time TIMESTAMP,
       PRIMARY KEY (order_id) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'partial-update', -- 使用部分更新数据合并机制产生宽表
       'changelog-producer' = 'lookup' -- 使用lookup增量数据产生机制以低延时产出变更数据
   );

   返回Query has been executed表示创建成功。

2. 实时消费ODS层orders、orders_pay表的变更数据

   在实时计算控制台的数据开发 > ETL页面，新建名为dwd的SQL流作业，并将如下代码复制到SQL编辑器后，部署作业并无状态启动作业。​

   通过该SQL作业，orders表会与product_catalog表进行维表关联，关联后的结果将与orders_pay一起写入dwd_orders表中，利用Paimon表的部分更新数据合并机制，将orders表和orders_pay表中order_id相同的数据进行打宽。

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   
   -- Paimon目前暂不支持在同一个作业里通过多条INSERT语句写入同一张表，因此这里使用UNION ALL。
   INSERT INTO paimoncatalog.order_dw.dwd_orders 
   SELECT 
       o.order_id,
       o.user_id,
       o.shop_id,
       o.product_id,
       dim.catalog_name,
       o.buy_fee,
       o.create_time,
       o.update_time,
       o.state,
       NULL,
       NULL,
       NULL
   FROM
       paimoncatalog.order_dw.orders o 
       LEFT JOIN paimoncatalog.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
       ON o.product_id = dim.product_id
   UNION ALL
   SELECT
       order_id,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       pay_id,
       pay_platform,
       create_time
   FROM
       paimoncatalog.order_dw.orders_pay;

3. 查看宽表dwd_orders的数据

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

   SELECT * FROM paimoncatalog.order_dw.dwd_orders ORDER BY order_id;

   

构建DWS层：指标计算

1. 创建DWS层的聚合表dws_users以及dws_shops

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

   -- 用户维度聚合指标表。
   CREATE TABLE paimoncatalog.order_dw.dws_users (
       user_id STRING,
       ds STRING,
       payed_buy_fee_sum BIGINT COMMENT '当日完成支付的总金额',
       PRIMARY KEY (user_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- 使用预聚合数据合并机制产生聚合表
       'fields.payed_buy_fee_sum.aggregate-function' = 'sum' -- 对 payed_buy_fee_sum 的数据求和产生聚合结果
       -- 由于dws_users表不再被下游流式消费，因此无需指定增量数据产生机制
   );
   
   -- 商户维度聚合指标表。
   CREATE TABLE paimoncatalog.order_dw.dws_shops (
       shop_id BIGINT,
       ds STRING,
       payed_buy_fee_sum BIGINT COMMENT '当日完成支付总金额',
       uv BIGINT COMMENT '当日不同购买用户总人数',
       pv BIGINT COMMENT '当日购买用户总人次',
       PRIMARY KEY (shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- 使用预聚合数据合并机制产生聚合表
       'fields.payed_buy_fee_sum.aggregate-function' = 'sum', -- 对 payed_buy_fee_sum 的数据求和产生聚合结果
       'fields.uv.aggregate-function' = 'sum', -- 对 uv 的数据求和产生聚合结果
       'fields.pv.aggregate-function' = 'sum' -- 对 pv 的数据求和产生聚合结果
       -- 由于dws_shops表不再被下游流式消费，因此无需指定增量数据产生机制
   );
   
   -- 为了同时计算用户视角的聚合表以及商户视角的聚合表，另外创建一个以用户 + 商户为主键的中间表。
   CREATE TABLE paimoncatalog.order_dw.dwm_users_shops (
       user_id STRING,
       shop_id BIGINT,
       ds STRING,
       payed_buy_fee_sum BIGINT COMMENT '当日用户在商户完成支付的总金额',
       pv BIGINT COMMENT '当日用户在商户购买的次数',
       PRIMARY KEY (user_id, shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- 使用预聚合数据合并机制产生聚合表
       'fields.payed_buy_fee_sum.aggregate-function' = 'sum', -- 对 payed_buy_fee_sum 的数据求和产生聚合结果
       'fields.pv.aggregate-function' = 'sum', -- 对 pv 的数据求和产生聚合结果
       'changelog-producer' = 'lookup', -- 使用lookup增量数据产生机制以低延时产出变更数据
       -- dwm层的中间表一般不直接提供上层应用查询，因此可以针对写入性能进行优化。
       'file.format' = 'avro', -- 使用avro行存格式的写入性能更加高效。
       'metadata.stats-mode' = 'none' -- 放弃统计信息会增加OLAP查询代价（对持续的流处理无影响），但会让写入性能更加高效。
   );

   返回Query has been executed表示创建成功。

2. DWD层dwd_orders表的变更数据

   在实时计算控制台数据开发 > ETL页签，新建名为dwm的SQL流作业，并将如下代码复制到SQL编辑器后，部署作业并无状态启动作业。

   通过该SQL作业，dwd_orders表的数据会写入dwm_users_shops表中，利用Paimon表的预聚合数据合并机制，自动对order_fee求和，算出用户在商户的消费总额。同时，自动对1求和，也能算出用户在商户的消费次数。

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   
   INSERT INTO paimoncatalog.order_dw.dwm_users_shops
   SELECT
       order_user_id,
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       order_fee,
       1 -- 一条输入记录代表一次消费
   FROM paimoncatalog.order_dw.dwd_orders
   WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL;

3. 实时消费DWM层dwm_users_shops表的变更数据

   在实时计算控制台的数据开发 > ETL页面，新建名为dws的SQL流作业，并将如下代码复制到SQL编辑器后，部署作业并无状态启动作业。

   通过该SQL作业，dwm_users_shops表的数据会写入dws_users表和dws_shops表中，利用Paimon表的预聚合数据合并机制，在dws_users表中，计算每个用户的总消费额（payed_buy_fee_sum），在dws_shops表中计算商户的总流水（payed_buy_fee_sum），商户的消费用户数量（对1求和）和消费总人次（pv）。

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   
   -- 与dwd不同，此处每一条INSERT语句写入的是不同的Paimon表，可以放在同一个作业中。
   BEGIN STATEMENT SET;
   
   INSERT INTO paimoncatalog.order_dw.dws_users
   SELECT 
       user_id,
       ds,
       payed_buy_fee_sum
   FROM paimoncatalog.order_dw.dwm_users_shops;
   
   -- 以商户为主键，部分热门商户的数据量可能远高于其他商户。
   -- 因此使用local merge在写入Paimon之前先在内存中进行预聚合，缓解数据倾斜问题。
   INSERT INTO paimoncatalog.order_dw.dws_shops /*+ OPTIONS('local-merge-buffer-size' = '64mb') */
   SELECT
       shop_id,
       ds,
       payed_buy_fee_sum,
       1, -- 一条输入记录代表一名用户在该商户的所有消费
       pv
   FROM paimoncatalog.order_dw.dwm_users_shops;
   
   END;

4. 查看dws_users表和dws_shops表的数据

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

   --查看dws_users表数据
   SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

   

   --查看dws_shops表数据
   SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

   

捕捉业务数据库的变化

前面已完成了流式湖仓的构建，下面将测试流式湖仓捕捉业务数据库变化的能力。

1. 向MySQL的order_dw数据库中插入如下数据。​

   INSERT INTO orders VALUES
   (100008, 'user_001', 12345, 3, 3000, '2023-02-15 17:40:56', '2023-02-15 18:42:56', 1),
   (100009, 'user_002', 12348, 4, 1000, '2023-02-15 18:40:56', '2023-02-15 19:42:56', 1),
   (100010, 'user_003', 12348, 2, 2000, '2023-02-15 19:40:56', '2023-02-15 20:42:56', 1);
   
   INSERT INTO orders_pay VALUES
   (2008, 100008, 1, '2023-02-15 18:40:56'),
   (2009, 100009, 1, '2023-02-15 19:40:56'),
   (2010, 100010, 0, '2023-02-15 20:40:56');

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

   • dws_users表

     SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

     

   • dws_shops表

     SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;