管理表 - Realtime Compute for Apache Flink

本文介紹如何通過 Flink SQL 建立和管理 Fluss 表。

使用限制

僅Realtime Compute引擎VVR 11.2及以上版本進行管理。

建立主鍵表

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

CREATE TABLE `my-catalog`.`my_db`.`my_pk_table` (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (shop_id, user_id) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

說明

建表時必須規劃分桶數量。

日誌表支援後期調整分桶配置，但主鍵表的分桶數一旦確立，便無法修改。因此，建議在建立表時直接根據資料規模設定分桶，計算公式如下：

分桶數 = 單分區總資料量 ÷ 目標桶大小

建立日誌表

CREATE TABLE `my-catalog`.`my_db`.`my_log_table` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'bucket.num' = '8'
);

建立分區表

目前 Fluss 的分區鍵只支援 String 類型。
對於分區主鍵表，其分區鍵（下面例子中的 dt 欄位）必須是主鍵的子集。

分區主鍵表

分區鍵為 dt，主鍵為 [dt, shot_id, user_id]。

CREATE TABLE `my-catalog`.`my_db`.`my_part_pk_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
);

分區日誌表

建立了一張分區日誌表，其分區鍵為 dt。

CREATE TABLE `my-catalog`.`my_db`.`my_part_log_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
);

建立分區

ALTER TABLE my-catalog.my_db.my_part_pk_table 
ADD PARTITION (dt = '2025-03-05');

建立自動分區表

自動分區表在分區表的基礎上，加入了自動分區的邏輯，Fluss 會根據你配置的自動分區策略為你提前建立好分區。

自動分區主鍵表

分區鍵為 dt，自動分區建立間隔為天（day）。

CREATE TABLE `my-catalog`.`my_db`.`my_auto_part_pk_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'day'
);

自動分區日誌表

分區鍵為 dt，自動分區建立間隔為天（day）。

CREATE TABLE `my-catalog`.`my_db`.`my_auto_part_log_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'day'
);

更新表

新增列

說明

該功能要求 Flink Realtime Compute引擎（VVR）版本為 11.5 及以上，同時 Fluss 版本需為 0.8-ali-3.0 及以上。為避免影響讀寫操作，建議先將 Flink 作業升級到 11.5 及以上版本，然後再將 Fluss 叢集升級到 0.8-ali-3.0。

Fluss 支援通過添加新列來演化表的結構，支援添加任意資料類型的列，包括複雜類型（如 ROW、MAP 和 ARRAY）。

這是一個輕量級、僅涉及中繼資料的操作，具有以下優勢：

零資料重寫：添加列時無需重寫或遷移現有的資料檔案。
即時執行：無論表的規模如何，該操作都能在毫秒層級完成。
高可用性：在整個表結構演化過程中，表始終保持線上且完全可用，不會對正在啟動並執行用戶端造成任何中斷。

目前，此功能具有以下限制：

列順序：新增列將始終追加到現有列列表的末尾。
可為空白性：為了確保與現有資料的相容性，只能向表中添加可為空白的列。
嵌套欄位：目前不支援在現有的嵌套 ROW 類型中添加欄位。此類操作屬於“更新列類型”的範疇，將在未來的版本中提供支援。

您可以使用 ALTER TABLE 語句添加單個列或多個列。

-- Add a single column at the end of the table
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD user_email STRING COMMENT 'User email address';

-- Add multiple columns at the end of the table
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD (
    user_email STRING COMMENT 'User email address',
    order_quantity INT
);

修改表屬性

通過 ALTER TABLE SET 語句可修改表的一個或多個連接器選項及儲存選項。若指定的選項已存在，新值將覆蓋原有配置。

修改儲存選項時，Fluss 叢集將Live App變更，無需重建表。

支援修改的選項

除 bootstrap.servers 外的所有 Read Options、Write Options、Lookup Options 和 Other Options。
以下 Storage Options：
- table.datalake.enabled：開啟或關閉湖流一體儲存。
- table.datalake.freshness：設定湖儲存的資料新鮮度。

-- 開啟湖流一體儲存
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.enabled' = 'true');

-- 設定湖儲存資料新鮮度為5分鐘
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.freshness' = '5min');

重要

若表已開啟湖流一體儲存（table.datalake.enabled），則湖格式首碼選項（如 paimon.*）不可再修改。

重設表屬性

通過 ALTER TABLE RESET 語句可將一個或多個連接器選項及儲存選項恢複為預設值。

-- 將湖流一體儲存恢複為預設關閉
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` RESET ('table.datalake.enabled');

複製表（Schema 複製）

Fluss 支援通過create table like文法建立一張 schema，分區資訊和表參數相同的表。

該能力仍是建立Fluss表，需在資料開發 > 資料查詢 > 建立臨時查詢指令碼中使用。

-- 有一張 datagen 的暫存資料表
CREATE TEMPORARY TABLE datagen (
 user_id BIGINT,
 item_id BIGINT,
 behavior STRING,
 dt STRING,
 hh STRING
) WITH (
 'connector' = 'datagen',
 'rows-per-second' = '10'
);
-- 建立一張去掉 like option 的 Fluss 表
CREATE TABLE my-catalog.my_db.my_like_db LIKE datagen (EXCLUDING OPTIONS);

刪除表

DROP TABLE `my-catalog`.`my_db`.`my_db`;