Partial Update（寬表拼接） - Realtime Compute for Apache Flink

本文將為您介紹Partial Update的使用。

為什麼需要Partial Update實現大寬表拼接

在傳統流式資料管道中，為構建寬表常需在主鍵上 Join 多個流或表。例如，在即時推薦系統中，為了獲得使用者 360° 視圖，往往需要整合來自偏好、購買、點擊、購物車等多個不同資料來源的更新。

雖然 Apache Flink 可實現多流 Join，但在大規模情境下存在多個痛點：

狀態規模龐大：需緩衝所有事件直到完成 Join，狀態可能長期甚至無限保留；
效能瓶頸：檢查點開銷大、易引發反壓，影響整體吞吐和穩定性；
調試困難：狀態複雜且不可見，問題排查成本高；
資料一致性差：TTL 設定可能導致事件提前被丟棄，造成結果錯誤。

因此，我們迫切的需要一種新的資料打寬方法，避免複雜 Join，並顯著降低資源消耗和維護複雜度。

基於Fluss Partial Update的資料打寬新模式

Fluss 引入基於Partial Update（部分更新）機制的資料打寬方法，支援在流處理作業中直接擴充維度資訊，無需依賴複雜的多流 Join。

與傳統方式不同，Fluss允許各資料來源根據主鍵獨立更新自己相關的欄位到同一張寬表中。例如，你可以定義一張以user_id為主鍵的user_profile表，包含所有可能欄位。每個資料流只需寫入自己知道的部分欄位，Fluss 儲存引擎會自動按主鍵合并這些更新。

其底層機制是：當某條部分更新到達時，Fluss 會尋找該主鍵已有的記錄，並僅更新本次提供的欄位，其餘欄位保持不變。合并後的新版本即時落盤，確保每條記錄始終是最新的完整狀態，其底層機制類似於資料庫中的“累加式更新”，每次只更新有變化的欄位，而非整條記錄。這種方式大幅降低了 Flink 作業的狀態壓力和計算複雜度，使資料管道更輕量、高效、易維護。

實踐樣本

前提條件

管理 Fluss Catalog。
管理資料庫。

步驟一：建立表

已建立名為fluss-demo的 Fluss Catalog。

登入Realtime Compute控制台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊資料開發 > 資料查詢 > 建立臨時查詢指令碼

在預設的fluss資料庫中建立三張表代表構建寬表的不同資料來源，及一張寬表作為結果表

-- Recommendations – model scores
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.recommendations (
    user_id  STRING,
    item_id  STRING,
    rec_score DOUBLE,
    rec_ts   TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Impressions – how often we showed something
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.impressions (
    user_id STRING,
    item_id STRING,
    imp_cnt INT,
    imp_ts  TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Clicks – user engagement
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.clicks (
    user_id  STRING,
    item_id  STRING,
    click_cnt INT,
    clk_ts    TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Result wide table
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,   -- updated by recs stream
    imp_cnt   INT,      -- updated by impressions stream
    click_cnt INT,      -- updated by clicks stream
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

單擊運行完成樣本表建立。

步驟二：建立樣本流作業

在左側導覽列，單擊資料開發 > ETL > 建立 > 建立流作業

將recommendations、impressions及clicks三張以user_id為共用主鍵的表資料分別插入以user_id為主鍵的大寬表user_rec_wide中。

BEGIN STATEMENT SET
;
INSERT INTO `fluss-demo`.fluss.user_rec_wide (user_id, item_id, rec_score)
SELECT
    user_id
    ,item_id
    ,rec_score
FROM `fluss-demo`.fluss.recommendations
;
INSERT INTO `fluss-demo`.fluss.user_rec_wide (user_id, item_id, imp_cnt)
SELECT
    user_id
    ,item_id
    ,imp_cnt
FROM `fluss-demo`.fluss.impressions
;
-- Apply click counts
INSERT INTO `fluss-demo`.fluss.user_rec_wide (user_id, item_id, click_cnt)
SELECT
    user_id
    ,item_id
    ,click_cnt
FROM `fluss-demo`.fluss.clicks
;
END
;

單擊右上方單擊部署 > 前往營運。單擊啟動對應作業。

作業啟動後進入狀態總覽頁面，健康分為100，歷史累計重啟次數為0。DAG拓撲顯示源表為 fluss-demo.fluss.recommendations、fluss-demo.fluss.impressions 和 fluss-demo.fluss.clicks，目標寬表為 user_rec_wide。

可以注意到計算拓撲中並不包含多流 Join節點，多張明細表均獨立寫入寬表。

步驟三：插入明細表資料

在 資料查詢 頁面，再次建立一個空白查詢指令碼，輸入以下SQL，向三張明細表分別插入測試資料：

-- Recommendations – model scores
INSERT INTO `fluss-demo`.fluss.recommendations VALUES
    ('user_101','prod_501',0.92 , TIMESTAMP '2025-05-16 09:15:02'),
    ('user_101','prod_502',0.78 , TIMESTAMP '2025-05-16 09:15:05'),
    ('user_102','prod_503',0.83 , TIMESTAMP '2025-05-16 09:16:00'),
    ('user_103','prod_501',0.67 , TIMESTAMP '2025-05-16 09:16:20'),
    ('user_104','prod_504',0.88 , TIMESTAMP '2025-05-16 09:16:45');
-- Impressions – how often each (user,item) was shown
INSERT INTO `fluss-demo`.fluss.impressions VALUES
    ('user_101','prod_501', 3, TIMESTAMP '2025-05-16 09:17:10'),
    ('user_101','prod_502', 1, TIMESTAMP '2025-05-16 09:17:15'),
    ('user_102','prod_503', 7, TIMESTAMP '2025-05-16 09:18:22'),
    ('user_103','prod_501', 4, TIMESTAMP '2025-05-16 09:18:30'),
    ('user_104','prod_504', 2, TIMESTAMP '2025-05-16 09:18:55');
-- Clicks – user engagement
INSERT INTO `fluss-demo`.fluss.clicks VALUES
    ('user_101','prod_501', 1, TIMESTAMP '2025-05-16 09:19:00'),
    ('user_101','prod_502', 2, TIMESTAMP '2025-05-16 09:19:07'),
    ('user_102','prod_503', 1, TIMESTAMP '2025-05-16 09:19:12'),
    ('user_103','prod_501', 1, TIMESTAMP '2025-05-16 09:19:20'),
    ('user_104','prod_504', 1, TIMESTAMP '2025-05-16 09:19:25');

步驟四：驗證大寬表資料寫入結果

在資料開發 > ETL > 建立 > 建立流作業 。

SELECT * FROM `fluss-demo`.fluss.user_rec_wide;

單擊調試按鈕。我們將觀察到三張明細表分寫寫入後，大寬表中匯總的流式計算結果：以user_id為主鍵計算產生了正確的大寬表資料。

查詢 user_rec_wide 寬表返回 5 條記錄，包含 user_id、item_id、rec_score、imp_cnt、click_cnt 五個欄位，驗證三張明細表資料已正確匯總。