Hologres:Impor Near Real-time Menggunakan Stage Penyimpanan Sementara

Pengenalan Tahap

Stage adalah lapisan penyimpanan sementara internal berkinerja tinggi milik Hologres. Atribut utamanya meliputi hal-hal berikut:

• Hosting internal: Dikelola secara otomatis oleh Hologres tanpa memerlukan konfigurasi atau operasi penyimpanan eksternal.

• Cache terstruktur: Mengorganisasi data sesuai skema tabel target untuk memastikan konsistensi tipe dan format.

• Keamanan transaksi: Mendukung semantik ACID guna menjamin keandalan dan konsistensi selama penulisan.

• Komit gabungan otomatis: Konektor seperti Spark secara otomatis menulis data dari penyimpanan sementara ke penyimpanan utama dalam batch yang dioptimalkan, sehingga meningkatkan efisiensi.

Implementasikan impor near real-time menggunakan Stage melalui alur berikut:

• Tulis data eksternal ke Stage.

• Kelompokkan data yang disimpan di Stage, lalu tulis ke Tabel internal.

• Data yang ditulis ke tabel internal langsung tersedia untuk kueri.

Keunggulan Tahap

• Menyeimbangkan latensi dan throughput: Data dapat dikueri dalam hitungan menit setelah ingestion—jauh lebih cepat daripada impor batch offline—dan mendukung kapasitas throughput yang jauh lebih tinggi dibandingkan penulisan streaming real-time.

• Menyederhanakan tautan data: Tulis data langsung ke Stage internal Hologres tanpa ketergantungan pada sistem eksternal. Integrasi didukung melalui sintaksis SQL COPY dan INSERT standar serta konektor resmi.

• Meningkatkan efisiensi resource: Stage secara otomatis menggabungkan file kecil untuk meningkatkan performa baca-tulis selanjutnya. Stage juga mendukung Serverless Computing dan virtual warehouse independen guna mengisolasi resource penulisan Stage, sehingga meningkatkan stabilitas sistem.

• Mendukung berbagai skenario: Penulisan near real-time ke Stage didukung oleh Spark Connector, Flink Connector, dan Holo Client.

Create Stage

Sintaksis perintah:

-- Buat Internal Stage
-- Ganti variabel:
--   <internal_stage_name>: Nama Stage, wajib, maksimal 128 karakter
--   <group_name>: Nama grup, opsional, default-nya internal_stage_name
--   <ttl_in_seconds>: TTL (detik), opsional, default-nya 7200, maksimal 864000
CALL HOLOGRES.HG_CREATE_INTERNAL_STAGE(
  '<internal_stage_name>', 
  ['<group_name>'],
  ['<ttl_in_seconds>']
);

Deskripsi parameter:

Delete Stage

Sintaksis perintah:

-- Hapus Internal Stage yang ditentukan
-- Ganti variabel: <internal_stage_name> dengan nama Stage yang akan dihapus
CALL HOLOGRES.HG_DROP_INTERNAL_STAGE(
    '<internal_stage_name>'
);

internal_stage_name: Wajib. Nama Internal Stage.

Query Stage Status

Kueri status Stage menggunakan tampilan sistem hologres.hg_internal_stages:

-- Kueri semua Stage atau filter berdasarkan nama
-- Ganti variabel: Jika <internal_stage_name> diganti dengan nama Stage, hanya Stage tersebut yang dikueri; hapus WHERE untuk mengkueri semua.
SELECT * FROM hologres.hg_internal_stages 
[WHERE stage_name = '<internal_stage_name>'];

Tampilan tersebut mencakup bidang-bidang berikut:

Query Stage Files

Kueri file Stage menggunakan tampilan sistem hologres.hg_internal_stage_files:

-- Kueri daftar file di bawah Stage, dapat difilter berdasarkan pencocokan kabur pada nama
-- Ganti variabel: <internal_stage_name> adalah nama Stage; <pattern%> adalah pencocokan kabur untuk nama file (misalnya 'batch_%')
SELECT * FROM hologres.hg_internal_stage_files 
[WHERE stage_name = '<internal_stage_name>'] 
  [AND file_name like '<pattern%>'];

Tampilan tersebut mencakup bidang-bidang berikut:

Delete Stage Files

Hapus file Stage menggunakan fungsi sistem hologres.hg_remove_internal_stage_file:

-- Hapus file yang ditentukan
-- Ganti variabel: <stage_name> adalah nama Stage; <file_name> adalah nama file yang akan dihapus
SELECT hologres.hg_remove_internal_stage_file ('<stage_name>', '<file_name>');

-- Hapus file Stage secara batch
-- Ganti variabel: <stage_name> adalah nama Stage; <glob_pattern> adalah wildcard nama file (misalnya '*.arrow'); <pattern%> adalah pencocokan kabur untuk nama file di 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%>'];

Client Write to Stage

Sintaksis perintah:

-- Tulis aliran data klien ke file yang ditentukan di Stage
-- Ganti variabel: <internal_stage_name> adalah nama Stage yang telah dibuat; <file_name> adalah nama file yang akan ditulis (mendukung huruf, angka, garis bawah, tanda hubung, titik, maksimal 128 karakter)
COPY EXTERNAL_FILES(
  path = 'internal_stage://<internal_stage_name>/<file_name>'
) FROM STDIN;

Saat menulis ke Stage menggunakan perintah COPY, Anda tidak dapat mengonfigurasi parameter WITH. Sebagai gantinya, definisikan path file Stage menggunakan parameter path. Karakter yang valid meliputi huruf (case-sensitive), angka, garis bawah, tanda hubung, dan titik, dengan panjang maksimum 128 karakter.

Read from Stage and Write to Internal Tables

Sintaksis perintah:

-- Baca file Arrow dari Stage dan tulis ke tabel internal
-- Ganti variabel:
--   <table_name>: Nama tabel internal target
--   <col_name>: Nama kolom (opsional, berkorespondensi satu-satu dengan kolom tabel target jika ditentukan)
--   <internal_stage_name>: Nama Stage. Tentukan beberapa nama yang dipisahkan koma untuk membaca dari beberapa Stage.
--   <col_type>: Tipe kolom (opsional, gunakan hanya saat menentukan tipe kolom di klausa 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> ... ] )];

Membaca dari Stage hanya mendukung file format Arrow.

Write to Stage Using Holo-client

Contoh ini menunjukkan cara melakukan penulisan near real-time ke Stage menggunakan Holo-client.

Dependensi Maven:

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

Kode contoh:

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 {
        // Catatan: jdbcUrl harus menggunakan protokol 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");

        // Buat nama Stage sementara
        String stageName = "temp_stage_" + System.currentTimeMillis();

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

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

            // Dapatkan skema tabel
            TableSchema schema = client.getTableSchema(tableName);

            // Definisikan kolom yang akan ditulis
            List<String> columns = new ArrayList<>();
            columns.add("id");
            columns.add("name");
            columns.add("address");

            // Gunakan RecordArrowWriter dan CopyInStageWrapper untuk menulis data ke Stage
            try (RecordArrowWriter arrowWriter =
                         new RecordArrowWriter(
                                 schema,
                                 columns,
                                 8192 // maxBatchSize, 1024 baris data membentuk satu Arrow RecordBatch
                         );
                 CopyInStageWrapper<com.alibaba.hologres.client.model.Record> copyInStage =
                         new CopyInStageWrapper<>(
                                 config,
                                 stageName,
                                 "data_file", // Awalan nama file
                                 arrowWriter,
                                 64 * 1024 * 1024 // fileSizeLimit, ukuran tiap file 64 MB
                         )) {

                // Tulis 10 catatan
                for (int i = 0; i < 10; ++i) {
                    Put put = new Put(schema);
                    // Sesuaikan dengan kolom di CopyInStageWrapper
                    put.setObject("id", i);
                    put.setObject("name", "name" + i);
                    put.setObject("address", "address" + i);

                    copyInStage.putRecord(put.getRecord());
                }
                // Panggil close sebelum program berakhir untuk memastikan data sepenuhnya ditulis
                // Demo ini menggunakan try-with-resources, tidak perlu menutup secara manual
                // copyInStage.close();
            }

            // Hasilkan pernyataan INSERT untuk menulis dari Stage ke tabel target
            String insertSql =
                    CopyUtil.buildInsertTableSelectFromStageSql(
                            schema,
                            columns,
                            Collections.singletonList(stageName),
                            OnConflictAction.INSERT_OR_UPDATE);

            try (java.sql.Statement stmt = conn.createStatement()) {
                stmt.execute(insertSql);
            }
            // Verifikasi hasil penulisan
            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 {
            // Bersihkan Stage sementara. Jika tidak dibersihkan, akan dibersihkan otomatis berdasarkan 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("Error membersihkan Stage: " + e.getMessage());
            }
        }
    }
}