ApsaraDB for SelectDB:Group Commit

Como funciona

O group commit não é um método de importação separado. Ele intercepta requisições qualificadas de INSERT INTO VALUES, Stream Load e HTTP Stream e as agrupa em um único commit interno. O commit automático é acionado quando qualquer um dos seguintes limites é atingido:

• O intervalo de commit expira (padrão: 10 segundos)

• O tamanho acumulado dos dados atinge o limite (padrão: 64 MB)

Três modos controlam esse comportamento:

Escolha entre modo síncrono e assíncrono:

• Utilize sync_mode em cenários de alta concorrência onde os dados devem estar visíveis logo após a importação. Esse modo bloqueia a execução até que a transação seja confirmada, garantindo que o resultado retornado confirme a persistência e a disponibilidade imediata para consulta.

• Prefira async_mode quando a latência de escrita for a maior prioridade. O servidor confirma a importação assim que os dados são gravados no WAL, sem aguardar a conclusão do commit interno. Caso o commit interno falhe, os logs do WAL recuperam os dados. A visibilidade dos dados não é imediata após o retorno da importação.

Crie a tabela de exemplo

Os exemplos neste tópico utilizam a seguinte tabela:

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;

Importe dados usando JDBC

O ApsaraDB for SelectDB suporta o recurso de prepared statement do MySQL via Java Database Connectivity (JDBC). Ao utilizar prepared statements, as instruções SQL e seus planos de importação são armazenados em cache na memória da sessão, reduzindo a sobrecarga de CPU em inserções repetidas.

1. Adicione a dependência do conector MySQL ao seu projeto:

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

2. Construa a URL JDBC com prepared statements no lado do servidor ativados:

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

3. Defina a variável de sessão group_commit. Escolha uma das abordagens:

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

   • Execute uma instrução SQL no momento da conexão: ``java try (Statement statement = conn.createStatement()) { statement.execute("SET group_commit = async_mode;"); } ``

4. Use um prepared statement para inserir linhas:

    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)) {
            // set session variable '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);
        // add rewriteBatchedStatements=true and cachePrepStmts=true in JDBC url
        // set session variables by sessionVariables=group_commit=async_mode in JDBC url
        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 rows 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();
        }
    }

Importe dados usando INSERT INTO

Ative o group commit definindo a variável de sessão group_commit antes de executar instruções INSERT INTO.

Modo assíncrono — os dados são agrupados e confirmados em segundo plano:

-- Enable group commit in async mode. The default value is off_mode.
mysql> SET group_commit = async_mode;

-- The returned label starts with "group_commit", confirming group commit is active.
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'}

-- Consecutive inserts sharing the same label and txnId are batched into one import job.
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 is not visible immediately after the import returns.
mysql> SELECT * FROM dt;
Empty SET (0.01 sec)

-- After ~10 seconds (controlled by group_commit_interval), data becomes visible.
mysql> SELECT * FROM dt;
+------+-------+-------+
| id   | name  | score |
+------+-------+-------+
|    1 | Bob   |    90 |
|    2 | Alice |    99 |
|    3 | John  |  NULL |
+------+-------+-------+
3 rows in set (0.02 sec)

Modo síncrono — retorna apenas após o commit da transação; os dados ficam visíveis imediatamente:

-- Enable group commit in sync mode.
mysql> SET group_commit = sync_mode;

-- The commit interval is controlled by group_commit_interval. The call blocks until the transaction commits.
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 is visible immediately.
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)

Desative o group commit:

mysql> SET group_commit = off_mode;

Importe dados usando Stream Load

Para pipelines baseados em logs ou HTTP, passe o cabeçalho group_commit para ativar o group commit nas requisições de Stream Load. Para mais informações sobre Stream Load, consulte Stream Load.

1. Crie um arquivo chamado data.csv:

    6,Amy,60
    7,Ross,98

2. Execute a importação com o cabeçalho do modo apropriado: Modo assíncrono:

    # Pass group_commit:async_mode as a request header.
    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

   Resposta esperada:

    {
        "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
    }

   Modo síncrono:

    # Pass group_commit:sync_mode as a request header.
    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

   Resposta esperada:

    {
        "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
    }

   O valor "GroupCommit": true na resposta confirma que o group commit está ativo. O rótulo sempre começa com group_commit.

Configure os limites de commit automático

Ajuste o intervalo de commit ou os limites de tamanho de dados por tabela usando ALTER TABLE.

-- Change the commit interval to 2 seconds.
ALTER TABLE dt SET ("group_commit_interval_ms" = "2000");

-- Change the data size threshold to 128 MB.
ALTER TABLE dt SET ("group_commit_data_bytes" = "134217728");

Limitações

insert table ${table_name} is blocked on schema change