Delta Join（多流Join新範式） - Realtime Compute for Apache Flink

本文將為您介紹Delta Join的用法與實現。

Delta Join：基於Fluss的雙流Join新方案

在即時數倉情境下，往往需要依賴多張即時資料表來構建統一的寬表。基於開源Flink + Kafka搭建的即時數倉，只能使用多個Kafka的Topic通過Flink Join拼接形成一張大寬表，來實現無論哪個Topic發生更新時，總能對整個寬表完成更新。

由於Kafka本身並非面向分析情境設計，因此只能依託開源Flink的流式Join來實現這個情境，因為需要雙邊驅動更新，並在緩衝全量上遊資料，會導致開源Flink狀態體積龐大，帶來資源成本高、營運複雜、效率低等問題。

Fluss對比Kafka實現了Delta Join的能力。這是一種新型的 Join 方案，能夠在保持雙流 Join 語義的同時，將資料更新行為下沉到Fluss表中完成，從而顯著降低Flink的資源消耗，提升作業的穩定性和執行效率。

Delta Join的優勢

無 Join State：省去冗餘資料存放區。
低成本：僅依賴 Fluss 主鍵表和二級索引。
更穩定高效 ：避免大狀態帶來的效能瓶頸。

Delta Join使用限制

左右表必須為 Fluss 的主鍵表（支援分區表）。
Fluss 主鍵表的分桶鍵（Bucket Key）需要為主鍵（Primary Key）首碼。
Delta Join 的 Join Key 必須包含 Fluss 主鍵表定義的分桶鍵（Bucket Key），如果是分區表，則 Join Key 還需要包含分區鍵（Partition Key）。
說明
例如定義：PRIMARY KEY (user_id, order_id, order_data)，且 'bucket.key' = 'user_id'。
- Bucket key是主鍵的首碼，因此現在使用者可以把 user_id 當作二級索引進行高效的資料尋找。
- 對於查詢：JOIN users u ON o.user_id = u.user_id；
  此時Join Key為user_id，包含了定義的分桶鍵，因此該查詢會被最佳化成 Delta Join。

使用樣本

某電商平台需要把訂單流和商品流關聯成寬表，供下遊查詢。

步驟一：建立源表和結果表

兩側源表都要讓 Join Key 命中索引。本例的 Join Key 是 (merchant_id, item_id)：在 orders 表中它是主鍵首碼，需要聲明為 Bucket Key，作為可用的首碼索引；在 items 表中它就是主鍵，不需要額外配置。

CREATE TABLE `my-catalog`.`my_db`.`orders` (
  merchant_id BIGINT,                    -- 商家 ID
  item_id BIGINT,                        -- 商品 ID
  order_id BIGINT,                       -- 訂單 ID
  amount DECIMAL(18, 2),                 -- 訂單當前金額
  PRIMARY KEY (merchant_id, item_id, order_id) NOT ENFORCED
) WITH (
  'bucket.key' = 'merchant_id,item_id'
);

CREATE TABLE `my-catalog`.`my_db`.`items` (
  merchant_id BIGINT,                    -- 商家 ID
  item_id BIGINT,                        -- 商品 ID
  item_name STRING,                      -- 商品名稱
  PRIMARY KEY (merchant_id, item_id) NOT ENFORCED   -- Join Key 與主鍵一致
);

CREATE TABLE `my-catalog`.`my_db`.`order_item_wide` (
  merchant_id BIGINT,
  order_id BIGINT,
  item_id BIGINT,
  item_name STRING,
  amount DECIMAL(18, 2),
  PRIMARY KEY (merchant_id, order_id, item_id) NOT ENFORCED
);

步驟二：編寫 Join 作業

USE CATALOG `my-catalog`;
USE `my_db`;

-- 啟用 Delta Join 改寫
SET 'table.optimizer.delta-join.strategy' = 'EVENTUAL';

-- Delta Join 語句本身不需要引入專用文法
INSERT INTO order_item_wide
SELECT
  o.merchant_id,
  o.order_id,
  o.item_id,
  i.item_name,
  o.amount
FROM orders AS o
JOIN items AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;

步驟三：驗證最佳化生效

作業部署上線、啟動後，在“狀態總覽”介面的作業拓撲圖中，如果看到如下的 Delta Join 節點，則表明雙流 Join 已經成功被改寫為 Delta Join。

相關參數

Flink 作業參數

參數名	預設值	說明	調整建議
`table.optimizer.delta-join.strategy`	`NONE`	是否改寫為 Delta Join。 `NONE`：關閉。 `AUTO`：嘗試改寫，失敗時自動回退為雙流 Join。 `EVENTUAL`：強制改寫，失敗時報錯。說明在Realtime Compute引擎 VVR 11.5 前的版本中，`EVENTUAL`需要替換為`FORCE`	調試和生產都推薦 `EVENTUAL`，改寫失敗會直接報出原因，也能避免作業變更後靜默退化成雙流 Join。
`table.optimizer.delta-join.ignore-non-unique-key-filter`	`false`	是否跳過 Join 結果上非唯一鍵過濾條件的校正。	僅在業務能容忍結果表殘留舊資料時設定為 `true`。該配置只跳過校正，不會改變過濾條件，也不會補發撤回訊息。
`table.exec.delta-join.cache-enabled`	`false`	是否開啟本地記憶體緩衝，快取命中時不再請求 Fluss。	記憶體壓力不大時推薦設定為 `true`，可以明顯減少對 Fluss 的點查請求。
`table.exec.delta-join.left.cache-size`	`10000`	緩衝左表點查結果的 Key 數量。僅在開啟緩衝時生效。	右邊每來一條資料，都會拿它的 Join Key 去查左表，查回的結果進入這塊緩衝，所以容量該設多大取決於右側資料驅動點查的熱點 Join Key 數量。該參數會佔用一部分記憶體，因此配置該參數時，也需要結合每條資料大小、TM記憶體大小來綜合考慮。在記憶體壓力不大的情況下，可以先用預設值試一下。GC 頻繁時適量減小。Fluss叢集點查壓力大時，適量增大。
`table.exec.delta-join.right.cache-size`	`10000`	緩衝右表點查結果的 Key 數量。僅在開啟緩衝時生效。	左側每來一條資料會去查右表。設定方法與左表緩衝相同。
`table.exec.async-lookup.buffer-capacity`	`100`	每個 Delta 運算元並發允許同時進行的非同步點查請求數。	Fluss 叢集壓力和 TM CPU、記憶體壓力都不大時調大，推薦調到千層級。該參數按每個 Delta 運算元的每個並發生效，作業在途請求總數約為並發數 × Delta 運算元數量 × 該值，調大前請按總量評估 Fluss 側的承載能力。出現記憶體緊張或 Fluss 側過載時調小。
`table.exec.async-lookup.timeout`	`3min`	單次非同步點查請求的逾時時間。	點查偶發逾時導致作業失敗時調大，用於覆蓋正常的尾部延遲。 Fluss 叢集已經過載時不建議靠調大該值解決，逾時請求會更長時間佔用並發槽位，反而加重反壓，此時應優先降低點查壓力或擴容 Fluss。

Fluss 表參數

以下參數在 Fluss 建表的 WITH 子句中設定，也可以通過 SQL Hint 針對單個作業調整。

參數名	預設值	說明	調整建議
`client.lookup.queue-size`	`25600`	用戶端等待處理的點查請求上限。	作業資料量大、點查排隊明顯時調大。TM 記憶體緊張時調小。
`client.lookup.max-batch-size`	`128`	合并為一個點查請求的最大條數。	請求量大、希望降低網路開銷時調大。對延遲敏感時調小。
`client.lookup.max-inflight-requests`	`128`	當前同時處理的點查請求數上限。	提高點查並發時調大，需同步關注 Fluss 叢集負載。
`client.lookup.batch-timeout`	`100ms`	等待批次攢滿的最長時間，逾時後立即發送。	對端到端延遲敏感時調小，希望提高批次填充率時調大。

通過 Hint 調整時，Hint 需要寫在被反查的表之後：

INSERT INTO order_item_wide
SELECT o.merchant_id, o.order_id, o.item_id, i.item_name, o.amount
FROM orders /*+ OPTIONS('client.lookup.queue-size' = '51200') */ AS o
JOIN items /*+ OPTIONS('client.lookup.queue-size' = '51200') */ AS i
  ON o.merchant_id = i.merchant_id AND o.item_id = i.item_id;