ApsaraDB for SelectDB:INSERT INTO 文を使用したデータのインポート

背景情報

INSERT INTO 文は、MySQL データベースなどのデータベースにデータをインポートするためによく使用されます。ApsaraDB for SelectDB は標準 SQL 構文と互換性があり、INSERT INTO 文を実行してデータをインポートできます。INSERT INTO 文は、次の 2 つのタイプに分けられます。

• INSERT INTO tbl SELECT ...

• INSERT INTO tbl (col1, col2, ...) VALUES (1, 2, ...), (1,3, ...);

  重要

  INSERT INTO VALUES 文の書き込みパフォーマンスは低くなっています。本番環境ではこの文を使用しないことをお勧めします。一度に少量のデータを ApsaraDB for SelectDB インスタンスに書き込むには、Stream Load を使用することをお勧めします。これにより、書き込みパフォーマンスが 100 倍向上します。

使用上の注意

特定のシナリオでは、少量のデータを ApsaraDB for SelectDB インスタンスに頻繁に書き込むと、ApsaraDB for SelectDB インスタンスのパフォーマンスが大幅に低下し、テーブルでデッドロックが発生する可能性があります。少量のデータを ApsaraDB for SelectDB インスタンスに同時に書き込み、テーブルへのデータ書き込みの頻度を 10 秒より大きい値に設定することをお勧めします。

INSERT INTO SELECT 文

INSERT INTO SELECT 文は、SelectDB 内のデータと外部データレイク内のデータを、SelectDB が提供する多数の SQL 関数とフェデレーテッドクエリを使用して効率的に計算および処理できます。その後、処理されたデータは、SelectDB 内の新しいテーブルにインポートされ、さらに分析されます。

内部テーブルのデータに対する ETL 操作の実行

データがすでに SelectDB 内のテーブルに保存されている場合は、INSERT INTO SELECT 文を実行することで、データに対して抽出、変換、ロード (ETL) 操作を実行し、データを ApsaraDB for SelectDB の新しいテーブルにインポートできます。たとえば、次の文を実行して、ApsaraDB for SelectDB の store_sales テーブルから region が bj に設定されているデータを、bj_store_sales という名前の新しいテーブルにインポートできます。

INSERT INTO bj_store_sales
SELECT id, total, user_id, sale_timestamp FROM store_sales WHERE region = "bj";

データレイクからのデータの同期

データがデータレイクなどの外部システムに保存されている場合は、SelectDB にカタログを作成し、カタログを外部システムのデータにマッピングしてから、INSERT INTO SELECT 文を実行してデータを SelectDB 内のテーブルにインポートできます。

カタログを使用して、Hive、Iceberg、Hudi、Elasticsearch、Java Database Connectivity (JDBC) などのデータソースを ApsaraDB for SelectDB と統合できます。次の情報は、カタログを使用して Hive データソースを ApsaraDB for SelectDB インスタンスと統合し、データレイクから SelectDB インスタンス内のテーブルにデータを同期する方法の例を示しています。他のデータソースの詳細については、「データレイクハウス」をご参照ください。

1. SelectDB インスタンスに接続します。詳細については、「MySQL クライアントを使用して ApsaraDB for SelectDB インスタンスに接続する」をご参照ください。

2. カタログを作成し、カタログを使用して Hive データソースを ApsaraDB for SelectDB インスタンスと統合します。詳細については、「Hive データソース」をご参照ください。

3. オプション。hive_db という名前のデータベースを作成します。

   ターゲットデータベースをすでに作成している場合は、この手順をスキップします。

   create database hive_db;

4. ターゲットデータベースに切り替えます。

   use hive_db;

5. テーブルを作成します。

   ターゲットテーブルをすでに作成している場合は、ターゲットテーブルの列のデータ型が Hive データソースのソーテーブルのデータ型にマッピングされているかどうかを確認します。列のデータ型マッピング

   ターゲットテーブルを作成していない場合は、ターゲットテーブルを作成し、ターゲットテーブルの列のデータ型が Hive データソースのソーステーブルのデータ型にマッピングされていることを確認します。詳細については、「列のデータ型マッピング」をご参照ください。

   CREATE TABLE test_Hive2SelectDB
   (
       id int,
       name varchar(50),
       age int
   )
   DISTRIBUTED BY HASH(id) BUCKETS 4
   PROPERTIES("replication_num" = "1");

6. オプション。テーブルデータを表示します。

   select * from test_Hive2SelectDB;

   

7. データを移行します。

   INSERT INTO SELECT 文を実行して、Hive データソースから ApsaraDB for SelectDB 内のテーブルにデータを同期し、データインポートジョブに一意のラベルを指定します。

   INSERT INTO test_Hive2SelectDB WITH LABEL test_label SELECT *  FROM hive_catalog.testdb.hive_t;

8. データのクエリ。

   ターゲットテーブルのデータは左側に表示され、ソースデータテーブルは右側に表示されます。

   

INSERT INTO VALUES 文

INSERT INTO VALUES 文は、MySQL データベースなどのデータベースにデータを書き込むためによく使用されます。テスト環境でのみ INSERT INTO VALUES 文を使用することをお勧めします。通常、SQL クライアントまたは JDBC プログラムを使用して、データ書き込みのリクエストを送信できます。

次のサンプルコードは、ApsaraDB for SelectDB でデータがインポートされるテーブルを作成する方法の例を示しています。

CREATE TABLE test_table
(
    id int,
    name varchar(50),
    age int
)
DISTRIBUTED BY HASH(id) BUCKETS 4
PROPERTIES("replication_num" = "1");

SQL クライアントを使用したデータのインポート

BEGIN;
INSERT INTO test_table VALUES (1, 'Zhang San', 32),(2, 'Li Si', 45),(3, 'Zhao Liu', 23);
INSERT INTO test_table VALUES (4, 'Wang Yi', 32),(5, 'Zhao Er', 45),(6, 'Li Er', 23);
INSERT INTO test_table VALUES (7, 'Li Yi', 32),(8, 'Wang San', 45),(9, 'Zhao Si', 23);
COMMIT;

JDBC プログラムを使用したデータのインポート

public static void main(String[] args) throws Exception {
    // バッチでデータをインポートするために使用される INSERT 文の数。
    int insertNum = 10;
    // バッチでインポートされるデータエントリの数。
    int batchSize = 10000;
    String URL="jdbc:mysql://<ホスト IP アドレス>:<MySQL プロトコルポート>/test_db?useLocalSessionState=true"; // ApsaraDB for SelectDB コンソールにログインして、[インスタンスの詳細] ページの [ネットワーク情報] セクションで VPC エンドポイントまたはパブリックエンドポイントを表示できます。
    Connection connection = DriverManager.getConnection(URL, "admin", "password");  // ApsaraDB for SelectDB インスタンスにログインするために使用されるアカウントのユーザー名とパスワード。
    Statement statement = connection.createStatement();
    statement.execute("begin;");
    // 複数の INSERT 文を追加します。
    for (int num = 0; num < insertNum; num++) {
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO test_table values ");
        for(int i = 0; i < batchSize; i++){
            if(i > 0){
                sql.append(",");
            }
            // 名前や年齢などのフィールドを含むデータ行を追加します。ビジネス要件に基づいて、データ行に含まれるフィールドを変更できます。
            sql.append("(1, 'Zhang San', 32)");
        }
        //add sql to batch: INSERT INTO tbl values(),(),()
        statement.addBatch(sql.toString());
    }
    statement.addBatch("commit;");
    statement.executeBatch();
    // リソースを閉じます。
    statement.close();
    connection.close();
}

ベストプラクティス

• 返された結果

  INSERT INTO 操作は同期操作です。結果が返された場合、挿入操作は完了です。返された結果に基づいて操作を実行する必要があります。

  1. INSERT INTO 操作は成功し、結果セットは空です。

     INSERT INTO 文の SELECT 句の結果セットが空の場合、次の出力に類似した結果が返されます。

     INSERT INTO tbl1 SELECT * FROM empty_tbl;
     Query OK, 0 rows affected (0.02 sec)

     Query OK は、操作が成功したことを示します。0 rows affected は、データがインポートされていないことを示します。

  2. INSERT INTO 操作は成功し、結果セットは空ではありません。

     結果セットが空でない場合、次のいずれかの出力に類似した結果が返されます。

     INSERT INTO tbl1 SELECT * FROM tbl2;
     Query OK, 4 rows affected (0.38 sec)
     {'label':'insert_8510c568-9eda-****-9e36-6adc7d35291c', 'status':'visible', 'txnId':'4005'}
      
     INSERT INTO tbl1 with label my_label1 SELECT * FROM tbl2;
     Query OK, 4 rows affected (0.38 sec)
     {'label':'my_label1', 'status':'visible', 'txnId':'4005'}
      
     INSERT INTO tbl1 SELECT * FROM tbl2;
     Query OK, 2 rows affected, 2 warnings (0.31 sec)
     {'label':'insert_f0747f0e-7a35-****-affa-13a235f4020d', 'status':'visible', 'txnId':'4005'}
      
     INSERT INTO tbl1 SELECT * FROM tbl2;
     Query OK, 2 rows affected, 2 warnings (0.31 sec)
     {'label':'insert_f0747f0e-7a35-****-affa-13a235f4020d', 'status':'committed', 'txnId':'4005'}

     Query OK は、操作が成功したことを示します。4 rows affected は、4 行のデータがインポートされたことを示します。2 warnings は、2 行のデータが除外されたことを示します。JSON 文字列も返されます。

     {'label':'my_label1', 'status':'visible', 'txnId':'4005'}
     {'label':'insert_f0747f0e-7a35-****-affa-13a235f4020d', 'status':'committed', 'txnId':'4005'}
     {'label':'my_label1', 'status':'visible', 'txnId':'4005', 'err':'some other error'}

     label は、指定したラベル、または自動的に生成されたラベルを示します。ラベルは、INSERT INTO 文を使用してデータインポートジョブを識別するために使用されます。各データインポートジョブには、単一のデータベース内で一意のラベルがあります。status は、インポートされたデータが表示されるかどうかを示します。インポートされたデータが表示される場合、status パラメーターには visible が返されます。インポートされたデータが表示されない場合、status パラメーターには committed が返されます。txnId は、挿入操作に対応するインポートトランザクションの ID を示します。err は、予期しないエラーを示します。

     除外された行をクエリするには、次の文を実行します。

     SHOW LOAD WHERE label="xxx";

     返された結果の URL を使用して、エラーデータをクエリできます。詳細については、このトピックのエラー行の概要を参照してください。データは一時的に表示されない場合があります。ただし、このバッチのデータは最終的に表示されます。次の文を実行して、このバッチのデータが表示されるかどうかを確認できます。

     SHOW TRANSACTION WHERE id=4005;

     返された結果の TransactionStatus 列に visible が表示されている場合、データは表示されます。

  3. INSERT INTO 操作が失敗します。

     INSERT INTO 操作が失敗した場合、データはインポートされず、次の出力に類似した結果が返されます。

     INSERT INTO tbl1 SELECT * FROM tbl2 WHERE k1 = "a";
     ERROR 1064 (HY000): all partitions have no load data. url: http://10.74.167.16:8042/api/_load_error_log?file=__shard_2/error_log_insert_stmt_ba8bb9e158e4879-ae8de8507c0bf8a2_ba8bb9e158e4879_ae8de8507c0bf8a2

     ERROR 1064 (HY000): all partitions have no load data は、失敗の原因を示します。返された結果の URL を使用して、エラーデータをクエリできます。

     SHOW LOAD WARNINGS ON "url";

• タイムアウト期間

  INSERT INTO 操作のタイムアウト期間は、query_timeout セッション変数によって指定されます。INSERT INTO 操作のデフォルトのタイムアウト期間は 5 分です。インポートジョブが指定されたタイムアウト期間内に完了しない場合、システムはジョブをキャンセルします。

• ラベルと原子性

  INSERT INTO 操作では、インポートジョブの原子性も保証されます。共通テーブル式 (CTE) を使用して INSERT INTO 文でサブクエリを定義する場合は、WITH LABEL と column を指定する必要があります。

• フィルタリングしきい値

  INSERT INTO 文を使用してデータをインポートする場合、max_filter_ratio パラメーターを指定してフィルタリングしきい値を指定することはできません。デフォルトのフィルタリングしきい値は 1 で、すべてのエラー行を無視できることを示します。

  データを除外できないビジネス シナリオでは、enable_insert_strict セッション変数を true に設定できます。これにより、データが除外された場合、INSERT INTO 文が正常に実行されなくなります。

• パフォーマンスの問題

  特にビッグデータのオンライン本番環境では、INSERT INTO VALUES 文を使用してデータをインポートしないことをお勧めします。INSERT INTO VALUES 文を使用してデータをインポートする必要がある場合は、複数のデータ行を 1 つの INSERT INTO 文にマージして、データをバッチでインポートします。1 バッチで 1,000 ～ 1,000,000 データ行をインポートすることをお勧めします。

• 特定の列の更新

  デフォルトでは、INSERT INTO 文は行ごとにデータを書き込みます。Unique データモデルの merge-on-write (MOW) 実装方法では、ビジネス要件に基づいて特定の列を更新するように次のセッション変数を構成できます。

  set enable_unique_key_partial_update=true

  変数の詳細については、「変数管理」をご参照ください。

  重要

  • この変数は、テーブルが Merge on Write (MOW) モードを使用する Unique モデルを使用している場合にのみ使用できます。

  • enable_unique_key_partial_update 変数と enable_insert_strict 変数の両方を true に設定すると、INSERT INTO 文を実行してデータのみを更新できます。 enable_insert_strict 変数のデフォルト値 true は、INSERT 文に対して厳密モードが有効になっていることを示します。INSERT INTO 文にテーブルに存在しないキーが含まれている場合は、エラーが報告されます。

  • この変数を true に設定した後、INSERT INTO 文を実行して特定の列を更新し、列にデータを挿入する場合は、enable_unique_key_partial_update セッション変数を true に、enable_insert_strict セッション変数を false に設定する必要があります。変数の設定の詳細については、「変数の構成」をご参照ください。

よくある質問

データインポート中に get table cloud commit lock timeout エラーメッセージが表示された場合はどうすればよいですか？

ApsaraDB for SelectDB インスタンスに頻繁にデータを書き込みます。その結果、デッドロックが発生します。少量のデータを ApsaraDB for SelectDB インスタンスに同時に書き込むことをお勧めします。また、テーブルへのデータ書き込みの頻度を 5 秒以上にすることをお勧めします。