使用Paimon的RoaringBitmap彙總函式（rbm32/rbm64）實現精確去重，適用於使用者訪問去重、裝置指紋統計等情境。 - E-MapReduce

Apache Paimon的主鍵表支援Aggregation Merge Engine，可在寫入時為非主鍵欄位指定彙總函式。對於精確去重情境，Paimon提供了rbm32和rbm64兩種彙總函式，底層基於RoaringBitmap實現。本文介紹如何通過自訂UDF配合Paimon的bitmap彙總功能，在Spark SQL中實現精確去重。

背景資訊

Paimon的Aggregation Merge Engine可以在寫入時為非主鍵欄位指定彙總函式（如sum、max、min等），Paimon會在compaction和讀取時自動對相同主鍵的資料進行彙總合并。

Paimon提供了rbm32和rbm64兩種彙總函式，底層基於RoaringBitmap實現。寫入時只需將整數值序列化為bitmap位元組寫入BINARY列，Paimon引擎會在compaction時自動對相同主鍵的bitmap執行OR合并。由於Paimon未內建將整數轉為bitmap位元組的UDF，需要自行實現。

本文提供了一組樣本UDF，示範如何在Spark SQL中配合Paimon的bitmap彙總功能實現精確去重。

前提條件

已建立EMR Serverless Spark工作空間。
已下載Paimon Bitmap UDF的JAR檔案（paimon-spark-bitmap-1.0-SNAPSHOT.jar）並上傳至OSS。
已建立SQL會話並啟動。
已建立SparkSQL類型的任務。

UDF說明

本文提供4個統一的UDF，自動相容32位和64位bitmap。

UDF	功能	32/64位行為
to_bitmap	將整數轉為bitmap序列化位元組。	輸入INT時產出32位格式（配合rbm32），輸入BIGINT時產出64位格式（配合rbm64）。
bitmap_count	擷取bitmap基數（去重後的元素數量）。	自動識別格式。
bitmap_contains	判斷指定值是否存在於bitmap中。	自動識別格式。
bitmap_to_string	將bitmap轉為逗號分隔的字串（調試用）。	自動識別格式。

rbm32樣本：使用者頁面訪問去重

以下樣本示範如何使用rbm32彙總函式，實現使用者頁面訪問的精確去重統計。rbm32適用於ID值在INT（32位整數）範圍內的情境。

步驟一：註冊UDF

註冊bitmap相關的UDF。本文使用臨時UDF進行說明，您可以根據需要註冊臨時或永久UDF。

說明

請將樣本中的oss://bucket/path/to/paimon-spark-bitmap-1.0-SNAPSHOT.jar替換為JAR檔案的實際OSS路徑。

CREATE TEMPORARY FUNCTION to_bitmap AS 'org.apache.paimon.spark.bitmap.ToBitmapUDF' USING JAR 'oss://bucket/path/to/paimon-spark-bitmap-1.0-SNAPSHOT.jar';
CREATE TEMPORARY FUNCTION bitmap_count AS 'org.apache.paimon.spark.bitmap.BitmapCountUDF' USING JAR 'oss://bucket/path/to/paimon-spark-bitmap-1.0-SNAPSHOT.jar';
CREATE TEMPORARY FUNCTION bitmap_contains AS 'org.apache.paimon.spark.bitmap.BitmapContainsUDF' USING JAR 'oss://bucket/path/to/paimon-spark-bitmap-1.0-SNAPSHOT.jar';
CREATE TEMPORARY FUNCTION bitmap_to_string AS 'org.apache.paimon.spark.bitmap.BitmapToStringUDF' USING JAR 'oss://bucket/path/to/paimon-spark-bitmap-1.0-SNAPSHOT.jar';

步驟二：建立Paimon彙總表

建立使用rbm32彙總函式的Paimon主鍵表。

CREATE TABLE user_page_visits (
    user_id INT,
    page_visits BINARY
) USING paimon
TBLPROPERTIES (
    'primary-key' = 'user_id',
    'merge-engine' = 'aggregation',
    'fields.page_visits.aggregate-function' = 'rbm32'
);

關鍵參數說明：

'primary-key' = 'user_id'：指定user_id為主鍵。
'merge-engine' = 'aggregation'：使用彙總合并引擎。
'fields.page_visits.aggregate-function' = 'rbm32'：對page_visits欄位使用rbm32彙總函式，適用於INT類型的值。

步驟三：寫入資料

使用to_bitmap將頁面ID轉為bitmap位元組後寫入。

-- 初始寫入
INSERT INTO user_page_visits VALUES
    (1, to_bitmap(100, 101, 102)),   -- 使用者1訪問了頁面100, 101, 102
    (2, to_bitmap(101, 103)),        -- 使用者2訪問了頁面101, 103
    (3, to_bitmap(102, 104));        -- 使用者3訪問了頁面102, 104

-- 追加寫入（Paimon自動對相同user_id的bitmap做OR合并）
INSERT INTO user_page_visits VALUES
    (1, to_bitmap(103, 105)),        -- 使用者1又訪問了103, 105
    (2, to_bitmap(104, 106));        -- 使用者2又訪問了104, 106

步驟四：查詢去重結果

查詢每個使用者訪問了多少個不同頁面。

SELECT user_id, bitmap_count(page_visits) AS unique_pages
FROM user_page_visits;

預期結果如下。

user_id	unique_pages	說明
1	5	100, 101, 102, 103, 105自動合并。
2	4	101, 103, 104, 106自動合并。
3	2	102, 104。

您還可以使用以下查詢進行更多操作。

-- 查詢使用者是否訪問過某個頁面
SELECT user_id, bitmap_contains(page_visits, 101) AS visited_page_101
FROM user_page_visits;

-- 查看bitmap中的具體值（調試用）
SELECT user_id, bitmap_to_string(page_visits) AS pages
FROM user_page_visits;

rbm64樣本：大ID情境（如裝置指紋）

當ID值超過INT範圍時（如裝置指紋ID），需要使用rbm64彙總函式，底層基於64位RoaringBitmap實現。rbm64與rbm32使用同一套UDF，無需額外註冊。

說明

如果在同一個會話中已經註冊過上述4個UDF，此處無需重複註冊。

步驟一：建立表

建立使用rbm64彙總函式的Paimon主鍵表。裝置ID超過INT範圍，因此使用rbm64。

CREATE TABLE app_device_visits (
    app_id INT,
    device_bitmap BINARY
) USING paimon
TBLPROPERTIES (
    'primary-key' = 'app_id',
    'merge-engine' = 'aggregation',
    'fields.device_bitmap.aggregate-function' = 'rbm64'
);

步驟二：寫入資料

to_bitmap輸入BIGINT時自動產出64位格式。

-- 初始寫入
INSERT INTO app_device_visits VALUES
    (1001, to_bitmap(CAST(8000000001 AS BIGINT), CAST(8000000002 AS BIGINT), CAST(8000000003 AS BIGINT))),
    (1002, to_bitmap(CAST(8000000002 AS BIGINT), CAST(8000000004 AS BIGINT)));

-- 追加寫入（自動合并）
INSERT INTO app_device_visits VALUES
    (1001, to_bitmap(CAST(8000000004 AS BIGINT), CAST(8000000006 AS BIGINT)));

步驟三：查詢去重結果

SELECT app_id,
       bitmap_count(device_bitmap) AS unique_devices,
       bitmap_contains(device_bitmap, CAST(8000000002 AS BIGINT)) AS has_device
FROM app_device_visits;