ApsaraDB for SelectDB：Gunakan fitur group commit untuk mengimpor data

Gunakan JDBC

Jika Anda mengeksekusi pernyataan INSERT INTO VALUES untuk mengimpor data menggunakan JDBC, ApsaraDB for SelectDB mendukung fitur prepared statement MySQL untuk mengurangi overhead penguraian SQL dan pembuatan rencana. Jika Anda menggunakan prepared statements, pernyataan SQL dan rencana impornya disimpan dalam cache memori tingkat sesi. Objek yang di-cache dapat digunakan untuk impor selanjutnya, mengurangi penggunaan CPU kluster Anda. Dalam contoh berikut, prepared statement dan JDBC digunakan untuk mengimpor data.

1. Tambahkan dependensi.

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

2. Tentukan URL JDBC dan aktifkan fitur prepared statement di server.

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

3. Konfigurasikan variabel sesi group_commit menggunakan salah satu dari metode berikut:

   • Tambahkan sessionVariables=group_commit=async_mode ke URL JDBC.

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

   • Eksekusi pernyataan SQL berikut:

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

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

Gunakan INSERT INTO

Bagian ini menjelaskan cara menyisipkan data dengan mengeksekusi pernyataan INSERT INTO dalam mode asinkron dan sinkron:

• Mode asinkron

  -- Konfigurasikan variabel sesi group_commit untuk mengaktifkan fitur group commit dalam mode asinkron. Nilai default variabel ini adalah off_mode. 
  mysql> SET group_commit = async_mode;
  
  -- Label yang dikembalikan dimulai dengan group_commit. Ini menunjukkan bahwa fitur group commit diaktifkan. 
  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'}
  
  -- Label dan txnId untuk pernyataan berikutnya sama dengan yang untuk pernyataan sebelumnya. Ini menunjukkan bahwa kedua pernyataan dieksekusi dalam pekerjaan impor yang sama. 
  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 yang diimpor tidak dapat diquery segera setelah impor selesai. 
  mysql> SELECT * FROM dt;
  Empty SET (0.01 sec)
  
  -- Eksekusi query setelah 10 detik. Data yang diimpor dapat diquery. Anda dapat menggunakan properti tabel group_commit_interval untuk mengontrol latensi visibilitas data. 
  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

  -- Konfigurasikan variabel sesi group_commit untuk mengaktifkan fitur group commit dalam mode sinkron. Nilai default variabel ini adalah off_mode. 
  mysql> SET group_commit = sync_mode;
  
  -- Interval komit ditentukan oleh properti tabel group_commit_interval. Label yang dikembalikan dimulai dengan group_commit. Ini menunjukkan bahwa fitur group commit diaktifkan. 
  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 yang diimpor dapat diquery segera setelah impor selesai. 
  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 fitur group commit

  mysql> SET group_commit = off_mode;

Gunakan Stream Load

Untuk informasi lebih lanjut tentang fitur Stream Load, lihat Stream Load.

1. Buat file bernama data.csv dan tambahkan konten berikut ke file:

   6,Amy,60
   7,Ross,98

2. Gunakan Stream Load untuk mengimpor data dalam mode asinkron atau sinkron:

   • Mode asinkron

     # Tambahkan header group_commit:async_mode ke URL untuk impor. 
     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
     }
     
     # Nilai parameter GroupCommit yang dikembalikan adalah true. Ini menunjukkan bahwa group commit berlaku. 
     # Label yang terkait dengan impor ini dimulai dengan group_commit.

   • Mode sinkron

     # Tambahkan header group_commit:sync_mode ke URL untuk impor. 
     
     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
     }
     
     # Nilai parameter GroupCommit yang dikembalikan adalah true. Ini menunjukkan bahwa group commit berlaku. 
     # Label yang terkait dengan impor ini dimulai dengan group_commit.