Realtime Compute for Apache Flink：Versioned Merge Engine

Versioned Merge Engine說明

• 合并策略

  基於版本號碼或事件時間戳記進行合并，保留每個主鍵對應的最高版本或最新時間戳記的記錄。

• 配置方式

  在建立表時，通過設定表屬性'table.merge-engine' = 'versioned'來啟用。

• 更新機制

  • 僅當新傳入記錄的指定欄位值（如版本號碼）大於或等於當前儲存值時，才會執行更新。

  • 若新值小於現有值或為NULL，則不觸發更新 。

• 適用情境

  • 適用於資料去重和亂序資料合併，作為流式計算中“去重轉換”的有效替代方案。

  • 可確保與上遊資料的最終一致性，即在一定時間視窗內，系統能自動修複資料差異，達到一致狀態。

• 使用限制

  • 不支援UPDATE和DELETE操作。

  • 不支援部分更新。

  • 變更日誌中的UPDATE_BEFORE和DELETE事件將被自動忽略。

使用樣本

版本列

版本列是表中儲存資料記錄版本號碼（或事件時間戳記）的列，用於標識每條記錄的唯一性，是啟用Versioned Merge Engine的必要條件。

'table.merge-engine' = 'versioned',
'table.merge-engine.versioned.ver-column' = '<列名>'

版本列支援的資料類型：

• INT

• BIGINT

• TIMESTAMP

• TIMESTAMP(p)（帶精度）

• TIMESTAMP_LTZ（本地時區時間戳記）

• TIMESTAMP_LTZ(p)（帶精度的本地時區時間戳記）

SQL樣本

CREATE TABLE VERSIONED (
    a INT NOT NULL PRIMARY KEY NOT ENFORCED,
    b STRING, 
    ts BIGINT
 ) WITH (
    'table.merge-engine' = 'versioned',
    'table.merge-engine.versioned.ver-column' = 'ts'
);

INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v1', 1000);

-- 插入ts < 1000的資料，不會更新
INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v2', 999);
SELECT * FROM VERSIONED WHERE a = 1;
-- 輸出
-- +---+-----+------+
-- | a | b   | ts   |
-- +---+-----+------+
-- | 1 | v1  | 1000 |
-- +---+-----+------+


-- 插入ts > 1000的資料，會更新
INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v3', 2000);
SELECT * FROM VERSIONED WHERE a = 1;
-- 輸出
-- +---+-----+------+
-- | a | b   | ts   |
-- +---+-----+------+
-- | 1 | v3  | 2000 |
-- +---+-----+------+

-- 插入ts為null的資料，不會更新
INSERT INTO VERSIONED (a, b, ts) VALUES (1, 'v4', CAST(null as BIGINT));
SELECT * FROM VERSIONED WHERE a = 1;
-- 輸出
-- +---+-----+------+
-- | a | b   | ts   |
-- +---+-----+------+
-- | 1 | v3  | 2000 |
-- +---+-----+------+