基于Flink+Hologres搭建实时数仓

创建Catalog

1. 创建Hologres Catalog。

   在实时计算控制台SQL开发页面的查询脚本页签，将如下代码拷贝到查询脚本，并修改目标参数取值，选中目标片段后单击左侧代码行上的运行。

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = '<USERNAME>',
     'password' = '${secret_values.ak_holo}',
     'dbname' = 'order_dw',
     'binlog' = 'true', -- 创建catalog时可以设置源表、维表和结果表支持的with参数，之后在使用此catalog下的表时会默认添加这些默认参数。
     'sdkMode' = 'jdbc', -- 推荐使用jdbc模式。
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- 宽表merge需要开启，防止回撤。
     'partial-insert.enabled' = 'true', -- 宽表merge需要开启此参数，实现部分列更新。
     'mutateType' = 'insertOrUpdate', -- 宽表merge需要开启此参数，实现部分列更新。
     'table_property.binlog.level' = 'replica', --也可以在创建catalog时传入持久化的hologres表属性，之后创建表时，默认都开启binlog。
     'table_property.binlog.ttl' = '259200'
   );
   

   您需要修改以下参数取值为您实际Hologres服务信息。

   参数	说明	备注
   endpoint	Hologres的Endpoint地址。	详情请参见实例配置。
   username	阿里云账号的AccessKey ID。	当前配置的AccessKey对应的用户需要能够访问所有的Hologres数据库，Hologres数据库权限请参见Hologres权限模型概述。 为了避免您的AK信息泄露，本示例通过使用名为ak_holo密钥的方式填写AccessKey Secret取值，详情请参见密钥管理。
   password	阿里云账号的AccessKey Secret。

   说明

   创建Catalog时可以设置默认的源表、维表和结果表的WITH参数，也可以设置创建Hologres物理表的默认属性，例如上方table_property开头的参数。详情请参见管理Hologres Catalog和实时数仓Hologres WITH参数。

2. 创建MySQL Catalog。

   将如下代码拷贝到查询脚本，并修改目标参数取值，选中目标片段后单击左侧代码行上的运行。

   CREATE CATALOG mysqlcatalog WITH(
     'type' = 'mysql',
     'hostname' = '<hostname>',
     'port' = '<port>',
     'username' = '<username>',
     'password' = '${secret_values.mysql_pw}',
     'default-database' = 'order_dw'
   );

   您需要修改以下参数取值为您实际的MySQL服务信息。

   参数	说明
   hostname	MySQL数据库的IP地址或者Hostname。
   port	MySQL数据库服务的端口号，默认值为3306。
   username	MySQL数据库服务的用户名。
   password	MySQL数据库服务的密码。 本示例通过使用名为mysql_pw密钥的方式填写密码取值，避免信息泄露，详情请参见密钥管理。

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

基于Catalog的CREATE DATABASE AS（CDAS）语句功能，可以一次性把ODS层建出来。ODS层一般不直接做OLAP或SERVING（KV点查），主要作为流式作业的事件驱动，开启binlog即可满足需求。Binlog是Hologres的核心能力之一，Hologres连接器也支持先全量读取再增量消费Binlog的全增量模式。

1. 创建CDAS同步作业ODS。

   1. 在实时计算控制台上，新建名为ODS的SQL流作业，并将如下代码拷贝到SQL编辑器。

      CREATE DATABASE IF NOT EXISTS dw.order_dw   -- 创建catalog时设置了table_property.binlog.level参数，因此通过CDAS创建的所有表都开启了binlog。
      AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- 可以根据需要选择上游数据库需要入仓的表。
      /*+ OPTIONS('server-id'='8001-8004') */ ;   -- 指定mysql-cdc源表。

      说明

      本示例默认将数据同步到数据库order_dw的Public Schema下。您也可以将数据同步到Hologres目标库的指定Schema中，详情请参见作为CDAS的目标端Catalog，指定后使用Catalog时的表名格式也会发生变化，详情请参见使用Hologres Catalog。

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

   3. 单击左侧导航栏的作业运维，单击刚刚部署的ODS作业操作列的启动，选择无状态启动启动作业。

2. 查看MySQL同步到Hologres的3张表数据。

   在HoloWeb开发页面连接Hologres实例并登录目标数据库后，在SQL编辑器上执行如下命令。

   ---查orders中的数据。
   SELECT * FROM orders;
   
   ---查orders_pay中的数据。
   SELECT * FROM orders_pay;
   
   ---查product_catalog中的数据。
   SELECT * FROM product_catalog;

   

构建DWD层：实时主题宽表

构建DWD层用到了Hologres连接器特有的部分列更新能力，可以使用INSERT DML方便地表达部分列更新的语义。作业中需要对不同的维表进行查询，是基于Hologres行存以及行列共存表提供的高性能的点查能力。同时，Hologres资源强隔离的架构，可以保证写入、读取、分析等作业之间互不干扰。

1. 通过Flink Catalog功能在Hologres中建DWD层的宽表dwd_orders。

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

   -- 宽表字段要nullable，因为不同的流写入到同一张结果表，每一列都可能出现null的情况。
   CREATE TABLE dw.order_dw.dwd_orders (
     order_id bigint not null,
     order_user_id string,
     order_shop_id bigint,
     order_product_id bigint,
     order_product_catalog_name string,
     order_fee numeric(20,2),
     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
   );
   
   -- 支持通过catalog修改Hologres物理表属性。
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' --修改binlog的超时时间为一周。
   );

2. 实现实时消费ODS层orders、orders_pay表的binlog。

   在实时计算控制台上，新建名为DWD的SQL作业，并将如下代码拷贝到SQL编辑器后，部署并启动作业。通过如下SQL作业，orders表会与product_catalog表进行维表关联，将最终结果写入dwd_orders表中，实现数据的实时打宽。

   BEGIN STATEMENT SET;
   
   INSERT INTO dw.order_dw.dwd_orders 
    (
      order_id,
      order_user_id,
      order_shop_id,
      order_product_id,
      order_fee,
      order_create_time,
      order_update_time,
      order_state,
      order_product_catalog_name
    ) SELECT o.*, dim.catalog_name 
      FROM dw.order_dw.orders as o
      LEFT JOIN dw.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
      ON o.product_id = dim.product_id;
   
   INSERT INTO dw.order_dw.dwd_orders 
     (pay_id, order_id, pay_platform, pay_create_time)
      SELECT * FROM dw.order_dw.orders_pay;
   
   END;

3. 查看宽表dwd_orders数据。

   在HoloWeb开发页面连接Hologres实例并登录目标数据库后，在SQL编辑器上执行如下命令。

   SELECT * FROM dwd_orders;

   

构建DWS层：实时指标计算

1. 通过Flink Catalog功能，在Hologres中创建dws层的聚合dws_users以及dws_shops。

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

   -- 用户维度聚合指标表。
   CREATE TABLE dw.order_dw.dws_users (
     user_id string not null,
     ds string not null,
     paied_buy_fee_sum numeric(20,2) not null comment '当日完成支付的总金额',
     primary key(user_id,ds) NOT ENFORCED
   );
   
   -- 商户维度聚合指标表。
   CREATE TABLE dw.order_dw.dws_shops (
     shop_id bigint not null,
     ds string not null,
     paied_buy_fee_sum numeric(20,2) not null comment '当日完成支付总金额',
     primary key(shop_id,ds) NOT ENFORCED
   );

2. 实时消费DWD层的宽表dw.order_dw.dwd_orders，在Flink中做聚合计算，最终写入Hologres中的DWS表。

   在实时计算控制台上，新建名为DWS的SQL流作业，并将如下代码拷贝到SQL编辑器后，部署并启动作业。

   BEGIN STATEMENT SET;
   
   INSERT INTO dw.order_dw.dws_users
     SELECT 
       order_user_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       SUM (order_fee)
       FROM dw.order_dw.dwd_orders c
       WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- 订单流和支付流数据都已写入宽表。
       GROUP BY order_user_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   
   INSERT INTO dw.order_dw.dws_shops
     SELECT 
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       SUM (order_fee)
      FROM dw.order_dw.dwd_orders c
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- 订单流和支付流数据都已写入宽表。
      GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   END;

3. 查看DWS层的聚合结果，其结果会根据上游数据的变更实时更新。

   在HoloWeb开发页面连接Hologres实例并登录目标数据库后，在SQL编辑器上执行如下命令。

   • 查询dws_users表结果。

     SELECT * FROM dws_users;

     

   • 查询dws_shops表结果。

     SELECT * FROM dws_shops;

     

数据探查

因为开启了Binlog，所以可直接探查到数据的变化情况。如果对中间结果需要即系（Ad-hoc）性质的业务数据探查，或者对最终计算结果进行数据正确性排查，此方案的每一层数据都实现了持久化，可以便捷的探查中间过程。

• 流模式探查

  1. 新建并启动数据探查流作业。

     在实时计算控制台上，新建名为Data-exploration的SQL流作业，并将如下代码拷贝到SQL编辑器后，部署并启动作业。

     -- 流模式探查，打印到print可以看到数据的变化情况。
     CREATE TEMPORARY TABLE print_sink(
       order_id bigint not null,
       order_user_id string,
       order_shop_id bigint,
       order_product_id bigint,
       order_product_catalog_name string,
       order_fee numeric(20,2),
       order_create_time timestamp,
       order_update_time timestamp,
       order_state int,
       pay_id bigint,
       pay_platform int,
       pay_create_time timestamp,
       PRIMARY KEY(order_id) NOT ENFORCED
     ) WITH (
       'connector' = 'print'
     );
     
     INSERT INTO print_sink SELECT *
     FROM dw.order_dw.dwd_orders /*+ OPTIONS('startTime'='2023-02-15 12:00:00') */ --这里的startTime是binlog生成的时间
     WHERE order_user_id = 'user_001';

  2. 查看数据探查结果。

     在作业运维详情页面，单击目标作业名称，在作业探查页签下左侧运行日志页签，单击运行Task Managers页签下的Path, ID。在Stdout页面搜索user_001相关的日志信息。

     

• 批模式探查

  在实时计算控制台上，创建SQL流作业，并将如下代码拷贝到SQL编辑器后，单击调试。详情请参见作业调试。

  批模式探查是获取当前时刻的终态数据，在Flink作业开发界面调试结果如下图所示。

  SELECT *
  FROM dw.order_dw.dwd_orders /*+ OPTIONS('binlog'='false') */ 
  WHERE order_user_id = 'user_001' and order_create_time > '2023-02-15 12:00:00'; --批量模式支持filter下推，提升批作业执行效率。