ApsaraDB for SelectDB:グループコミット機能を使用してデータをインポートする

JDBC を使用する

JDBC を使用して INSERT INTO VALUES 文を実行してデータをインポートする場合、ApsaraDB for SelectDB は MySQL の prepared statement 機能をサポートしており、SQL 解析とプラン生成のオーバーヘッドを削減します。prepared statements を使用すると、SQL 文とそのインポートプランはセッションレベルのメモリキャッシュにキャッシュされます。キャッシュされたオブジェクトは、後続のインポートに使用できます。これにより、クラスタの CPU 使用率が削減されます。次の例では、prepared statement と JDBC を使用してデータをインポートします。

1. 依存関係を追加します。

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

2. JDBC URL を指定し、サーバーで prepared statement 機能を有効にします。

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

3. 次のいずれかの方法でセッション変数 group_commit を設定します。

   • sessionVariables=group_commit=async_mode を JDBC URL に追加します。

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

   • 次の SQL 文を実行します。

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

4. 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)) {
               // セッション変数 '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);
           // JDBC url に rewriteBatchedStatements=true と cachePrepStmts=true を追加する
           // JDBC url の sessionVariables=group_commit=async_mode でセッション変数を設定する
           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 行
                       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();
           }
       }

INSERT INTO を使用する

このセクションでは、非同期モードと同期モードで INSERT INTO 文を実行してデータを挿入する方法について説明します。

• 非同期モード

  -- セッション変数 group_commit を設定して、非同期モードでグループコミット機能を有効にします。変数のデフォルト値は off_mode です。
  mysql> SET group_commit = async_mode;
  
  -- 返されるラベルは group_commit で始まります。これは、グループコミット機能が有効になっていることを示します。
  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'}
  
  -- 次の文のラベルと txnId は、前の文と同じです。これは、2 つの文が同じインポートジョブで実行されていることを示します。
  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'}
  
  -- インポートされたデータは、インポート完了直後にクエリできません。
  mysql> SELECT * FROM dt;
  Empty SET (0.01 sec)
  
  -- 10 秒後にクエリを実行します。インポートされたデータをクエリできます。テーブルプロパティ group_commit_interval を使用して、データの可視性のレイテンシを制御できます。
  mysql> SELECT * FROM dt;
  +------+-------+-------+
  | id   | name  | score |
  +------+-------+-------+
  |    1 | Bob   |    90 |
  |    2 | Alice |    99 |
  |    3 | John  |  NULL |
  +------+-------+-------+
  3 rows in set (0.02 sec)

• 同期モード

  -- セッション変数 group_commit を設定して、同期モードでグループコミット機能を有効にします。変数のデフォルト値は off_mode です。
  mysql> SET group_commit = sync_mode;
  
  -- コミット間隔は、テーブルプロパティ group_commit_interval によって指定されます。返されるラベルは group_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'}
  
  -- インポートされたデータは、インポート完了直後にクエリできます。
  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)

• グループコミット機能を無効にする

  mysql> SET group_commit = off_mode;

Stream Load を使用する

Stream Load 機能の詳細については、「Stream Load」をご参照ください。

1. data.csv という名前のファイルを作成し、次の内容をファイルに追加します。

   6,Amy,60
   7,Ross,98

2. Stream Load を使用して、非同期モードまたは同期モードでデータをインポートします。

   • 非同期モード

     # インポートの URL に group_commit:async_mode ヘッダーを追加します。
     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
     }
     
     # 返される GroupCommit パラメーターの値は true です。これは、グループコミットが有効になっていることを示します。
     # このインポートに関連付けられているラベルは、group_commit で始まります。

   • 同期モード

     # インポートの URL に group_commit:sync_mode ヘッダーを追加します。
     
     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
     }
     
     # 返される GroupCommit パラメーターの値は true です。これは、グループコミットが有効になっていることを示します。
     # このインポートに関連付けられているラベルは、group_commit で始まります。