讀取資料 - Realtime Compute for Apache Flink

本文將為您介紹讀取Fluss資料的流批方式。

資料讀取

流模式讀取

支援多種消費模式，對於主鍵表，預設消費模式是initial，首先消費全量資料，然後再消費增量資料。

登入Realtime Compute管理主控台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊資料開發 > ETL。
單擊+後，單擊建立流作業，填寫檔案名稱並選擇引擎版本，單擊建立。

編寫作業代碼。

SELECT * FROM `fluss-catalog`.`my_db`.`my_table`;

# earliest模式
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

批模式讀取

主鍵表與日誌表均支援 LIMIT 讀取，可快速返回指定條數結果，極大便利資料預覽。
主鍵表與日誌表（0.9-ali-1.0 及以上版本）均已支援 COUNT(*) 統計查詢。
主鍵表額外支援基於主鍵和索引鍵的精確匹配的點查功能。

登入Realtime Compute管理主控台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊資料開發 > ETL。
單擊+後，單擊建立批作業，填寫檔案名稱並選擇引擎版本和SQL方言，單擊建立。

編寫作業代碼。

# LIMIT 讀取
SELECT * FROM `fluss-catalog`.`my_db`.`my_table` LIMIT 10;

# 精確查詢（僅主鍵表）
SELECT * FROM `fluss-catalog`.`my_db`.`my_pk_tbl` WHERE shop_id = 10000 and user_id = 123456;

# 彙總查詢
SELECT count(*) FROM `fluss-catalog`.`my_db`.`my_log_table`;

查詢最佳化

謂詞下推

在 SELECT ... WHERE 中編寫的過濾條件，Flink 自動將合格謂詞下推到 Fluss 服務端。服務端根據每個 RecordBatch 的列統計資訊評估過濾謂詞，跳過不匹配的資料區塊，減少網路傳輸和還原序列化開銷。

支援的過濾操作符

比較操作符：=、<>、>、>=、<、<=
IN (...)
IS NULL、IS NOT NULL
BETWEEN ... AND ...
LIKE：支援首碼匹配（'abc%'）、尾碼匹配（'%abc'）和包含匹配（'%abc%'）
邏輯組合：AND、OR

使用限制

僅適用於日誌表，主鍵表不支援。
過濾條件中引用的所有列必須開啟列統計資訊（table.statistics.columns）。
任一引用列未開啟統計，該過濾條件不會被下推。
基於 batch 級統計的粗粒度過濾，可能存在 false positive（多返回資料），查詢結果語義不受影響。

使用樣本

步驟一：建表並開啟列統計

CREATE TABLE sensor_log (
  sensor_id BIGINT,
  temperature DOUBLE,
  humidity DOUBLE,
  event_time TIMESTAMP(3)
) WITH (
  'table.statistics.columns' = 'sensor_id,temperature'
);

步驟二：使用 WHERE 條件查詢

SELECT sensor_id, temperature, event_time
FROM sensor_log
WHERE temperature > 30.0 AND sensor_id = 1001;

步驟三：通過 EXPLAIN 驗證謂詞下推

EXPLAIN SELECT sensor_id, temperature, event_time
FROM sensor_log
WHERE temperature > 30.0 AND sensor_id = 1001;

EXPLAIN 輸出中，TableSourceScan 節點的 filter=[...] 子句表明過濾條件已被下推到 Fluss 服務端。

說明

EXPLAIN 輸出中 Calc 節點仍保留過濾條件是正常行為。Flink 用戶端對接收到的資料再次執行過濾，作為安全兜底。

關於謂詞下推的原理和配置詳情，請參見表特性。

資料查詢

說明

主鍵表與日誌表均支援 LIMIT 讀取，可快速返回指定條數結果，極大便利資料預覽。
主鍵表與日誌表（0.9-ali-1.0 及以上版本）均已支援 COUNT(*) 統計查詢。
主鍵表額外支援基於主鍵和索引鍵的精確匹配的點查功能。

登入Realtime Compute管理主控台。
單擊目標工作空間操作列下的控制台。

在左側導覽列，單擊資料查詢 > 建立查詢指令碼。

SELECT * FROM `fluss-catalog`.`my_db`.`customer` 
WHERE c_custkey = 1;

選中代碼塊，單擊運行即可查看結果。

消費模式

消費模式	主鍵表	日誌表	適用情境
full（預設）	首先消費全量資料，然後再消費增量資料。	從最早的位點開始消費。	適用於需要處理全量資料的情境，如全量資料分析。
earliest	從最早的 binlog 位點開始消費。	從最早的位點開始消費。	適用於需要處理歷史變更資料的情境，如初次載入。
latest	從最新的 binlog 位點開始消費，	從最新的位點開始消費	適用於僅關注最新資料的情境，如即時監控。
timestamp	從指定的時間（由配置項 scan.startup.timestamp 指定）開始消費 binlog。	從指定的時間（由配置項 scan.startup.timestamp 指定）開始消費	適用於基於特定時間點的資料恢複或回溯。

SQL樣本

// 預設full模式
SELECT * FROM my_table ;

// earliest 消費模式
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'earliest') */;

// latest 消費模式
SELECT * FROM my_table /*+ OPTIONS('scan.startup.mode' = 'latest') */;

// 使用時間戳的方式
SELECT * FROM my_table
/*+ OPTIONS('scan.startup.mode' = 'timestamp',
'scan.startup.timestamp' = '1678883047356') */;

// 使用時間字串的方式
SELECT * FROM my_table
/*+ OPTIONS('scan.startup.mode' = 'timestamp',
'scan.startup.timestamp' = '2023-12-09 23:09:12') */;