背景信息

• 傳統(tǒng)數(shù)倉架構(gòu)

  傳統(tǒng)中要實現(xiàn)離線、近實時或?qū)崟r的查詢需求，需要引用多套系統(tǒng)來實現(xiàn)。傳統(tǒng)數(shù)倉Lambda架構(gòu)中存在一些問題，包括架構(gòu)復雜、組件繁多、運維難度高及成本大。由于計算引擎的不一致性，數(shù)據(jù)加工標準難以統(tǒng)一，同時造成了數(shù)據(jù)冗余和同步難題，增加了資源消耗。此外，缺乏有效的數(shù)據(jù)倉庫分層策略，導致查詢延遲問題突出，操作不便，開發(fā)效率低下。

  

• 基于Hologres Dynamic Table的多模式統(tǒng)一計算架構(gòu)

  Hologres從3.0版本開始支持Dynamic Table，提供全量刷新和增量刷新兩種刷新模式，支持數(shù)據(jù)自動刷新與流動刷新，通過多模式的統(tǒng)一計算，能有效地解決實時數(shù)倉分層、流批一體等強需求，同時也能實現(xiàn)性能、成本、時效性等的完美平衡。

  

場景介紹

Hologres Dynamic Table增量刷新和全量刷新模式介紹如下：

• 增量刷新：場景1：累計增量成交分析。

• 全量刷新：場景2：歷史廣告數(shù)據(jù)分析與數(shù)據(jù)回刷。

數(shù)據(jù)鏈路如下。

• 數(shù)據(jù)源：本次示例中的數(shù)據(jù)來源為MaxCompute中的公開數(shù)據(jù)集，需要將數(shù)據(jù)寫入到Hologres內(nèi)部表，成為Dynamic Table的Base表。

• 數(shù)據(jù)加工：在Hologres中將Base表的數(shù)據(jù)，加工成兩張Dynamic Table，分別配置增量刷新和全量刷新，形成DWS，滿足業(yè)務(wù)的不同查詢時效性。

• 數(shù)據(jù)應(yīng)用：本次示例使用DataV展示Dynamic Table中數(shù)據(jù)的變化情況。

準備工作

--創(chuàng)建外表
CREATE SCHEMA foreign_holo;
IMPORT FOREIGN SCHEMA "BIGDATA_PUBLIC_DATASET#commerce" LIMIT to
(adv_raw_sample,adv_ad_feature,user_profile,behavior_log) 
  FROM server odps_server INTO foreign_holo options(if_table_exist 'error');


--創(chuàng)建廣告基本信息維表
DROP TABLE IF EXISTS public.dim_ad_feature;
BEGIN;
CREATE TABLE public.dim_ad_feature (
   adgroup_id bigint NOT NULL,
    cate_id bigint,
    campaign_id bigint,
    customer_id bigint,
    brand bigint,
    price double precision
    ,PRIMARY KEY (adgroup_id)
) WITH (
orientation = 'column',
storage_format = 'orc',
distribution_key = 'adgroup_id'
);
END;
--導入維表數(shù)據(jù)
INSERT INTO dim_ad_feature SELECT 
   try_cast(adgroup_id as bigint) as adgroup_id,
   try_cast(cate_id as bigint) as cate_id,
   try_cast(campaign_id as bigint) as campaign_id,
   try_cast(customer_id as bigint) as customer_id,
   try_cast(brand as bigint) as brand,
    price::double precision
 FROM foreign_holo.adv_ad_feature;

--創(chuàng)建用戶行為表
DROP TABLE IF EXISTS public.behavior_log;
BEGIN;
CREATE TABLE public.behavior_log (
    user_id bigint,
    ds text,
    "time" text,
    behavior_type text,
    cate bigint,
    brand bigint
) WITH (
orientation = 'column',
storage_format = 'orc',
binlog_level = 'replica',
binlog_ttl = '2592000',
bitmap_columns = 'ds,"time",behavior_type',
dictionary_encoding_columns = 'ds:auto,"time":auto,behavior_type:auto',
distribution_key = 'cate'
);
END;
--導入數(shù)據(jù)
INSERT INTO behavior_log SELECT 
    "user"::bigint,
    date(to_timestamp(time_stamp::bigint)),
    to_timestamp(time_stamp::bigint),
    btag::text,
    cate::bigint,
    brand::bigint
 FROM foreign_holo.behavior_log 
 WHERE date(to_timestamp(time_stamp::bigint))='2017-05-13' ;


--創(chuàng)建廣告明細表
DROP TABLE IF EXISTS public.raw_sample;
BEGIN;
CREATE TABLE public.raw_sample (
    user_id bigint,
    adgroup_id bigint,
    ds text,
    "time" text,
    pid text,
    noclk integer,
    clk integer
) WITH (
orientation = 'column',
storage_format = 'orc',
binlog_level = 'replica',
binlog_ttl = '2592000',
bitmap_columns = 'ds,"time",pid',
dictionary_encoding_columns = 'ds:auto,"time":auto,pid:auto',
distribution_key = 'user_id'
);
END;
--導入數(shù)據(jù)
INSERT INTO raw_sample SELECT 
    user_id::bigint,
    adgroup_id::bigint,
    date(to_timestamp(time_stamp::bigint)),
    to_timestamp(time_stamp::bigint),
    pid::text,
    noclk::integer,
    clk::integer
 FROM foreign_holo.adv_raw_sample;


--創(chuàng)建用戶基礎(chǔ)信息維表
DROP TABLE IF EXISTS public.dim_user_profile;
BEGIN;
CREATE TABLE public.dim_user_profile (
    user_id bigint NOT NULL,
    cms_seg_id bigint,
    cms_group_id bigint,
    final_gender_code integer,
    age_level integer,
    pvalue_level integer,
    shopping_level integer,
    occupation integer,
    new_user_class_level integer
    ,PRIMARY KEY (user_id)
) WITH (
orientation = 'column',
storage_format = 'orc',
distribution_key = 'user_id'
);
END;
--導入數(shù)據(jù)
INSERT INTO dim_user_profile SELECT 
    try_cast(userid as bigint),
    try_cast(cms_segid as bigint),
    try_cast(cms_group_id as bigint),
    try_cast(final_gender_code as integer),
    try_cast(age_level as integer),
    try_cast(pvalue_level as integer),
    try_cast(shopping_level as integer),
    try_cast(occupation as integer),
    try_cast(new_user_class_level as integer)
FROM foreign_holo.user_profile WHERE userid != 'userid';

場景1：累計增量成交分析

在實際應(yīng)用中，某些業(yè)務(wù)場景并不需要實時分析，例如：業(yè)務(wù)需要累積分析每小時的行為數(shù)據(jù)、成交數(shù)據(jù)等，允許數(shù)據(jù)有一定的延遲（近實時分析場景）。可以通過Dynamic Table的增量刷新，對新增數(shù)據(jù)的進行計算，減少數(shù)據(jù)計算量，同時也能提升數(shù)據(jù)加工的時效性，滿足近實時數(shù)據(jù)分析需求。

創(chuàng)建Dynamic Table

1. 登錄DataWorks控制臺，選擇數(shù)據(jù)開發(fā)與治理 > 數(shù)據(jù)開發(fā)。

2. 在數(shù)據(jù)開發(fā)（新版）頁面，選擇目標工作空間，單擊進入數(shù)據(jù)開發(fā)（新版）。

3. 單擊左側(cè)，在數(shù)據(jù)目錄頁面，選擇Hologres。

4. 單擊添加實例。

5. 在DataWorks數(shù)據(jù)源頁簽，單擊已綁定目標數(shù)據(jù)源操作列的添加為數(shù)據(jù)目錄。

6. 選擇目標數(shù)據(jù)庫 > 目標模式 > 動態(tài)表 > +。

7. 填寫表名：例如commerce_behavior_summary_incrmental。若您自定義其他表名，下方查詢SQL一并進行修改。

8. 在SQL開發(fā)界面填寫SQL如下。

   SELECT 
     behavior_log.ds, 
     substring(behavior_log.time, 12, 2) as hour, 
     behavior_log.behavior_type, 
     COUNT(*) AS total_behavior_count, 
     COUNT(DISTINCT behavior_log.user_id) AS uv_count 
   FROM 
     behavior_log 
     JOIN dim_user_profile FOR SYSTEM_TIME AS OF PROCTIME() AS user_profile ON behavior_log.user_id = user_profile.user_id 
   WHERE
     user_profile.shopping_level > 1
   GROUP BY 
     behavior_log.ds, 
     hour, 
     behavior_log.behavior_type;

9. 單擊預編譯，對語法正確性進行校驗。

10. 右側(cè)選擇刷新策略，完成參數(shù)配置。

    參數(shù)名稱	說明
    刷新模式	選擇增量刷新（Incremental）。
    數(shù)據(jù)刷新開始時間	選擇建表成功立即開始。
    數(shù)據(jù)刷新周期間隔	1分鐘。
    Hologres 計算資源	選擇Serverless 資源。
    Hologres 計算資源規(guī)格	32Core。

    

11. 填寫完成后，提交發(fā)布，發(fā)布成功后代表已經(jīng)創(chuàng)建成功，增量刷新任務(wù)將會根據(jù)設(shè)置的刷新間隔周期性刷新Dynamic Table，實現(xiàn)數(shù)據(jù)的增量計算。

查詢數(shù)據(jù)

增量Dynamic Table的刷新任務(wù)執(zhí)行后，可以開始查詢數(shù)據(jù)的增量變化，示例查詢SQL如下。

• 每小時瀏覽量。

  --每小時瀏覽量
  SELECT hour, total_behavior_count, uv_count FROM commerce_behavior_summary_incrmental WHERE ds = '2017-05-13' AND behavior_type = 'pv' ORDER BY hour;

• 每小時購買量。

  --每小時購買量
  SELECT hour, total_behavior_count, uv_count FROM commerce_behavior_summary_incrmental WHERE ds = '2017-05-13' AND behavior_type = 'buy' ORDER BY hour;

增量大屏展示

當增量刷新任務(wù)啟動成功后，可以直接使用Dynamic Table對接應(yīng)用查詢數(shù)據(jù)，本實踐直接對接DataV大屏實時展示數(shù)據(jù)。操作步驟如下。

1. 創(chuàng)建Hologres數(shù)據(jù)源（可選）。

   將數(shù)據(jù)所在的Hologres實例和數(shù)據(jù)庫創(chuàng)建為DataV的數(shù)據(jù)源，詳情請參見DataV。

2. 創(chuàng)建可視化應(yīng)用。

   1. 登錄DataV控制臺。

   2. 在工作臺頁面，單擊創(chuàng)建PC端看板。

   3. 選擇已有模板或自定義模板，完成可視化數(shù)據(jù)展示。

示例展示每小時累計行為數(shù)據(jù)，使用DataV的柱狀圖可以直觀地展示增量數(shù)據(jù)的變化。

場景2：歷史廣告數(shù)據(jù)分析與數(shù)據(jù)回刷

業(yè)務(wù)通常會有歷史數(shù)據(jù)關(guān)聯(lián)分析的場景，可以通過Dynamic Table的全量刷新（full）模式支持這種場景。批量刷新將會一次性刷新全量的數(shù)據(jù)，通過Dynamic Table就能完成歷史數(shù)據(jù)的加工與計算，然后快速分析。本次實踐通過全量刷新模式來計算過去7天廣告點擊排行榜Top10，通過DataWorks頁面可視化創(chuàng)建Dynamic Table全量刷新模式。

創(chuàng)建Dynamic Table

1. 登錄DataWorks控制臺，選擇數(shù)據(jù)開發(fā)與治理 > 數(shù)據(jù)開發(fā) > 進入數(shù)據(jù)開發(fā)（新版）。

2. 選擇目標工作空間，單擊左側(cè)圖標，進入數(shù)據(jù)目錄頁面。

3. 選擇HOLOGRES，單擊選擇圖標添加實例。

4. 在DataWorks數(shù)據(jù)源頁簽中單擊已綁定的目標數(shù)據(jù)源操作列的添加為數(shù)據(jù)目錄，完成綁定操作。

5. 選擇目標數(shù)據(jù)庫 > 目標模式 > 動態(tài)表 > +。

6. 填寫表名：例如commerce_top_clicks_cate_full。若您自定義其他表名，下方查詢SQL一并進行修改。

7. 在SQL開發(fā)界面填寫SQL如下。

   SELECT 
     raw_sample.ds, 
     ad_feature.cate_id, 
     COUNT(*) AS total_browses, 
     COUNT(DISTINCT raw_sample.user_id) AS total_unique_visitors, 
     SUM(
       CASE WHEN raw_sample.clk = '1' THEN 1 ELSE 0 END
     ) AS total_clicks 
   FROM 
     raw_sample raw_sample 
     JOIN dim_ad_feature FOR SYSTEM_TIME AS OF PROCTIME() AS ad_feature ON raw_sample.adgroup_id = ad_feature.adgroup_id 
   GROUP BY 
     raw_sample.ds, 
     ad_feature.cate_id;

8. 單擊預編譯，對語法正確性進行校驗。

9. 右側(cè)選擇刷新策略，完成參數(shù)配置。

   參數(shù)名稱	說明
   刷新模式	選擇全量刷新（Full）。
   數(shù)據(jù)刷新開始時間	選擇建表成功立即開始。
   數(shù)據(jù)刷新周期間隔	24小時。
   Hologres 計算資源	選擇Serverless 資源。
   Hologres 計算資源規(guī)格	32Core。

   

10. 填寫完成后，提交發(fā)布，發(fā)布成功后代表已經(jīng)創(chuàng)建成功，全量刷新任務(wù)將會根據(jù)設(shè)置的刷新間隔周期性刷新Dynamic Table，實現(xiàn)數(shù)據(jù)的全量計算。

查詢數(shù)據(jù)

全量刷新的Dynamic Table的刷新任務(wù)執(zhí)行完成后，代表歷史數(shù)據(jù)全量計算完成，可以直接查詢Dynamic Table進行數(shù)據(jù)分析。示例SQL如下。

--過去7天廣告點擊排行
SELECT 'cate_id_' || cate_id, sum(total_clicks) AS sum_total_clicks FROM commerce_top_clicks_cate_full WHERE ds <= '2017-05-13' AND ds >= TO_CHAR(ds::date - INTERVAL '6 days', 'YYYY-MM-DD') GROUP BY cate_id ORDER BY sum_total_clicks DESC limit 10;

全量大屏展示

可以直接將Dynamic Table對接DataV大屏展示數(shù)據(jù)。本實踐已經(jīng)提前準備好了DataV大屏，并連接好了數(shù)據(jù)源和配置大屏，能直接展示數(shù)據(jù)。

歷史數(shù)據(jù)回刷

在實際場景中，歷史數(shù)據(jù)可能會變化，為了保證數(shù)據(jù)口徑的一致性，需要對歷史數(shù)據(jù)刷新。可以通過Dynamic Table全量刷新的模式來完成歷史數(shù)據(jù)的回刷，減少任務(wù)的維護，提升回刷的效率。本實踐演示回刷場景，在DataWorks新建Hologres SQL節(jié)點，執(zhí)行如下SQL。

1. 模擬上游維表更新。

   模擬業(yè)務(wù)中上游維表更新的場景，在DataWorks界面執(zhí)行如下SQL對維表數(shù)據(jù)進行更新。

   1. 查詢出數(shù)據(jù)。

      SELECT * FROM raw_sample
          JOIN dim_ad_feature FOR SYSTEM_TIME AS OF PROCTIME () AS ad_feature ON raw_sample.adgroup_id = ad_feature.adgroup_id
      WHERE ad_feature.cate_id = '6261';

   2. 模擬維表更新，將維表按照主鍵更新。

      INSERT INTO dim_ad_feature  (adgroup_id,cate_id) VALUES ('451376','888888'),('650847','000000')
      ON CONFLICT (adgroup_id)
      DO UPDATE SET cate_id = EXCLUDED.cate_id
      WHERE dim_ad_feature.cate_id= '6261';

2. 回刷Dynamic Table。

   當維表更新完成后，需要對Dynamic Table執(zhí)行一次刷新，匹配新的維表數(shù)據(jù)，從而保證數(shù)據(jù)口徑的一致性。直接對Dynamic Table執(zhí)行Refresh，即可完成回刷。

   REFRESH TABLE commerce_top_clicks_cate_full;

回刷完成后，查看DataV大屏，全量數(shù)據(jù)面板中數(shù)據(jù)已經(jīng)更新。

附錄：Dynamic Table

增量刷新Dynamic Table

--累計增量成交分析，增量刷新
DROP TABLE IF EXISTS commerce_behavior_summary_incrmental;
CREATE DYNAMIC TABLE commerce_behavior_summary_incrmental WITH (
    auto_refresh_enable='true',
    refresh_mode = incremental, 
    incremental_auto_refresh_schd_start_time='immediate',
    incremental_auto_refresh_interval = '1 minutes',
    incremental_guc_hg_computing_resource = 'serverless',
    incremental_guc_hg_experimental_serverless_computing_required_cores='32'
)
 AS 
SELECT 
  behavior_log.ds, 
  substring(behavior_log.time, 12, 2) as hour, 
  behavior_log.behavior_type, 
  COUNT(*) AS total_behavior_count, 
  COUNT(DISTINCT behavior_log.user_id) AS uv_count 
FROM 
  behavior_log 
  JOIN dim_user_profile FOR SYSTEM_TIME AS OF PROCTIME() AS user_profile ON behavior_log.user_id = user_profile.user_id 
WHERE
  user_profile.shopping_level > 1
GROUP BY 
  behavior_log.ds, 
  hour, 
  behavior_log.behavior_type;

全量刷新Dynamic Table

--歷史廣告數(shù)據(jù)分析，全量刷新
DROP TABLE IF EXISTS commerce_top_clicks_cate_full;
CREATE DYNAMIC TABLE commerce_top_clicks_cate_full    WITH (
    auto_refresh_enable='true',
    refresh_mode = full, 
    full_auto_refresh_schd_start_time='immediate',
    full_auto_refresh_interval = '24 hours',
    full_guc_hg_computing_resource = 'serverless'
)
 AS 
SELECT 
  raw_sample.ds, 
  ad_feature.cate_id, 
  COUNT(*) AS total_browses, 
  COUNT(DISTINCT raw_sample.user_id) AS total_unique_visitors, 
  SUM(
    CASE WHEN raw_sample.clk = '1' THEN 1 ELSE 0 END
  ) AS total_clicks 
FROM 
  raw_sample 
  JOIN dim_ad_feature FOR SYSTEM_TIME AS OF PROCTIME() AS ad_feature ON raw_sample.adgroup_id = ad_feature.adgroup_id 
GROUP BY 
  raw_sample.ds, 
  ad_feature.cate_id;