ApsaraDB for SelectDB:Group Commit

Cara kerja

Group commit bukan metode impor terpisah. Fitur ini mencegat permintaan INSERT INTO VALUES, Stream Load, dan HTTP Stream yang memenuhi syarat, lalu mengelompokkannya menjadi satu commit internal. Auto commit dipicu ketika salah satu ambang batas tercapai:

• Interval commit berakhir (default: 10 detik)

• Ukuran data terakumulasi mencapai batas (default: 64 MB)

Tiga mode mengatur perilaku ini:

Memilih antara mode sync dan async:

• Gunakan sync_mode untuk skenario konkurensi tinggi di mana data harus langsung terlihat setelah impor. Mode ini memblokir hingga transaksi di-commit, sehingga hasil yang dikembalikan menjamin data telah dipersist dan dapat langsung di-query.

• Gunakan async_mode ketika latensi penulisan adalah prioritas tertinggi. Server mengakui impor segera setelah data ditulis ke WAL, tanpa menunggu commit internal selesai. Jika commit internal gagal, log WAL digunakan untuk memulihkan data. Data tidak langsung terlihat setelah impor mengembalikan respons.

Buat tabel contoh

Contoh dalam topik ini menggunakan tabel berikut:

CREATE TABLE `dt` (
    `id` int(11) NOT NULL,
    `name` varchar(50) NULL,
    `score` int(11) NULL
) ENGINE=OLAP
DUPLICATE KEY(`id`)
DISTRIBUTED BY HASH(`id`) BUCKETS 1;

Impor data menggunakan JDBC

ApsaraDB for SelectDB mendukung fitur prepared statement MySQL melalui Java Database Connectivity (JDBC). Saat Anda menggunakan prepared statement, pernyataan SQL dan rencana impornya di-cache dalam memori tingkat session, sehingga mengurangi overhead CPU pada penyisipan berulang.

1. Tambahkan dependensi konektor MySQL ke proyek Anda:

    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.49</version>
    </dependency>

2. Buat URL JDBC dengan prepared statement sisi server diaktifkan:

    jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030/db?useServerPrepStmts=true

3. Atur variabel session group_commit. Gunakan salah satu pendekatan berikut:

   • Tambahkan ke URL JDBC: ``jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030/db?useServerPrepStmts=true&sessionVariables=group_commit=async_mode``

   • Jalankan pernyataan SQL saat koneksi dibuat: ``java try (Statement statement = conn.createStatement()) { statement.execute("SET group_commit = async_mode;"); }``

4. Gunakan prepared statement untuk menyisipkan baris:

    private static final String JDBC_DRIVER = "com.mysql.jdbc.Driver";
    private static final String URL_PATTERN = "jdbc:mysql://%s:%d/%s?useServerPrepStmts=true";
    private static final String HOST = "selectdb-cn-****.selectdbfe.rds.aliyuncs.com";
    private static final int PORT = 9030;
    private static final String DB = "db";
    private static final String TBL = "dt";
    private static final String USER = "admin";
    private static final String PASSWD = "***";
    private static final int INSERT_BATCH_SIZE = 10;
   
    public static void main(String[] args) {
        groupCommitInsert();
        //groupCommitInsertBatch
    }
   
    private static void groupCommitInsert() throws Exception {
        Class.forName(JDBC_DRIVER);
        try (Connection conn = DriverManager.getConnection(String.format(URL_PATTERN, HOST, PORT, DB), USER, PASSWD)) {
            // atur variabel session 'group_commit'
            try (Statement statement = conn.createStatement()) {
                statement.execute("SET group_commit = async_mode;");
            }
   
            String query = "INSERT INTO " + TBL + " VALUES(?, ?, ?)";
            try (PreparedStatement stmt = conn.prepareStatement(query)) {
                for (int i = 0; i < INSERT_BATCH_SIZE; i++) {
                    stmt.setInt(1, i);
                    stmt.setString(2, "name" + i);
                    stmt.setInt(3, i + 10);
                    int result = stmt.executeUpdate();
                    System.out.println("rows: " + result);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
   
    private static void groupCommitInsertBatch() throws Exception {
        Class.forName(JDBC_DRIVER);
        // tambahkan rewriteBatchedStatements=true dan cachePrepStmts=true di URL JDBC
        // atur variabel session melalui sessionVariables=group_commit=async_mode di URL JDBC
        try (Connection conn = DriverManager.getConnection(
                String.format(URL_PATTERN + "&rewriteBatchedStatements=true&cachePrepStmts=true&sessionVariables=group_commit=async_mode", HOST, PORT, DB), USER, PASSWD)) {
   
            String query = "INSERT INTO " + TBL + " VALUES(?, ?, ?)";
            try (PreparedStatement stmt = conn.prepareStatement(query)) {
                for (int j = 0; j < 5; j++) {
                    // 10 baris per insert
                    for (int i = 0; i < INSERT_BATCH_SIZE; i++) {
                        stmt.setInt(1, i);
                        stmt.setString(2, "name" + i);
                        stmt.setInt(3, i + 10);
                        stmt.addBatch();
                    }
                    int[] result = stmt.executeBatch();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

Impor data menggunakan INSERT INTO

Aktifkan group commit dengan mengatur variabel session group_commit sebelum menjalankan pernyataan INSERT INTO.

Mode asinkron — data dikelompokkan dan di-commit di latar belakang:

-- Aktifkan group commit dalam mode async. Nilai default adalah off_mode.
mysql> SET group_commit = async_mode;

-- Label yang dikembalikan diawali dengan "group_commit", yang mengonfirmasi group commit aktif.
mysql> INSERT INTO dt VALUES(1, 'Bob', 90), (2, 'Alice', 99);
Query OK, 2 rows affected (0.05 sec)
{'label':'group_commit_a145ce07f1c972fc-bd2c54597052a9ad', 'status':'PREPARE', 'txnId':'181508'}

-- Insert berturut-turut yang berbagi label dan txnId yang sama dikelompokkan menjadi satu pekerjaan impor.
mysql> INSERT INTO dt(id, name) VALUES(3, 'John');
Query OK, 1 row affected (0.01 sec)
{'label':'group_commit_a145ce07f1c972fc-bd2c54597052a9ad', 'status':'PREPARE', 'txnId':'181508'}

-- Data tidak langsung terlihat setelah impor mengembalikan respons.
mysql> SELECT * FROM dt;
Empty SET (0.01 sec)

-- Setelah ~10 detik (dikontrol oleh group_commit_interval), data menjadi terlihat.
mysql> SELECT * FROM dt;
+------+-------+-------+
| id   | name  | score |
+------+-------+-------+
|    1 | Bob   |    90 |
|    2 | Alice |    99 |
|    3 | John  |  NULL |
+------+-------+-------+
3 rows in set (0.02 sec)

Mode sinkron — hanya mengembalikan respons setelah transaksi di-commit; data langsung terlihat:

-- Aktifkan group commit dalam mode sync.
mysql> SET group_commit = sync_mode;

-- Interval commit dikontrol oleh group_commit_interval. Panggilan ini memblokir hingga transaksi di-commit.
mysql> INSERT INTO dt VALUES(4, 'Bob', 90), (5, 'Alice', 99);
Query OK, 2 rows affected (10.06 sec)
{'label':'group_commit_d84ab96c09b60587_ec455a33cb0e9e87', 'status':'PREPARE', 'txnId':'3007', 'query_id':'fc6b94085d704a94-a69bfc9a202e66e2'}

-- Data langsung terlihat.
mysql> SELECT * FROM dt;
+------+-------+-------+
| id   | name  | score |
+------+-------+-------+
|    1 | Bob   |    90 |
|    2 | Alice |    99 |
|    3 | John  |  NULL |
|    4 | Bob   |    90 |
|    5 | Alice |    99 |
+------+-------+-------+
5 rows in set (0.03 sec)

Nonaktifkan group commit:

mysql> SET group_commit = off_mode;

Impor data menggunakan Stream Load

Untuk pipeline berbasis logging atau HTTP, kirimkan header group_commit untuk mengaktifkan group commit pada permintaan Stream Load. Untuk informasi lebih lanjut tentang Stream Load, lihat Stream Load.

1. Buat file bernama data.csv:

    6,Amy,60
    7,Ross,98

2. Jalankan impor dengan header mode yang sesuai: Mode asinkron:

    # Kirim group_commit:async_mode sebagai header permintaan.
    curl --location-trusted -u {user}:{passwd} -T data.csv \
      -H "group_commit:async_mode" \
      -H "column_separator:," \
      http://{selectdbHost}:{selectdbHttpPort}/api/db/dt/_stream_load

   Tanggapan yang diharapkan:

    {
        "TxnId": 7009,
        "Label": "group_commit_c84d2099208436ab_96e33fda01eddba8",
        "Comment": "",
        "GroupCommit": true,
        "Status": "Success",
        "Message": "OK",
        "NumberTotalRows": 2,
        "NumberLoadedRows": 2,
        "NumberFilteredRows": 0,
        "NumberUnselectedRows": 0,
        "LoadBytes": 19,
        "LoadTimeMs": 35,
        "StreamLoadPutTimeMs": 5,
        "ReadDataTimeMs": 0,
        "WriteDataTimeMs": 26
    }

   Mode sinkron:

    # Kirim group_commit:sync_mode sebagai header permintaan.
    curl --location-trusted -u {user}:{passwd} -T data.csv \
      -H "group_commit:sync_mode" \
      -H "column_separator:," \
      http://{selectdbHost}:{selectdbHttpPort}/api/db/dt/_stream_load

   Tanggapan yang diharapkan:

    {
        "TxnId": 3009,
        "Label": "group_commit_d941bf17f6efcc80_ccf4afdde9881293",
        "Comment": "",
        "GroupCommit": true,
        "Status": "Success",
        "Message": "OK",
        "NumberTotalRows": 2,
        "NumberLoadedRows": 2,
        "NumberFilteredRows": 0,
        "NumberUnselectedRows": 0,
        "LoadBytes": 19,
        "LoadTimeMs": 10044,
        "StreamLoadPutTimeMs": 4,
        "ReadDataTimeMs": 0,
        "WriteDataTimeMs": 10038
    }

   "GroupCommit": true dalam tanggapan mengonfirmasi bahwa group commit aktif. Label selalu diawali dengan group_commit.

Konfigurasi ambang batas auto commit

Sesuaikan interval commit atau ambang batas ukuran data per tabel menggunakan ALTER TABLE.

-- Ubah interval commit menjadi 2 detik.
ALTER TABLE dt SET ("group_commit_interval_ms" = "2000");

-- Ubah ambang batas ukuran data menjadi 128 MB.
ALTER TABLE dt SET ("group_commit_data_bytes" = "134217728");

Batasan

insert table ${table_name} is blocked on schema change