Realtime Compute for Apache Flink：Fluss SQL連接器

功能說明

Fluss 主鍵表支援全增量一體化消費：先讀取表中的歷史全量快照資料，全量資料讀取完畢後自動無縫切換到增量流式消費，保證資料的完整性和一致性。

消費模式

通過 scan.startup.mode 參數控制消費起始位點：

SQL 樣本

-- 全增量一體化消費（預設模式）
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;

-- 從指定時間戳記開始消費，通過 Hint 指定參數
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'timestamp', 'scan.startup.timestamp' = '2025-01-01 00:00:00') */;

-- 從最新位點開始消費
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

日誌訂閱

Fluss 主鍵表天然支援 CDC（Change Data Capture）日誌訂閱，下遊 Flink 作業可以即時消費到完整的變更事件流（INSERT、UPDATE_BEFORE、UPDATE_AFTER、DELETE），無需額外開啟 Binlog 配置。

這意味著：

• 主鍵表的所有變更都會自動產生 CDC 日誌。

• 下遊可直接訂閱 Fluss 主鍵表擷取完整的變更流，用於即時資料同步或流式 ETL。

SQL 樣本

-- 直接讀取主鍵表即可擷取 CDC 變更流
SELECT * FROM `fluss_catalog`.`fluss_db`.orders;
-- 輸出包含完整的 INSERT / UPDATE / DELETE 事件

虛擬表（Virtual Table）

除了直接訂閱 CDC 流之外，Fluss 還提供了虛擬表機制，通過在基礎資料表名後追加特定尾碼即可訪問帶有中繼資料的變更日誌，無需額外儲存資料。虛擬表是系統自動產生的，不需要使用者手動建立。

-- 讀取 orders 表的 changelog 虛擬表
SELECT * FROM `fluss_catalog`.`fluss_db`.`orders$changelog`;

-- 通過 Hint 指定從最新位點開始消費
SELECT * FROM `fluss_catalog`.`fluss_db`.`orders$changelog`
  /*+ OPTIONS('scan.startup.mode' = 'latest') */;

-- 過濾出刪除事件
SELECT * FROM `fluss_catalog`.`fluss_db`.`orders$changelog`
WHERE _change_type = 'delete';

-- 讀取 users 表的 binlog 虛擬表
SELECT * FROM `fluss_catalog`.`fluss_db`.`users$binlog`;

-- 擷取更新事件中欄位變更前後的值
SELECT
  `before`.user_name AS old_name,
  `after`.user_name  AS new_name,
  _commit_timestamp  AS change_time
FROM `fluss_catalog`.`fluss_db`.`users$binlog`
WHERE _change_type = 'update';

功能說明

Fluss 支援部分列更新（Partial Update），允許多個資料流獨立地向同一張主鍵寬表寫入各自關心的欄位，Fluss 引擎會自動按主鍵進列欄位層級的合并。這種方式無需在 Flink 中維護複雜的多流 Join 狀態，避免了大狀態問題。

使用情境

典型情境為即時使用者畫像寬表構建：不同的資料來源（如推薦系統、曝光日誌、點擊日誌、購物車資料）分別更新使用者畫像表中各自對應的欄位，最終形成一張完整的使用者 360 度視圖。

使用限制

• 目標表必須為主鍵表。

• 每個資料流的寫入欄位必須包含完整的主鍵欄位。

• 計算拓撲中不包含多流 Join 節點，直接由 Fluss 在儲存層完成 Merge。

SQL 樣本

-- 1. 建立寬表
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,
    imp_cnt   INT,
    click_cnt INT,
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

-- 2. 多流分別寫入各自欄位
-- 資料流 1：寫入推薦分數
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, rec_score)
SELECT user_id, item_id, rec_score 
FROM `fluss_catalog`.fluss_db.recommendations;

-- 資料流 2：寫入曝光次數
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, imp_cnt)
SELECT user_id, item_id, imp_cnt 
FROM `fluss_catalog`.fluss_db.impressions;

-- 資料流 3：寫入點擊次數
INSERT INTO `fluss_catalog`.fluss_db.user_rec_wide (user_id, item_id, click_cnt)
SELECT user_id, item_id, click_cnt 
FROM `fluss_catalog`.fluss_db.clicks;

上述三條 INSERT 語句可作為獨立的 Flink 作業運行，Fluss 儲存層會自動將相同主鍵的欄位合并成完整的行。

功能說明

Delta Join 是 Fluss 提供的新一代雙流 Join 範式，將資料更新行為下沉到 Fluss 儲存層完成，利用主鍵表和二級索引實現高效尋找，無需在 Flink 中維護 Join State。

核心優勢：

• 無 Join State：省去冗餘的狀態儲存，避免大狀態導致的 GC 問題。

• 低資源消耗：實測 Flink 記憶體與 CPU 限定下降超 86%。

• 更穩定高效：避免狀態膨脹帶來的效能瓶頸。

與傳統 Join 對比

使用限制

• 左右表必須為 Fluss 的主鍵表（支援分區表）。

• 分桶鍵（Bucket Key）需為主鍵首碼。

• Join Key 必須包含分桶鍵；若為分區表，還需包含分區鍵。

SQL 樣本

1. 建立表

CREATE TABLE orders (
  user_id BIGINT,
  order_date DATE,
  order_id BIGINT,
  amount DECIMAL(10, 2),
  PRIMARY KEY (user_id, order_id, order_date) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

CREATE TABLE order_enhance (
  user_id BIGINT,
  order_id BIGINT,
  risk_level TINYINT,
  is_fraud BOOLEAN,
  update_time TIMESTAMP(3),
  PRIMARY KEY (user_id, order_id) NOT ENFORCED
) WITH ('bucket.key' = 'user_id, order_id');

2. 執行 Delta Join

SELECT * FROM orders o
INNER JOIN order_enhance e
ON o.user_id = e.user_id AND o.order_id = e.order_id;

3. 通過 Hint 最佳化 Lookup 效能

SELECT * FROM orders o /*+ OPTIONS('client.lookup.queue-size'='2560') */
INNER JOIN order_enhance e /*+ OPTIONS('client.lookup.queue-size'='2560') */
ON o.user_id = e.user_id AND o.order_id = e.order_id;

作業運行參數配置

使用 Delta Join 時，需在 Flink 作業的運行參數中添加以下配置：

table.exec.delta-join.cache-enabled: 'true'
table.exec.async-lookup.buffer-capacity: '3000'
table.optimizer.delta-join.strategy: FORCE
table.exec.delta-join.left.cache-size: '100'
table.exec.delta-join.right.cache-size: '1000'

成功啟用後，可在 Flink 作業狀態總覽中看到 Delta Join 節點，表明最佳化已生效。

功能說明

Fluss 主鍵表支援作為 Flink SQL 的維表使用，通過 Lookup Join 可以在流處理中即時關聯維度資料。Fluss 支援高效的 KV 點查和二級索引尋找，適合低延遲的維表關聯情境。

使用限制

• 僅支援主鍵表作為維表。

• Join 條件必須包含維表的全部主鍵欄位。

同步與非同步模式

SQL 樣本

-- 通過 Hint 配置 Lookup 緩衝，進行維表關聯
SELECT
  o.order_id,
  o.amount,
  u.user_name,
  p.product_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.cache' = 'PARTIAL', 'lookup.partial-cache.expire-after-write' = '60s', 'lookup.partial-cache.max-rows' = '10000') */
  FOR SYSTEM_TIME AS OF o.proctime AS u
  ON o.user_id = u.user_id
LEFT JOIN `fluss_catalog`.`fluss_db`.product_dim
  FOR SYSTEM_TIME AS OF o.proctime AS p
  ON o.product_id = p.product_id;

-- 使用同步 Lookup 模式（適合調試情境）
SELECT o.order_id, o.amount, d.user_name
FROM `fluss_catalog`.`fluss_db`.orders AS o
LEFT JOIN `fluss_catalog`.`fluss_db`.user_dim
  /*+ OPTIONS('lookup.async' = 'false') */
  FOR SYSTEM_TIME AS OF o.proctime AS d
  ON o.user_id = d.user_id;

功能說明

Fluss 主鍵表支援配置不同的 Merge Engine，用於控制相同主鍵的多條記錄如何合并。通過 table.merge-engine 參數設定，支援以下策略：

SQL 樣本

-- 使用 first_row 實現去重
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.event_dedup (
    event_id STRING,
    event_type STRING,
    event_time TIMESTAMP(3),
    payload STRING,
    PRIMARY KEY (event_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'first_row'
);

-- 使用 versioned 按版本列更新
CREATE TABLE IF NOT EXISTS `fluss_catalog`.fluss_db.user_profile (
    user_id BIGINT,
    user_name STRING,
    age INT,
    update_version BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'table.merge-engine' = 'versioned',
  'table.merge-engine.versioned.ver-column' = 'update_version'
);

功能說明

Fluss 支援分區表，通過 PARTITIONED BY 將資料按分區鍵劃分到不同的物理分區中，適用於按時間或業務維度管理資料生命週期和提升查詢效能的情境。分區表支援單欄位分區和多欄位分區，並提供手動管理、自動管理和動態建立三種正交的分區管理原則。

基本規則

• 分區鍵的資料類型需要是 STRING。

• 主鍵表的分區鍵需要是主鍵的子集。

• 自動分區規則僅在建表時配置，建表後不可修改。

分區管理原則

三種策略是正交的，可以組合使用。例如，可以同時開啟自動管理（按時間自動建立和清理）和動態建立（兜底處理未預建立的分區）。

SQL 樣本

1. 建立單欄位分區表（自動分區管理）

CREATE TABLE `fluss_catalog`.fluss_db.site_access (
  site_id BIGINT,
  user_name STRING,
  pv BIGINT,
  event_day STRING,
  PRIMARY KEY (site_id, event_day) NOT ENFORCED
) PARTITIONED BY (event_day) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-precreate' = '5',
  'table.auto-partition.num-retention' = '30',
  'table.auto-partition.time-zone' = 'Asia/Shanghai'
);

2. 建立多欄位分區表

CREATE TABLE `fluss_catalog`.fluss_db.region_access (
  access_id BIGINT,
  user_name STRING,
  event_date STRING,
  region STRING,
  PRIMARY KEY (access_id, event_date, region) NOT ENFORCED
) PARTITIONED BY (event_date, region) WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.key' = 'event_date',
  'table.auto-partition.time-unit' = 'DAY',
  'table.auto-partition.num-retention' = '7'
);

說明：多欄位分區時，需通過 table.auto-partition.key 指定用於自動分區的時間鍵。多欄位分區僅支援自動到期清理，不支援自動預建立（num-precreate 強製為 0）。

3. 手動管理分區

-- 查看所有分區
SHOW PARTITIONS `fluss_catalog`.fluss_db.site_access;

-- 手動添加分區
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  ADD PARTITION (event_day = '20250315');

-- 手動刪除分區
ALTER TABLE `fluss_catalog`.fluss_db.site_access
  DROP PARTITION (event_day = '20250101');

4. 動態建立分區

寫入資料時，如果目標資料分割不存在，用戶端會自動建立對應分區（預設開啟）：

-- 寫入資料，若分區 '20250701' 不存在則自動建立
INSERT INTO `fluss_catalog`.fluss_db.site_access
VALUES (1, 'hello', 100, '20250701');

動態建立由用戶端參數 client.writer.dynamic-create-partition.enabled（預設 true）控制，受叢集層級的 max.partition.num（預設 1000）和 max.bucket.num（預設 128000）限制。

分區下推

流式或批式查詢分區表時，Fluss 支援分區下推（分區裁剪），通過 WHERE 條件中的分區鍵過濾，僅讀取匹配的分區資料，減少不必要的 I/O 開銷：

-- 僅讀取指定日期的分區資料
SELECT * FROM `fluss_catalog`.fluss_db.site_access
WHERE event_day = '20250315';

-- 多欄位分區表按 region 過濾
SELECT * FROM `fluss_catalog`.fluss_db.region_access
WHERE region = 'US';

自動分區參數說明

最佳實務

• 選擇合適的分區粒度：根據資料量和查詢模式選擇時間粒紋。資料量大、查詢時間範圍小的情境建議按 DAY 或 HOUR；資料量較小的情境可按 MONTH 或 QUARTER。

• 合理設定保留原則：通過 num-retention 控制歷史分區數量，避免分區數過多導致中繼資料膨脹和叢集管理負擔增大。

• 組合使用自動管理與動態建立：自動管理負責常規的時間分區建立和清理，動態建立作為兜底機制處理異常或延遲資料。

• 注意叢集限制：單表分區數受 max.partition.num 限制（預設 1000），總 Bucket 數受 max.bucket.num 限制（預設 128000），分區數 x bucket.num 不應超過此上限。

功能說明

Fluss 支援湖流一體架構，可將 Fluss 中的即時資料自動同步到資料湖（如 Apache Paimon），實現冷熱資料自動階層式存放區，兼顧即時查詢的低延遲與歷史資料的低儲存成本。

關鍵特性：

• 自動將 Fluss 中的流式資料寫入 Paimon 湖儲存。

• 冷熱資料自動分層，熱資料保留在 Fluss 流儲存中，冷資料沉降到 Paimon。

• 通過統一的 Flink SQL 查詢介面同時訪問流儲存和湖儲存中的資料。

配置參數

功能說明

Fluss 支援在 Flink SQL 的批模式下對錶資料進行快速探查，包括 LIMIT 查詢、主鍵點查和 COUNT(*) 統計，無需啟動持續啟動並執行流式作業即可快速查看錶中的資料。這對於資料開發調試、資料驗證和問題排查非常實用。

LIMIT 查詢

支援對主鍵表和日誌表執行 LIMIT 查詢，快速擷取表中的少量樣本資料。

-- 從主鍵表中查看前 10 條資料
SELECT * FROM `fluss_catalog`.`fluss_db`.orders LIMIT 10;

-- 從日誌表中查看前 5 條資料
SELECT * FROM `fluss_catalog`.`fluss_db`.events LIMIT 5;

-- 結合投影，僅查看部分列
SELECT order_id, amount, order_time 
FROM `fluss_catalog`.`fluss_db`.orders LIMIT 20;

主鍵點查

對於主鍵表，支援通過 WHERE 條件按主鍵進行精確點查，利用 Fluss 的 KV 儲存引擎實現毫秒級響應。

-- 按主鍵精確查詢單條記錄
SELECT * FROM `fluss_catalog`.`fluss_db`.orders
WHERE order_id = 1001;

-- 複合主鍵的點查
SELECT * FROM `fluss_catalog`.`fluss_db`.user_orders
WHERE user_id = 12345 AND order_id = 6789;

COUNT(*) 統計

支援對主鍵表執行 COUNT(*) 查詢，快速擷取表中的總記錄數。

-- 統計主鍵表中的總行數
SELECT COUNT(*) FROM `fluss_catalog`.`fluss_db`.orders;