Realtime Compute for Apache Flink：主鍵表

基本概念

Fluss中的主鍵表需要確保指定主鍵的唯一性，並支援INSERT、UPDATE和DELETE操作。

例如，以下 Flink SQL 陳述式建立了一個主鍵表，其中shop_id、user_id和ds被定義為主鍵，並將資料分布到4個分桶中。

CREATE TABLE pk_table (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  ds STRING,
  PRIMARY KEY (shop_id,user_id,ds) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

• 當多條具有相同主鍵的資料被寫入時，會根據處理時間，預設保留最後一條寫入的資料。更多寫入策略請參見Merge Engines。

• 如果是分區表 ，主鍵必須包含所有分區鍵欄位（例如分區鍵是ds，則ds必須為主鍵之一）。因為資料是按分區鍵劃分，且保證每個分區內資料的主鍵唯一性可以維護，避免出現不同分區內有相同主鍵的資料。

資料分桶

對於主鍵表，Fluss 根據每條記錄的分桶鍵（bucket key）的雜湊值決定其歸屬的分桶。

• 分桶鍵必須是主鍵的一部分（不包括分區鍵）。

  例如主鍵為shop_id，user_id，ds。分區鍵為ds。則分桶鍵可以為'bucket.key' = 'user_id,shop_id'、'bucket.key' = 'user_id'或'bucket.key' = 'shop_id'。

• 若未顯式指定分桶鍵，則預設使用主鍵中除去分區鍵的部分作為分桶鍵。

  例如主鍵為shop_id、user_id，ds。分區鍵為ds。未指定分桶鍵，則分桶鍵為'bucket.key' = 'user_id,shop_id'。

• 相同雜湊值的資料會被分配到同一個分桶中，便於平行處理和負載平衡。

更多詳情請參見資料分桶。

部分列更新

對於主鍵表，Fluss 支援部分列更新（Partial Update），您可以唯寫入部分列來對資料進行逐步更新，並最終得到完整的資料。需要注意的是，要寫入的列必須包含主鍵列。

CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
);

可以分別寫入部分列資料，但必須包含主鍵列，如下圖所示

Merge Engines

Fluss中的合并引擎（Merge Engine）是一個核心組件，旨在高效處理和整合主鍵表的資料更新操作。它為使用者提供了靈活性，允許定義如何將新進資料記錄與具有相同主鍵的現有記錄進行合并。此外，使用者還可以指定不同的合并引擎，以根據具體使用情境自訂合并行為。

以下是支援的合并引擎：

• LastRow Merge Engine

  當新記錄與現有記錄具有相同主鍵時，保留最新的記錄（依據處理時間，最後寫入的記錄）。

• FirstRow Merge Engine

  • 與預設策略相反，該引擎會保留最早的記錄（依據處理時間，即最先寫入的記錄）。

  • 適用於需要保留初始資料狀態的情境。

• Versioned Merge Engine

  • 支援基於版本號碼的合并邏輯。

  • 使用者可以根據記錄中的版本號碼欄位來決定保留哪個版本的資料。

  • 適用於需要管理資料版本化或累加式更新的情境。

• Aggregation Merge Engine

  • 對於相同主鍵進行分組，並根據指定的彙總函式，對每個非主鍵欄位逐一進行計算。

  • 涵蓋大量常用的彙總函式，簡化了資料彙總的情境。

Changelog 產生

Fluss 會記錄主鍵表中資料增刪與更新的變更資料，即 changelog，下遊消費者可以直接消費 changelog 來得到表的變更。

如果寫入 Fluss 主鍵表的資料依次為插入兩條資料(1, 2.0, 'apple')，(1, 4.0, 'banana'), 再刪除一條資料(1, 4.0, 'banana')，則將產生如下的變更資料：

+I(1, 2.0, 'apple')
-U(1, 2.0, 'apple')
+U(1, 4.0, 'banana')
-D(1, 4.0, 'banana')

本機存放區與分區生命週期

主鍵表的儲存結構包含用於回撤流的 LogTablet 和用於狀態查詢的 KvTablet。由於兩者機制不同，僅依賴常規 TTL 無法完全釋放本地磁碟空間。

• LogTablet 管理： table.log.ttl參數僅控制 Changelog 的保留時間，不影響 KvTablet 的資料存放區大小。

• KvTablet 管理（推薦）： 若需徹底清理到期資料並釋放 KvTablet 佔用的本地磁碟空間，必須啟用分區生命週期管理（Partition TTL）。

  • 開啟自動分區： 設定 table.auto-partition.enabled = true。

  • 設定保留原則： 配置 table.auto-partition.num-retention 指定保留的分區數量。

  • 執行邏輯： 系統將定期輪詢並物理刪除超出保留數量的舊分區，從而實現本機存放區空間的有效回收。

自增列

自增列可自動為每行資料產生唯一的遞增數值，無需手動指定。典型應用情境包括：

• 加速去重計數： 將 STRING 類型的標識符映射為自增整數，對整數值執行去重計數，查詢速度可提升數倍至數十倍。

• RoaringBitmap 彙總： 配合彙總合并引擎中內建的 rbm32（32 位）和 rbm64（64 位元）函數，將自增整數 ID 彙總為 RoaringBitmap，實現高效的去重計數。

CREATE TABLE uid_mapping (
  user_id STRING,
  uid BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'auto-increment.fields' = 'uid',
  'bucket.num' = '2'
);

基本特性

• 唯一性： 自增列產生的值在整張表範圍內唯一。

• 單調性： 每個表分桶會在 TabletServer 上緩衝一批自增 ID，因此全域範圍內無法保證嚴格單調遞增，僅保證大致按時間順序遞增。

  緩衝數量由表屬性 table.auto-increment.cache-size 控制，預設值為 100,000。較大的緩衝可提升分配效能，但會降低單調性。該屬性在表建立後不可修改。

樣本說明

• table.auto-increment.cache-size = 100000（預設值）

• 表有 2 個分桶，系統啟動時向每個分桶預分配一段 ID 範圍：

  • 分桶 0 → [1, 100000]

  • 分桶 1 → [100001, 200000]

如果資料落到分桶 1，uid 就從 100001 開始。分桶 0 用完 100,000 個 ID 後，會再申請下一批 [200001, 300000]，以此類推。每個分桶可以分配的 ID 總量沒有限制。100000不是上限，是緩衝批次大小。

情境應用

自增列可與 Flink Lookup Join 的 lookup.insert-if-not-exists 選項配合使用，實現流處理過程中自動構建字典表：當尋找鍵未命中時，Fluss 自動插入新行並分配唯一的自增 ID。

典型情境是將高基數的字串標識符映射為緊湊的整數 ID，以便後續使用 RoaringBitmap 進行高效去重彙總。詳情請參見維表關聯。

使用限制

• 僅支援主鍵表。

• 不允許顯式指定自增列的值，只能由系統隱式分配。

• 每張表只能有一個自增列。

• 自增列的類型必須為 INT 或 BIGINT。

• 不支援指定自增列的起始值和步長。

結合 Lookup Join 構建字典表

通過將自增列與 Flink Lookup Join 的 `lookup.insert-if-not-exists` 選項結合，可以在流處理過程中自動構建字典表——當尋找鍵未命中時，Fluss 自動插入新行並分配自增 ID。這對於將高基數字串標識符映射為緊湊整數 ID、用於高效的 RoaringBitmap 去重彙總尤為實用。詳情和樣本請參見維表關聯。

主鍵表的讀取與查詢

• 資料清理：Fluss 服務僅管理自身的儲存資源，不執行 Paimon 儲存層的資料清理或記憶體回收操作。

• 即時查詢限制：Fluss 僅提供生命週期內（未到期）資料的查詢服務。一旦資料超過保留期限，系統將其從 Fluss 層移除，使用者無法通過 Fluss 介面檢索到該部分資料。

• 歷史資料訪問：對於已到期的分區資料，系統後續將支援通過 Paimon 的 Union Read 機制進行查詢，以實現對全量歷史資料的訪問。