創建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下推，提升批作業執行效率。