ApsaraDB for SelectDB:Broker Load を使用したデータのインポート

メリット

Broker Load には、次のメリットがあります。

• 大量のデータ: Broker Load を使用すると、一度に数百 GB のオフラインデータをインポートできます。

• 非同期の高い同時実行性: Broker Load を使用すると、データをブロックすることなく、非同期モードでデータをインポートできます。 これにより、クラスタのリソース使用率が向上します。

• 高い互換性: Broker Load を使用すると、HDFS や Amazon S3 などのリモートストレージシステムからデータを読み取ることができます。

• 使いやすさ:

  • MySQL プロトコルを介して Broker Load ジョブを作成し、データをインポートできます。

  • SHOW LOAD 文を実行して、データのインポートの進捗状況と結果をリアルタイムで監視できます。

適用シナリオ

Broker Load を使用すると、HDFS、OSS、Amazon S3 などの分散ストレージシステムから大量のデータを効率的に読み取ることができます。

• 100 MB などの少量のデータのデータインポート適時性は、10 秒レベルです。

• 100 GB などの大量のデータのデータインポート適時性は、10 分レベルです。

Broker Load ジョブの作成

Broker Load ジョブは、ブローカーを使用して、HDFS や Amazon S3 などのリモートストレージシステムから ApsaraDB for SelectDB インスタンスのテーブルにデータを読み込んでインポートします。

構文

LOAD LABEL load_label
(
data_desc1[, data_desc2, ...]
)
WITH broker_type
[broker_properties]
[load_properties];

パラメータ

( "key1" = "val1", "key2" = "val2", ...)

[MERGE|APPEND|DELETE]
DATA INFILE
(
"file_path1"[, file_path2, ...]
)
[NEGATIVE]
INTO TABLE `table_name`
[PARTITION (p1, p2, ...)]
[COLUMNS TERMINATED BY "column_separator"]
[FORMAT AS "file_type"]
[(column_list)]
[COLUMNS FROM PATH AS (c1, c2, ...)]
[PRECEDING FILTER predicate]
[SET (column_mapping)]
[WHERE predicate]
[DELETE ON expr]
[ORDER BY source_sequence]
[PROPERTIES ("key1"="value1", ...)]

例

1. ApsaraDB for SelectDB インスタンスでデータをインポートするテーブルを作成します。 サンプルコード:

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

2. データをインポートするファイルを作成します。

   • 次の内容を含む file1.txt という名前のファイルを作成します。

     1,tomori,32,shanghai
     2,anon,22,beijing
     3,taki,23,shenzhen
     4,rana,45,hangzhou
     5,soyo,14,shanghai
     6,saki,25,hangzhou
     7,mutsumi,45,shanghai
     8,uika,26,shanghai
     9,umiri,27,shenzhen
     10,nyamu,37,shanghai

   • 次の内容を含む file2.csv という名前のファイルを作成します。

     1,saki,25,hangzhou
     2,mutsumi,45,shanghai
     3,uika,26,shanghai
     4,umiri,27,shenzhen
     5,nyamu,37,shanghai

3. ファイルのデータをテーブルにインポートします。

   • HDFS から file1.txt ファイルをインポートします。サンプルコード:

     LOAD LABEL example_db.label1
     (
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file1.txt")
         INTO TABLE `my_table`
         COLUMNS TERMINATED BY ","
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     );

     file1.txt ファイルを test_table テーブルにインポートします。 列はコンマ (,) で区切ります。 HDFS からデータをインポートする場合は、broker_properties パラメータで fs.defaultFS プロパティを指定して、システムが HDFS クラスタに想定どおりに接続し、対応するファイルを見つけられるようにする必要があります。

   • HDFS から 2 つのファイルを 2 つのテーブルにインポートします。 サンプルコード:

     LOAD LABEL test_db.test_02
     (
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file2.csv")
         INTO TABLE `test_table`
         COLUMNS TERMINATED BY ","
         (id,name,temp_age,address)    
         SET (
             age = temp_age + 1
         ),
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file1.txt")
         INTO TABLE `test_table2`
         COLUMNS TERMINATED BY ","
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     );

     file1.txt ファイルを test_table テーブルにインポートします。file2.csv ファイルを test_table2 テーブルにインポートし、file2.csv ファイルの temp_age 列の値に基づいて、age 列の値を 1 ずつ増分します。

   • 高可用性 (HA) モードでデプロイされた HDFS クラスタからバッチデータをインポートします。 サンプルコード:

     LOAD LABEL test_db.test_03
     (
         DATA INFILE("hdfs://hdfs_host:hdfs_port/example/*")
         INTO TABLE `test_table`
         COLUMNS TERMINATED BY "\\x01"
     )
     WITH HDFS
     (
         "hadoop.username" = "hive",
         "fs.defaultFS" = "hdfs://my_ha",
         "dfs.nameservices" = "my_ha",
         "dfs.ha.namenodes.my_ha" = "my_namenode1, my_namenode2",
         "dfs.namenode.rpc-address.my_ha.my_namenode1" = "nn1_host:rpc_port",
         "dfs.namenode.rpc-address.my_ha.my_namenode2" = "nn2_host:rpc_port",
         "dfs.client.failover.proxy.provider.my_ha" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
     );

     Hive のデフォルト デリミタ \\x01 を使用し、ワイルドカード (*) を使用して data ディレクトリ内のすべてのファイルを指定します。

   • file1.txt ファイルのデータをフィルタリングして、フィルタ条件を満たすデータ行のみをインポートします。 サンプルコード:

     LOAD LABEL test_db.test_04
     (
         DATA INFILE("hdfs://host:port/example/file1.txt")
         INTO TABLE `test_table2`
         COLUMNS TERMINATED BY ","
         (id,name,age,address)   
         WHERE age < 20
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     );
     

     age 列の値が 20 未満の行のみがインポートされます。

   • HDFS から file1.txt ファイルをインポートします。 インポートのタイムアウト期間とフィルタリング比率を指定します。 age 列の値が 20 未満の行を除くすべての行をファイルからインポートします。 サンプルコード:

     LOAD LABEL test_db.test_05
     (
         MERGE DATA INFILE("hdfs://hdfs_host:hdfs_port/example/file1.txt")
         INTO TABLE `test_table`
         COLUMNS TERMINATED BY ","
         (id,name,age,address)   
         DELETE ON age < 20
     )
     WITH HDFS
     (
         "fs.defaultFS" = "hdfs://hdfs_host:hdfs_port"
     )
     PROPERTIES
     (
         "timeout" = "3600",
         "max_filter_ratio" = "0.1"
     );

     MERGE モードでデータをインポートします。 test_table テーブルは一意キーモデルを使用する必要があります。インポートされる行の age 列の値が 20 未満の場合、その行は削除対象としてマークされます。インポート ジョブのタイムアウト期間は 3,600 秒で、エラーデータ行の最大フィルタリング率は 10% です。

Broker Load ジョブのキャンセル

Broker Load ジョブが CANCELLED または FINISHED 状態ではない場合、ジョブを手動でキャンセルできます。 キャンセルするインポートジョブのラベルを指定する必要があります。 インポートジョブがキャンセルされると、ジョブに書き込まれたデータはロールバックされ、有効になりません。

構文

CANCEL LOAD
[FROM db_name]
WHERE [LABEL = "load_label" | LABEL like "label_pattern"];

パラメータ

例

• ラベルが example_db_test_load_label であるインポートジョブを example_db データベースからキャンセルします。

  CANCEL LOAD
  FROM example_db
  WHERE LABEL = "example_db_test_load_label";

• ラベルに example_ が含まれるインポートジョブを example_db データベースからキャンセルします。

  CANCEL LOAD
  FROM example_db
  WHERE LABEL like "example_";

Broker Load ジョブのステータスの照会Broker ロード ジョブ

Broker Load は非同期のインポート方式です。 インポート文が正常に実行された場合は、データのインポートが完了したのではなく、Broker Load ジョブが正常に送信されたことのみを示します。 Broker Load ジョブのステータスを照会するには、SHOW LOAD 文を実行します。

構文

SHOW LOAD
[FROM db_name]
[
   WHERE
   [LABEL [ = "your_label" | LIKE "label_matcher"]]
   [STATE = ["PENDING"|"ETL"|"LOADING"|"FINISHED"|"CANCELLED"|]]
]
[ORDER BY ...]
[LIMIT limit][OFFSET offset];

パラメータ

ベストプラクティス

• インポートジョブのステータスの照会

  Broker Load は非同期のインポート方式です。 インポート文が正常に実行された場合は、データのインポートが完了したのではなく、Broker Load ジョブが正常に送信されたことのみを示します。 インポートジョブのステータスを照会するには、SHOW LOAD 文を実行します。

• インポートジョブのキャンセル

  送信済みだが完了していないインポートジョブをキャンセルするには、CANCEL LOAD 文を実行します。 インポートジョブがキャンセルされると、ジョブに書き込まれたデータはロールバックされ、有効になりません。

• ラベル、インポートトランザクション、複数テーブルの原子性

  ApsaraDB for SelectDB のすべてのインポートジョブは、本質的にアトミックです。 インポートジョブで複数のテーブルにデータをインポートする場合でも、原子性は保証されます。 また、ApsaraDB for SelectDB はラベルを使用して、インポートされたデータが失われたり重複したりしないようにします。

• 列のマッピング、導出、フィルタリング

  ApsaraDB for SelectDB は、インポート文での列変換と列フィルタリングに関するさまざまな操作をサポートしています。 ほとんどのビルトイン関数とユーザー定義関数 (UDF) がサポートされています。 詳細については、「ソースデータの変換」をご参照ください。

• エラーデータ行のフィルタリング

  ApsaraDB for SelectDB では、インポートジョブでフォーマットが正しくないデータ行をスキップできます。フィルタリング率は、max_filter_ratio パラメーターで指定します。デフォルト値は 0 で、データをフィルタリングしないゼロトレランスポリシーを指定します。エラーデータ行が見つかった場合、インポートジョブは失敗します。インポート中にエラーデータ行を無視する場合は、このパラメーターを 0 ～ 1 の値に設定できます。このようにして、ApsaraDB for SelectDB はデータ形式が正しくない行を自動的にスキップします。

  トレランス比率の計算方法の詳細については、「ソースデータの変換」をご参照ください。

• 厳密モード

  strict_mode プロパティは、インポート ジョブを厳密モードで実行するかどうかを指定するために使用します。インポート ジョブが厳密モードで実行される場合、列のマッピング、変換、およびフィルタリングが影響を受けます。

• タイムアウト期間

  Broker Load ジョブのデフォルトのタイムアウト期間は 4 時間です。 タイミングは、インポートジョブが送信された時点から開始されます。 インポートジョブがタイムアウト期間内に完了しない場合、インポートジョブは失敗します。

• データ量とジョブ数の制限

  Broker Load ジョブでは、一度に 100 GB 未満のデータをインポートすることをお勧めします。 理論的には、インポートジョブでインポートできるデータ量に制限はありません。 ただし、インポートするデータ量が非常に大きい場合、インポートジョブの実行に長時間かかる可能性があり、インポートジョブが失敗した場合の再試行のコストが高くなります。

  また、クラスタ内のノード数は限られています。 したがって、インポートされるデータの最大量には制限が設定されています。これは、ノード数に 3 GB を掛けた値です。 これにより、システムリソースが適切に使用されるようになります。 大量のデータをインポートする必要がある場合は、複数のインポートジョブを送信することをお勧めします。

  ApsaraDB for SelectDB は、クラスタ内の同時インポートジョブ数を 3 ～ 10 の範囲に制限しています。 送信したインポートジョブの数が制限を超えると、超過したインポートジョブはキューで待機します。 キューの最大長は 100 です。 100 を超えるインポートジョブが待機している場合、超過したインポートジョブは直接拒否されます。

  説明

  待機時間もインポートジョブの合計時間に含まれます。 タイムアウトエラーが発生した場合、ジョブはキャンセルされます。 インポートジョブのステータスを監視して、ジョブの送信頻度を適切に制御することをお勧めします。