ApsaraDB for SelectDB:Use the group commit feature to import data

Use JDBC

If you execute the INSERT INTO VALUES statement to import data by using JDBC, ApsaraDB for SelectDB supports the prepared statement feature of MySQL to reduce the overhead of SQL parsing and plan generation. If you use prepared statements, the SQL statements and their import plans are cached in the session-level memory cache. The cached objects can be used for subsequent import. This reduces the CPU utilization of your cluster. In the following example, a prepared statement and JDBC are used to import data.

1. Add the dependency.

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

2. Specify the JDBC URL and enable the prepared statement feature on the server.

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

3. Configure the session variable group_commit by using either of the following methods:

   • Add sessionVariables=group_commit=async_mode to the JDBC URL.

     url = jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030/db?useServerPrepStmts=true&sessionVariables=group_commit=async_mode

   • Execute the following SQL statement:

     try (Statement statement = conn.createStatement()) {
         statement.execute("SET group_commit = async_mode;");
     }

4. Use a prepared statement.

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

Use INSERT INTO

This section describes how to insert data by executing the INSERT INTO statement in asynchronous and synchronous mode:

• Asynchronous mode

  -- Configure the session variable group_commit to enable the group commit feature in asynchronous mode. The default value of the variable is off_mode. 
  mysql> SET group_commit = async_mode;
  
  -- The label that is returned starts with group_commit. This indicates the group commit feature is enabled. 
  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'}
  
  -- The label and txnId for the following statement are the same as those for the preceding statement. This indicates that the two statements are executed in the same 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'}
  
  -- The imported data cannot be queried immediately after the import is complete. 
  mysql> SELECT * FROM dt;
  Empty SET (0.01 sec)
  
  -- Execute a query after 10 seconds. The imported data can be queried. You can use the table property group_commit_interval to control the latency of data visibility. 
  mysql> SELECT * FROM dt;
  +------+-------+-------+
  | id   | name  | score |
  +------+-------+-------+
  |    1 | Bob   |    90 |
  |    2 | Alice |    99 |
  |    3 | John  |  NULL |
  +------+-------+-------+
  3 rows in set (0.02 sec)

• Synchronous mode

  -- Configure the session variable group_commit to enable the group commit feature in synchronous mode. The default value of the variable is off_mode. 
  mysql> SET group_commit = sync_mode;
  
  -- The commit interval is specified by the table property group_commit_interval. The label that is returned starts with group_commit. This indicates that the group commit feature is enabled. 
  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'}
  
  -- The imported data can be queried immediately after the import is complete. 
  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)

• Disable the group commit feature

  mysql> SET group_commit = off_mode;

Use Stream Load

For more information about the Stream Load feature, see Stream Load.

1. Create a file named data.csv and add the following content to the file:

   6,Amy,60
   7,Ross,98

2. Use Stream Load to import data in asynchronous or synchronous mode:

   • Asynchronous mode

     # Add the group_commit:async_mode header to the URL for the import. 
     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
     {
         "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
     }
     
     # The value of the GroupCommit parameter that is returned is true. This indicates that the group commit takes effect. 
     # The label that is associated with this import starts with group_commit.

   • Synchronous mode

     # Add the group_commit:sync_mode header to the URL for the import. 
     
     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
     {
         "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
     }
     
     # The value of the GroupCommit parameter that is returned is true. This indicates that the group commit takes effect. 
     # The label that is associated with this import starts with group_commit.