Hologres：Portrait analysis - RoaringBitmap最佳化方案

使用RoaringBitmap的方案的可以適用於如下情境。

• 標籤數量多：需要對多張大表進行關聯運算，可以使用BITMAP_AND替換JOIN運算，可以降低記憶體消耗，Bitmap外掛程式庫使用SIMD指令最佳化可以將CPU使用率提升1~2個數量級。

• 資料規模大且需去重運算：如數十億資料需要去重，Bitmap結構天然去重，避免精確UV計算和記憶體的開銷。

在介紹Bitmap計算方案之前，我們需要區分畫像系統中常用的兩類標籤資料，針對這兩類資料的計算模式大相徑庭，我們需要依據資料模型和運算模式將合理的部分轉化為Bitmap格式儲存。

• 屬性標籤：主要描述使用者的屬性情況，例如使用者性別、所在省份、已婚狀態等，資料相對穩定，通常進行精準條件過濾。這類資料進行Bitmap壓縮比會很高，且Bitmap適合對應的運算。

• 行為標籤：主要描述使用者行為特徵，描述使用者在某個時間做了一件什麼事，比如使用者店鋪瀏覽購買行為、使用者登入活躍行為等，資料變更頻率高，通常需要進行範圍掃描，彙總過濾。這類資料不適合進行Bitmap壓縮，壓縮比會很差，運算模式不適合Bitmap直接運算。

屬性標籤處理情境通常是描述使用者的屬性情況，資料相對穩定，通常進行精準條件過濾，通過Bitmap能實現高效的壓縮和運算。

• 方案介紹

  假設現有的DMP（Data Management Platform）系統中存在兩張屬性標籤表，如下圖所示。dws_userbase表描述使用者基礎屬性，dws_usercate_prefer表描述使用者偏好。

  典型的人群人數預估的情境比如計算[province = 北京] & [cate_prefer = 時裝] 的人數，通常可以關聯、過濾、去重最終獲得人數。但在資料規模較大的情境，關聯、去重運算會帶來嚴重的效能負擔。

  Bitmap最佳化方案通過預構建標籤的Bitmap表來減少即席運算的成本，比如將兩張表中的列拆開分別構建Bitmap表如下圖所示，通過兩表中對應Bitmap的與運算計算出人數。rb_dws_userbase_province表描述省份和uid的Bitmap關係，rb_dws_usercate_prefer_cprefer描述類目和uid的Bitmap關係。

  但是按列拆分的方案也存在一定的問題，當多列之間存在層級的組織關係時，上述的拆分和運算方式可能會導致計算錯誤的情況，如下圖所示。在描述店鋪使用者新客、老客、潛客資訊的dws_shop_cust表中，按照列拆分成描述店鋪名稱的Bitmap表rb_dws_shop_cust_shop_id，描述客戶類型的Bitmap表rb_dws_shop_cust_cust_type。當計算[shop_id = A] & [cust_type = 新客] 的人群集合，會得到[1]的uid集合，然而在真實的表中這條資料不存在。產生這種錯誤的原因是cust_type、shop_id兩個欄位存在一定的關聯性，在數倉模型中cust_type是shop_id維度指標資料，脫離統計維度單獨使用指標是錯誤的。因此可以將維度shop_id和指標cust_type組合值作為構建Bitmap的單元，產生rb_dws_shop_cust_sid_ctype表來避免這類錯誤。

  根據以上方案介紹，需要將uid壓縮儲存進Bitmap，通過Bitmap與或非運算實現標籤的對應運算。

• 方案實踐

  • 使用者資訊編碼處理

    使用者標識可能是字元類型的，由於Bitmap只能儲存整數資訊，因此需要先將uid進行整數編碼進Hologres中，可以通過插入包含自增序列Serial的表完成字元ID的編碼。

    -- 建立字典表
    CREATE TABLE dws_uid_dict (
        encode_uid bigserial,
          uid text primary key
    );
    
    -- 錄入標籤表中的uid
    INSERT INTO dws_uid_dict(uid)
    SELECT uid
    FROM dws_userbase ON conflict DO NOTHING;

    對使用者標識進行編碼不僅可以方便儲存進Bitmap，還可以使ID資訊保持連續性。如下圖所示bitmap2由於儲存的ID資料稀疏，儲存效率相比bitmap1低很多，因此對ID編碼處理，也會降低儲存成本，提升運算效率。

    對於數實值型別，如果較為稀疏也可以考慮編碼，但是進行編碼有利有弊，例如在廣告DMP系統中除了需要高效能的畫像能力，也需要即時輸出人群明細的能力；而一旦需要輸出人群明細，就需要Join使用者ID表進行還原，這一部分的效能開銷，也應當納入技術選型的考慮因素中。因此您可以依據使用情境權衡選擇，是否進行編碼建議如下。

    • 字元ID：建議編碼。

    • 整數ID且需要頻繁還原原始值：建議不編碼。

    • 整數ID且不需要還原原始值：建議編碼。

  • Bitmap 加工和查詢

    依據上述方案按列拆分的思路，將dws_userbase表與dws_shop_cust表進行拆分，按照分別為省份、性別的列Bitmap拆分為一個表，但是性別只有男、女兩個選項，壓縮出來的Bitmap只能分佈於叢集中的兩個節點，計算儲存都很不平均，叢集的資源並不能充分利用。因此有必要將Bitmap拆分成多段，並將它們打散到叢集中來提升並發執行的能力，假設將Bitmap打散成65536段，SQL命令如下。

    -- dws_userbase 結構見寬表方案
    BEGIN;
    CREATE TABLE dws_shop_cust
    (
      uid text not null primary key,
      shop_id text,
        cust_type text
    );
    call set_table_property('dws_shop_cust', 'distribution_key', 'uid');
    END;
    
    -- 建立bitmap外掛程式
    CREATE EXTENSION roaringbitmap;
    
    BEGIN;
    CREATE TABLE rb_dws_userbase_province (
        province text,
      bucket int,
        bitmap roaringbitmap
    );
    call set_table_property('rb_dws_userbase_province', 'distribution_key', 'bucket');
    END;
    
    BEGIN;
    CREATE TABLE rb_dws_shop_cust_sid_ctype (
        shop_id text,
      cust_type text,
      bucket int,
        bitmap roaringbitmap
    );
    call set_table_property('rb_dws_shop_cust_sid_ctype', 'distribution_key', 'bucket');
    END;
    
    -- 寫入bitmap表
    INSERT INTO rb_dws_userbase_province
    SELECT province,
           encode_uid / 65536 as "bucket",
           rb_build_agg(b.encode_uid) AS bitmap
    FROM dws_userbase a join dws_uid_dict b on a.uid = b.uid
    GROUP BY province, "bucket";
    
    INSERT INTO rb_dws_shop_cust_sid_ctype
    SELECT shop_id,
           cust_type,
           encode_uid / 65536 AS "bucket",
           rb_build_agg(b.encode_uid) AS bitmap
    FROM dws_shop_cust a
    JOIN dws_uid_dict b ON a.uid = b.uid
    GROUP BY shop_id, cust_type, "bucket";

    當進行[shop_id = A] & [cust_type = 新客] & [province = 北京]的客群人數預估時，按照標籤的與或非邏輯組織對應的Bitmap關係運算，即可完成對應的查詢，SQL命令如下。

    SELECT SUM(RB_CARDINALITY(rb_and(ub.bitmap, uc.bitmap)))
    FROM
      (SELECT rb_or_agg(bitmap) AS bitmap,
              bucket
       FROM rb_dws_userbase_province
       WHERE province = '北京'
       GROUP BY bucket) ub
    JOIN
      (SELECT rb_or_agg(bitmap) AS bitmap,
              bucket
       FROM rb_dws_shop_cust_sid_ctype
       WHERE shop_id = 'A'
         AND cust_type = '新客'
       GROUP BY bucket) uc ON ub.bucket = uc.bucket;

當我們計算[province=北京] & [shop_id=A且7天未購買]的使用者時，SQL命令如下所示。

-- 原始行為表
BEGIN;
CREATE TABLE dws_usershop_behavior
(
  uid int not null,
  shop_id text not null,
  pv_cnt int,
  trd_cnt int,
  ds integer not null
);
call set_table_property('dws_usershop_behavior', 'distribution_key', 'uid');
COMMIT;

-- 編碼行為表
BEGIN;
CREATE TABLE dws_usershop_behavior_bucket
(
  encode_uid int not null,
  shop_id text not null,
  pv_cnt int,
  trd_cnt int,
  ds int not null,
  bucket int
);
CALL set_table_property('dws_usershop_behavior_bucket', 'orientation', 'column');
call set_table_property('dws_usershop_behavior_bucket', 'distribution_key', 'bucket');
CALL set_table_property('dws_usershop_behavior_bucket', 'clustering_key', 'shop_id,encode_uid');
COMMIT;

-- 寫入分桶的事實資料
INSERT INTO dws_usershop_behavior_bucket
SELECT *,
             encode_uid,
       shop_id,
       pv_cnt,
       trd_cnt,
       encode_uid / 65536
FROM dws_usershop_behavior a JOIN dws_uid_dict b
on a.uid = b.uid;

-- 事實資料和屬性資料聯合運算
SELECT sum(rb_cardinality(bitmap)) AS cnt
FROM
  (SELECT rb_and(ub.bitmap, us.bitmap) AS bitmap,
          ub.bucket
   FROM
     (SELECT rb_or_agg(bitmap) AS bitmap,
             bucket
      FROM rb_dws_userbase_province
      WHERE province = '北京'
      GROUP BY bucket) AS ub
   JOIN
     (SELECT rb_build_agg(uid) AS bitmap,
             bucket
      FROM
        (SELECT uid,
                bucket
         FROM dws_usershop_behavior_bucket
         WHERE shop_id = 'A' AND ds > to_char(current_date-7, 'YYYYMMdd')::int
         GROUP BY uid,
                  bucket HAVING sum(trd_cnt) = 0) tmp
      GROUP BY bucket) us ON ub.bucket = us.bucket) r

為了避免Bitmap資料計算對生產業務的影響，可以選擇在離線完成Bitmap資料的加工，通過Hologres外部表格能力從MaxCompute或者Hive中直接載入資料。離線處理Bitmap的過程與線上流程思路類似，也可以通過編碼、彙總方式產生資料，在MaxCompute中離線構建Bitmap資料樣本如下。

-- 選擇project
USE bitmap_demo; 

-- 建立原始表
CREATE TABLE mc_dws_uid_dict (
encode_uid bigint,
bucket bigint,
uid string
);

CREATE TABLE mc_dws_userbase
(
  uid string,
  province string,
  gender string,
  marriaged string
);

-- 對新增的UID進行編碼
-- 計算新增編碼uid
WITH uids_to_encode AS
  (SELECT DISTINCT(ub.uid),
          CAST(ub.uid / 65336 AS BIGINT) AS bucket
   FROM mc_dws_userbase ub
   LEFT JOIN mc_dws_uid_dict d ON ub.uid = d.uid
   WHERE d.uid IS NULL),
-- 計算每個分桶中待編碼的uid數量，通過sum視窗累加出bucket編碼的起始值
uids_bucket_encode_offset AS
  (SELECT bucket,
          sum(cnt) over (ORDER BY bucket ASC) - cnt AS bucket_offset
   FROM
     (SELECT count(1) AS cnt,
             bucket
      FROM uids_to_encode
      GROUP BY bucket) x),
-- 計算已使用編碼值最大值
dict_used_id_offset AS
  (SELECT max(encode_uid) AS used_id_offset FROM mc_dws_uid_dict)
-- 新增使用者的編碼 = 已使用編碼值最大值 + Bucket編碼起始值 + RowNumber序號
INSERT INTO mc_dws_uid_dict
SELECT
       COALESCE((SELECT used_id_offset FROM dict_used_id_offset),0) + bucket_offset + rn,
       bucket,
       uid
FROM
  (SELECT row_number() OVER (partition BY ub.bucket ORDER BY ub.uid) AS rn,
          ub.bucket,
          bo.bucket_offset,
          uid
   FROM uids_to_encode ub
   JOIN uids_bucket_encode_offset bo ON ub.bucket = bo.bucket) j


-- 建立bitmap相關函數
add jar function_jar_dir/mc-bitmap-functions.jar as mc_bitmap_func.jar -f;
create function mc_rb_cardinality as com.alibaba.hologres.RbCardinalityUDF using mc_bitmap_func.jar;
create function mc_rb_build_agg as com.alibaba.hologres.RbBuildAggUDAF using mc_bitmap_func.jar;

-- 建立bitmap表並寫入
CREATE TABLE mc_rb_dws_userbase_province
(
  province string,
  bucket int,
  bitmap string
);
INSERT INTO mc_rb_dws_userbase_province
SELECT province,
       b.bucket_num,
       mc_rb_build_agg(b.encode_uid) AS bitmap
FROM mc_dws_userbase a join mc_dws_uid_dict b on a.uid = b.uid
GROUP BY province, b.bucket_num;

在Hologres中執行以下命令。

-- 建立MaxCompute外部表格
CREATE TABLE mc_rb_dws_userbase_province (
    province text,
  bucket int,
    bitmap text
) server odps_server options(project_name 'bitmap_demo', table_name 'mc_rb_dws_userbase_province');
-- 將外表中bitmap資料寫到Hologres中
INSERT INTO rb_dws_userbase_province
SELECT province,
       bucket::INT,
       roaringbitmap_text(bitmap, FALSE)
FROM mc_rb_dws_userbase_province;

通過上述步驟，可以將資料載入至Hologres，然後按照標籤圈選條件，組合Bitmap運算加速查詢。

• MaxCompute中計算Bitmap資料可使用mc-bitmap。

• 離線加工使用Bitmap方案更多資訊請參見離線UV計算

在Realtime Compute情境中，可以使用Flink和Hologres組合的方式，基於RoaringBitmap即時對使用者標籤去重，主要思路如下。

1. 將使用者ID字典表作為維表，利用Hologres的Insert on Conflict機制處理新增編碼，在Flink中進行維表Join。

2. 將Join結果流按照標籤維度進行RoaringBitmap彙總。

3. 輸出的Bitmap結果寫入Hologres對應的Bitmap表中。