Realtime Compute for Apache Flink：Serverless StarRocks訪問Fluss（Beta）

步驟一：建立 Fluss Catalog

1. 登入 EMR Serverless StarRocks 控制台，在執行個體列表頁面單擊目標執行個體操作列的串連執行個體。詳情請參見通過EMR StarRocks Manager串連StarRocks執行個體。

2. 在 SQL Editor 中執行以下 SQL，將 Fluss 註冊為 External Catalog：

   CREATE EXTERNAL CATALOG <catalog_name>
   PROPERTIES (
       'type' = 'fluss',
       'bootstrap.servers' = '<Fluss叢集bootstrap地址>',
       'fluss.option.client.security.protocol' = 'SASL',
       'fluss.option.client.security.sasl.mechanism' = 'PLAIN',
       'fluss.option.client.security.sasl.username' = '<Fluss使用者名稱>',
       'fluss.option.client.security.sasl.password' = '<Fluss密碼>'
   );

   參數說明

   參數	是否必填	說明
   catalog_name	是	Catalog 名稱。只能包含字母、數字、底線，且以字母開頭，長度不超過 64 個字元。
   type	是	資料來源類型，固定為 fluss。
   bootstrap.servers	是	Fluss 叢集的 bootstrap 地址。可在 Fluss 執行個體詳情頁擷取。
   fluss.option.client.security.protocol	是	認證協議。阿里雲上預設為 SASL。
   fluss.option.client.security.sasl.mechanism	是	SASL 認證機制。預設為 PLAIN（帳號密碼登入）。
   fluss.option.client.security.sasl.username	是	SASL 認證的帳號名。
   fluss.option.client.security.sasl.password	是	SASL 認證的帳號密碼。

3. Catalog 建立完成後，可以通過三段式命名（Catalog.Database.Table）直接查詢，無需切換上下文。Fluss 的預設資料庫名稱為 fluss：

   SELECT * FROM <catalog_name>.<database_name>.<table_name>;

   切換到 Fluss Catalog 後查詢：

   -- 切換 Catalog 和資料庫
   SET CATALOG <catalog_name>;
   USE <db_name>;
   
   -- 直接查詢目標表
   SELECT COUNT(*) FROM <table_name> LIMIT 10;

   直接切換到對應庫查詢

   USE <catalog_name>.<db_name>;

步驟二：通過尾碼指定查詢模式

對於已開啟湖流一體的 Fluss 表（建表時設定 table.datalake.enabled = 'true'），資料會同時存在於 Fluss 叢集（即時熱資料）和 Paimon 資料湖（分層歷史資料）兩個層次。StarRocks 支援通過在表名後添加尾碼來指定查詢模式，靈活訪問不同層次的資料。各查詢模式說明如下：

以線上銷售訂單表 order 為例，該表即時記錄使用者的下單資料（包含訂單 ID、訂單金額、下單時間、地區等欄位），並已開啟湖流一體。以下展示三種查詢模式的典型業務情境：

-- 情境一：即時監控今日訂單量與金額（$rt，秒級延遲）
SELECT COUNT(*) AS order_cnt, SUM(order_amount) AS total_amount
FROM `fluss_catalog`.`fluss`.order$rt
WHERE order_time >= CURRENT_DATE;

-- 情境二：計算本周與上周銷售額環比增長率（$lake，查詢 Paimon 歷史資料）
SELECT
    this_week AS 本周銷售額,
    last_week AS 上周銷售額,
    ROUND((this_week - last_week) * 100.0 / NULLIF(last_week, 0), 2) AS 環比增長率
FROM (
    SELECT
        SUM(CASE WHEN YEARWEEK(order_time) = YEARWEEK(CURRENT_DATE)     THEN order_amount ELSE 0 END) AS this_week,
        SUM(CASE WHEN YEARWEEK(order_time) = YEARWEEK(CURRENT_DATE) - 1 THEN order_amount ELSE 0 END) AS last_week
    FROM `fluss_catalog`.`fluss`.order$lake
) t;

-- 情境三：查詢近 7 天各地區的訂單總額（無尾碼，Union Read 自動合并冷熱資料）
SELECT region, SUM(order_amount) AS total_amount
FROM `fluss_catalog`.`fluss`.order
WHERE order_time >= CURRENT_DATE - INTERVAL '7' DAY
GROUP BY region
ORDER BY total_amount DESC;

查看已建立的 Catalog

-- 查看所有 Catalog
SHOW CATALOGS;

-- 查看指定 Catalog 的建立語句
SHOW CREATE CATALOG <catalog_name>;

查看 Fluss 表結構

DESCRIBE <catalog_name>.<database_name>.<table_name>;

查看 Fluss 資料庫列表

SHOW DATABASES FROM <catalog_name>;

刪除 Fluss Catalog

刪除 Catalog 僅移除 StarRocks 中的映射關係，不會影響 Fluss 中的實際資料。

DROP CATALOG <catalog_name>;