基于Flink搭建流式湖仓OpenLake方案

创建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数据源中创建的账号和密码。
   password	MySQL数据库服务的密码。	是

   关于MySQL Catalog的更多使用方法详情请参见管理MySQL Catalog。

构建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;