Hologres：通過臨時儲存Stage近即時匯入

Stage 介紹

Stage 是 Hologres 內部的高效能臨時儲存，具有以下核心特性：

• 內部託管：由 Hologres 自動管理，無需使用者操作外部儲存。

• 結構化緩衝：按目標表的 Schema 組織資料，確保類型和格式一致性。

• 事務安全：支援 ACID 語義，保障寫入過程的可靠性與一致性。

• 自動合并提交：Spark 等 Connector 支援自動將臨時儲存以最佳化批次方式寫入主儲存，提升效率。

通過以下流程實現 Stage 近即時匯入：

• 外部資料寫入 Stage。

• Stage 資料攢批寫入內表。

• 內表資料寫入即可見。

Stage 優勢

• 平衡延遲與吞吐：資料從寫入到可查控制在分鐘級，顯著優於離線大量匯入，同時支援遠高於即時資料流式寫入的吞吐能力。

• 簡化資料鏈路：資料直接寫入 Hologres 內部 Stage，無外部系統依賴；通過標準 SQL 的 COPY、INSERT 文法及官方 Connector 適配。

• 提升資源效率：Stage 小檔案自動合并，提升後續讀寫效能；支援 Serverless Computing、獨立計算群組實現 Stage 寫入資源隔離，提升系統穩定性。

• 適用情境廣泛：Spark Connector、Flink Connector、Holo Client 均支援 Stage 近即時寫入模式。

建立 Stage

命令格式：

-- 建立 Internal Stage
-- 可替換變數：
--   <internal_stage_name>：Stage 名稱，必填，最大 128 字元
--   <group_name>：分組名，可選，預設與 internal_stage_name 相同
--   <ttl_in_seconds>：生命週期（秒），可選，預設 7200，最大 864000
CALL HOLOGRES.HG_CREATE_INTERNAL_STAGE(
  '<internal_stage_name>', 
  ['<group_name>'],
  ['<ttl_in_seconds>']
);

參數說明：

刪除 Stage

命令格式：

-- 刪除指定 Internal Stage
-- 可替換變數：<internal_stage_name> 替換為要刪除的 Stage 名稱
CALL HOLOGRES.HG_DROP_INTERNAL_STAGE(
    '<internal_stage_name>'
);

internal_stage_name：必填。Internal Stage 名稱。

查詢 Stage 狀態

可通過系統檢視表 hologres.hg_internal_stages 查詢 Stage 狀態：

-- 查詢全部 Stage 或按名稱過濾
-- 可替換變數：<internal_stage_name> 替換為 Stage 名稱時僅查詢該 Stage；省略 WHERE 則查詢全部
SELECT * FROM hologres.hg_internal_stages 
[WHERE stage_name = '<internal_stage_name>'];

包含如下欄位：

查詢 Stage 檔案

可通過系統檢視表 hologres.hg_internal_stage_files 查詢 Stage 檔案：

-- 查詢 Stage 下檔案清單，可按名稱模糊過濾
-- 可替換變數：<internal_stage_name> Stage 名稱；<pattern%> 檔案名稱模糊比對（如 'batch_%'）
SELECT * FROM hologres.hg_internal_stage_files 
[WHERE stage_name = '<internal_stage_name>'] 
  [AND file_name like '<pattern%>'];

包含如下欄位：

刪除 Stage 檔案

可通過系統函數 hologres.hg_remove_internal_stage_file 刪除 Stage 檔案：

-- 刪除指定檔案
-- 可替換變數：<stage_name> Stage 名稱；<file_name> 要刪除的檔案名稱
SELECT hologres.hg_remove_internal_stage_file ('<stage_name>', '<file_name>');

-- 大量刪除 Stage 檔案
-- 可替換變數：<stage_name> Stage 名稱；<glob_pattern> 檔案名稱通配（如 '*.arrow'）；<pattern%> WHERE 中的檔案名稱模糊比對
SELECT
    stage_name,
    file_name,
    hologres.hg_remove_internal_stage_file (stage_name, file_name) AS hg_remove_internal_stage_file
FROM
    hologres.hg_list_internal_stage_files ('<stage_name>',['<glob_pattern>']) 
[WHERE file_name like '<pattern%>'];

用戶端寫入 Stage

命令格式：

-- 將用戶端資料流寫入 Stage 指定檔案
-- 可替換變數：<internal_stage_name> 已建立的 Stage 名稱；<file_name> 寫入的檔案名稱（支援字母、數字、底線、連字號、英文句號，最大 128 字元）
COPY EXTERNAL_FILES(
  path = 'internal_stage://<internal_stage_name>/<file_name>'
) FROM STDIN;

通過 COPY 命令寫入 Stage 時不支援配置 COPY 文法中的 WITH 參數。通過 path 參數定義 Stage 檔案路徑，支援字母（大小寫敏感）、數字、底線、連字號、英文句號，最大長度 128 字元。

讀取 Stage 寫入內表

命令格式：

-- 從 Stage 讀取 Arrow 檔案並寫入內表
-- 可替換變數：
--   <table_name> 目標內表名
--   <col_name> 列名（可選，指定時與目標表列一一對應）
--   <internal_stage_name> Stage 名稱，可寫多個用逗號分隔以讀取多個 Stage
--   <col_type> 列類型（可選，僅在 AS 子句中指定列類型時使用）
INSERT INTO <table_name> [ ( <col_name> [ , <col_name> ... ] ) ]
SELECT *
FROM EXTERNAL_FILES(
  path = 'internal_stage://<internal_stage_name>, internal_stage://<internal_stage_name>',
  format = arrow
)
[AS ( <col_name> <col_type>[ , <col_name> <col_type> ... ] )];

讀取 Stage 僅支援 Arrow 格式檔案。

通過 Holo-client 寫入 Stage

以下樣本示範如何使用 Holo-client 通過 Stage 進行近即時寫入。

Maven 依賴：

<dependency>
  <groupId>com.alibaba.hologres</groupId>
  <artifactId>holo-client</artifactId>
  <version>2.7.0</version>
</dependency>

範例程式碼：

import com.alibaba.hologres.client.HoloClient;
import com.alibaba.hologres.client.HoloConfig;
import com.alibaba.hologres.client.Put;
import com.alibaba.hologres.client.copy.CopyUtil;
import com.alibaba.hologres.client.copy.in.CopyInStageWrapper;
import com.alibaba.hologres.client.copy.in.arrow.RecordArrowWriter;
import com.alibaba.hologres.client.model.OnConflictAction;
import com.alibaba.hologres.client.model.TableSchema;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class CopyStageDemo {

    public static void main(String[] args) throws Exception {
        // 注意：jdbcUrl需使用jdbc:hologres協議
        String jdbcUrl = "jdbc:hologres://host:port/db";
        String username = "";
        String password = "";

        /*
        CREATE TABLE copy_stage_demo (id INT NOT NULL, name TEXT NOT NULL, address TEXT, PRIMARY KEY(id));
        */
        String tableName = "copy_stage_demo";

        HoloConfig config = new HoloConfig();
        config.setJdbcUrl(jdbcUrl);
        config.setUsername(username);
        config.setPassword(password);
        config.setRegion("local");

        // 建立臨時Stage名稱
        String stageName = "temp_stage_" + System.currentTimeMillis();

        try (Connection conn = DriverManager.getConnection(jdbcUrl, username, password);
             HoloClient client = new HoloClient(config)) {

            // 建立內部Stage
            String createStageSql =
                    "call hologres.hg_create_internal_stage('"
                            + stageName
                            + "', 'default_group', 7200);";
            try (java.sql.Statement stmt = conn.createStatement()) {
                stmt.execute(createStageSql);
            }

            // 擷取表結構
            TableSchema schema = client.getTableSchema(tableName);

            // 定義要寫入的列
            List<String> columns = new ArrayList<>();
            columns.add("id");
            columns.add("name");
            columns.add("address");

            // 使用RecordArrowWriter和CopyInStageWrapper寫入資料到Stage
            try (RecordArrowWriter arrowWriter =
                         new RecordArrowWriter(
                                 schema,
                                 columns,
                                 8192 // maxBatchSize，每1024行資料群組成一個Arrow RecordBatch
                         );
                 CopyInStageWrapper<com.alibaba.hologres.client.model.Record> copyInStage =
                         new CopyInStageWrapper<>(
                                 config,
                                 stageName,
                                 "data_file", // 檔案名稱首碼
                                 arrowWriter,
                                 64 * 1024 * 1024 // fileSizeLimit，每個檔案大小64 MB
                         )) {

                // 寫入10條資料
                for (int i = 0; i < 10; ++i) {
                    Put put = new Put(schema);
                    // 和CopyInStageWrapper的columns保持一致
                    put.setObject("id", i);
                    put.setObject("name", "name" + i);
                    put.setObject("address", "address" + i);

                    copyInStage.putRecord(put.getRecord());
                }
                // 程式結束之前需要調用close，保證資料完全寫入
                // demo使用了try-with-resources，無需手動close
                // copyInStage.close();
            }

            // 產生從Stage寫入目標表的INSERT語句
            String insertSql =
                    CopyUtil.buildInsertTableSelectFromStageSql(
                            schema,
                            columns,
                            Collections.singletonList(stageName),
                            OnConflictAction.INSERT_OR_UPDATE);

            try (java.sql.Statement stmt = conn.createStatement()) {
                stmt.execute(insertSql);
            }
            // 驗證寫入結果
            try (java.sql.Statement stmt = conn.createStatement()) {
                try (java.sql.ResultSet rs = stmt.executeQuery("select * from " + tableName)) {
                    while (rs.next()) {
                        System.out.println(
                                "id: "
                                        + rs.getInt(1)
                                        + ", name: "
                                        + rs.getString(2)
                                        + ", address: "
                                        + rs.getString(3));
                    }
                }
            }
        } finally {
            // 清理臨時Stage，不清理的話也會根據TTL自動清理
            try (Connection conn = DriverManager.getConnection(jdbcUrl, username, password);
                 java.sql.Statement stmt = conn.createStatement()) {
                String dropStageSql = "call hologres.hg_drop_internal_stage('" + stageName + "');";
                stmt.execute(dropStageSql);
            } catch (Exception e) {
                System.err.println("清理Stage時出錯: " + e.getMessage());
            }
        }
    }
}