Hologres:一時ストレージ Stage を使用した準リアルタイムインポート

Stage の概要

Stage は、Hologres の内部にあるパフォーマンス専有型の一時ストレージレイヤーです。その主な属性は次のとおりです：

• 内部ホスティング：Hologres によって自動的に管理されます。外部ストレージの構成や操作は必要ありません。

• 構造化キャッシュ：ターゲットテーブルのスキーマに従ってデータを整理し、型とフォーマットの一貫性を確保します。

• トランザクションの安全性：ACID セマンティクスをサポートし、書き込み中の信頼性と一貫性を保証します。

• 自動マージコミット：Spark などのコネクタは、一時ストレージからメインストレージに最適化されたバッチでデータを自動的に書き込み、効率を向上させます。

Stage を使用した準リアルタイムインポートは、次のフローで実装します：

• 外部データを Stage に書き込みます。

• Stage に保存されたデータをバッチ処理し、内部テーブルに書き込みます。

• 内部テーブルに書き込まれたデータは、すぐにクエリで可視になります。

Stage の利点

• レイテンシーとスループットのバランス：データは取り込み後、数分以内にクエリ可能になり、オフラインバッチインポートよりも大幅に高速です。また、リアルタイムストリーミング書き込みよりもはるかに高いスループット容量をサポートします。

• データリンクの簡素化：外部システムに依存せず、データを Hologres の内部 Stage に直接書き込みます。標準の SQL COPY および INSERT 構文、および公式コネクタを介した統合がサポートされています。

• リソース効率の向上：Stage は小さなファイルを自動的にマージし、後続の読み取りおよび書き込みパフォーマンスを向上させます。サーバーレスコンピューティングと独立した仮想ウェアハウスをサポートし、Stage の書き込みリソースを分離することで、システムの安定性を向上させます。

• 幅広いシナリオをサポート：Spark コネクタ、Flink コネクタ、および Holo Client によって、Stage への準リアルタイム書き込みがサポートされています。

Stage の作成

コマンド構文：

-- 内部 Stage を作成します
-- 変数を置き換えます：
--   <internal_stage_name>：Stage 名、必須、最大 128 文字
--   <group_name>：グループ名、オプション、デフォルトは internal_stage_name
--   <ttl_in_seconds>：TTL (秒)、オプション、デフォルトは 7200、最大 864000
CALL HOLOGRES.HG_CREATE_INTERNAL_STAGE(
  '<internal_stage_name>', 
  ['<group_name>'],
  ['<ttl_in_seconds>']
);

パラメーターの説明：

Stage の削除

コマンド構文：

-- 指定された内部 Stage を削除します
-- 変数を置き換えます：<internal_stage_name> を削除する Stage 名に置き換えます
CALL HOLOGRES.HG_DROP_INTERNAL_STAGE(
    '<internal_stage_name>'
);

internal_stage_name：必須。内部 Stage の名前です。

Stage ステータスのクエリ

システムビュー hologres.hg_internal_stages を使用してクエリ ステージのステータスを照会します：

-- すべての Stage をクエリするか、名前でフィルターします
-- 変数を置き換えます：<internal_stage_name> を Stage 名に置き換えると、その Stage のみをクエリします。WHERE を省略すると、すべてをクエリします。
SELECT * FROM hologres.hg_internal_stages 
[WHERE stage_name = '<internal_stage_name>'];

このビューには次のフィールドが含まれます：

クエリ ステージファイル

システムビュー 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_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 に書き込む場合、WITH パラメーターは構成できません。代わりに、path パラメーターを使用して Stage ファイルのパスを定義します。有効な文字には、文字 (大文字と小文字を区別)、数字、アンダースコア、ハイフン、ピリオドが含まれます。最大長は 128 文字です。

Stage からの読み取りと内部テーブルへの書き込み

コマンド構文：

-- Stage から Arrow ファイルを読み取り、内部テーブルに書き込みます
-- 変数を置き換えます：
--   <table_name>：ターゲットの内部テーブル名
--   <col_name>：列名 (オプション、指定した場合、ターゲットテーブルの列と 1 対 1 で対応)
--   <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 行のデータで 1 つの 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 の列と一致させます
                    put.setObject("id", i);
                    put.setObject("name", "name" + i);
                    put.setObject("address", "address" + i);

                    copyInStage.putRecord(put.getRecord());
                }
                // プログラムが終了する前に close を呼び出して、データが完全に書き込まれるようにします
                // このデモでは 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());
            }
        }
    }
}