Realtime Compute for Apache Flink：管理 Fluss Catalog

前提條件

• 已建立 Fluss 叢集，具體操作請參見建立 Fluss 叢集。

• Fluss 叢集與 Flink 工作空間位於同一地區且網路互連（位於同一 VPC 或已配置網路連通）。

使用限制

• 不支援修改 Catalog。如需修改，需先刪除已建立的 Catalog 再重新建立。

• 密碼方式建立的 Catalog 使用管理員身份串連 Fluss 叢集，所有使用該 Catalog 的作業均以管理員權限訪問資料，無法按使用者隔離資料存取權限。

• 免密方式建立的 Catalog 根據當前操作使用者身份自動進行許可權認證，不同使用者使用同一個 Catalog 時，只能訪問其在 Fluss 上被授權的資料。

• 免密方式建立的 Catalog 依賴 RAM 許可權。建立 Catalog 時，需具備 fluss:ListClusters、fluss:GetCluster、fluss:GetToken、fluss:GetFeatureList 許可權；啟動引用該 Catalog 的作業時，至少需具備 fluss:GetToken 許可權。

• 免密方式下，實際訪問 Fluss 叢集的身份需登入為 Fluss 叢集成員並開啟密碼認證，且已獲得目標資料庫或表的資源授權。未滿足時，作業啟動失敗並提示無法擷取 Fluss 運行時憑證。

建立 Fluss Catalog

根據 Fluss 叢集版本的不同，建立 Catalog 的方式有所差異：

• Fluss 叢集版本低於 0.9-ali-5.0：僅支援密碼方式建立 Catalog，需要使用管理使用者名和密碼。

• Fluss 叢集版本為 0.9-ali-5.0 及以上：推薦使用免密方式建立 Catalog，無需輸入使用者名稱和密碼，系統會根據當前操作使用者身份自動完成許可權認證。密碼方式仍可用，但不推薦。

CREATE CATALOG fluss_catalog WITH (
  'type' = 'fluss',
  'bootstrap.servers' = '<bootstrap.servers>',
  'default-database' = 'fluss',
  'client.security.protocol' = 'SASL',
  'client.security.sasl.mechanism' = 'PLAIN',
  'client.security.sasl.username' = '<username>',
  'client.security.sasl.password' = '<password>'
);

建立自訂 Catalog

Fluss 迭代快於Realtime Compute Flink。此處提供最新的自訂 Catalog 包，用於提前體驗尚未隨 VVR 版本發布的 Fluss Catalog 功能。

自訂 Catalog 僅支援通過控制台上傳 JAR 包建立。建立後的使用方式與內建 Fluss Catalog 一致。

1. 下載自訂 Catalog JAR 包：fluss-ali-vvr-11-0.9-ali-7.1.jar

2. 在建立 Catalog對話方塊中，選擇自訂 Catalog頁簽。

3. 單擊建立自訂 Catalog 類型，上傳已下載的 JAR 包，單擊下一步。

4. 載入完成後，選擇Catalog 類型為fluss-latest，單擊確認。

5. 選擇此 Catalog，單擊下一步，建立自訂 Catalog。

   CREATE CATALOG fluss_catalog WITH (
     'type' = 'fluss-latest',
     'bootstrap.servers' = '<bootstrap.servers>',
     'default-database' = 'fluss',
     'client.security.protocol' = 'SASL',
     'client.security.sasl.mechanism' = 'PLAIN',
     'client.security.sasl.username' = '<username>',
     'client.security.sasl.password' = '<password>'
   );

   參數	說明	是否必填	備忘
   type	Catalog 類型。	是	固定值為 fluss-latest。
   bootstrap.servers	Fluss 叢集的伺服器位址列表。	是	可通過 Fluss 控制台的叢集詳情進行查看。
   default-database	預設串連的資料庫名稱。	否	預設值為 fluss。
   client.security.protocol	安全認證協議。	是	固定值為 SASL。
   client.security.sasl.mechanism	SASL 認證機制。	是	固定值為 PLAIN。
   client.security.sasl.username	Fluss 執行個體的管理使用者名。	是	可通過 Fluss 控制台的叢集詳情進行查看。
   client.security.sasl.password	Fluss 執行個體的管理員密碼。	是

更新自訂 Fluss Catalog

不支援直接修改已有 Fluss Catalog 的版本。更新方式為：先刪除原有 Catalog，再使用新版本的 Fluss Catalog JAR 包建立同名 Catalog。

Catalog 名稱必須保持不變，以確保作業重啟或重新上線時仍能引用該 Catalog，並載入新版本 JAR 包。

更新對正在啟動並執行 Flink 作業無感，不觸發作業重啟，不影響運行中的作業狀態。後續使用新版本恢複作業時，狀態保持相容。

查看 Fluss Catalog

Fluss Catalog 配置成功後，按以下步驟查看 Fluss 中繼資料。

1. 登入Realtime Compute控制台。

2. 單擊目標工作空間操作列下的控制台。

3. 在左側導覽列，單擊資料管理。

4. 在 Catalog 列表頁面，查看 Catalog 名稱和類型。

5. 單擊查看，查看目標 Catalog 下的資料庫和表。

使用 Fluss Catalog

CREATE TABLE `${catalog_name}`.`${db_name}`.`${table_name}` (
  ...
);

-- 建立主鍵表
CREATE TABLE `fluss_catalog`.`fluss`.`product` (
  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 `fluss_catalog`.`fluss`.`orders` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
);

USE CATALOG fluss_catalog;
USE fluss;

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

INSERT INTO `${catalog_name}`.`${db_name}`.`${table_name}`
SELECT ...
FROM ${other_source_table};

INSERT INTO `fluss_catalog`.`fluss`.`product`
SELECT shop_id, user_id, num_orders, total_amount
FROM source_table;

INSERT INTO ${other_sink_table}
SELECT ...
FROM `${catalog_name}`.`${db_name}`.`${table_name}`;

INSERT INTO sink_table
SELECT shop_id, user_id, num_orders, total_amount
FROM `fluss_catalog`.`fluss`.`product`;

刪除 Fluss Catalog

DROP CATALOG ${catalog_name};

相關文檔

• Fluss 表資料寫入和消費：介紹如何向 Fluss 表中寫入資料和從 Fluss 表中消費資料。

• 管理自訂 Catalog：如果內建的 Catalog 無法滿足業務需求，可使用自訂 Catalog。