ApsaraDB for SelectDB:Group Commit

Fonctionnement

Le group commit n'est pas une méthode d'importation distincte. Il intercepte les requêtes INSERT INTO VALUES, Stream Load et HTTP Stream éligibles pour les regrouper en un seul commit interne. Le commit automatique se déclenche dès que l'un des seuils suivants est atteint :

• L'intervalle de commit expire (par défaut : 10 secondes).

• La taille cumulée des données atteint la limite (par défaut : 64 Mo).

Trois modes régissent ce comportement :

Choix entre le mode synchrone et asynchrone :

• Privilégiez sync_mode pour les scénarios à forte concurrence où les données doivent être visibles immédiatement après l'importation. Ce mode bloque l'exécution jusqu'au commit de la transaction ; le résultat retourné confirme ainsi que les données sont persistées et interrogeables sans délai.

• Optez pour async_mode lorsque la latence d'écriture constitue la priorité absolue. Le serveur accuse réception de l'importation dès l'écriture des données dans le WAL, sans attendre la fin du commit interne. En cas d'échec de ce commit, les journaux WAL permettent de récupérer les données. Celles-ci ne sont pas visibles immédiatement après le retour de l'importation.

Création de la table d'exemple

Les exemples de cette rubrique utilisent la table suivante :

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;

Importation de données via JDBC

ApsaraDB for SelectDB prend en charge la fonctionnalité de requête préparée de MySQL via Java Database Connectivity (JDBC). L'utilisation de requêtes préparées permet de mettre en cache les instructions SQL et leurs plans d'importation dans la mémoire au niveau de la session, réduisant ainsi la surcharge CPU lors d'insertions répétées.

1. Ajoutez la dépendance du connecteur MySQL à votre projet :

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

2. Construisez l'URL JDBC avec l'activation des requêtes préparées côté serveur :

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

3. Définissez la variable de session group_commit. Utilisez l'une des approches suivantes :

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

   • Exécutez une instruction SQL lors de la connexion : ``java try (Statement statement = conn.createStatement()) { statement.execute("SET group_commit = async_mode;"); } ``

4. Utilisez une requête préparée pour insérer des lignes :

    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();
        }
    }

Importation de données avec INSERT INTO

Activez le group commit en définissant la variable de session group_commit avant d'exécuter des instructions INSERT INTO.

Mode asynchrone — les données sont regroupées et validées en arrière-plan :

-- 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)

Mode synchrone — la réponse n'est renvoyée qu'après le commit de la transaction ; les données sont immédiatement visibles :

-- 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)

Désactivation du group commit :

mysql> SET group_commit = off_mode;

Importation de données avec Stream Load

Pour les pipelines basés sur des journaux ou HTTP, transmettez l'en-tête group_commit afin d'activer le group commit sur les requêtes Stream Load. Pour plus d'informations sur Stream Load, consultez Stream Load.

1. Créez un fichier nommé data.csv :

    6,Amy,60
    7,Ross,98

2. Exécutez l'importation avec l'en-tête correspondant au mode souhaité : Mode asynchrone :

    # 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

   Réponse attendue :

    {
        "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 synchrone :

    # 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

   Réponse attendue :

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

   La présence de "GroupCommit": true dans la réponse confirme que le group commit est actif. Le libellé commence toujours par group_commit.

Configuration des seuils de commit automatique

Ajustez l'intervalle de commit ou les seuils de taille des données par table à l'aide de 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");

Limites

insert table ${table_name} is blocked on schema change