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

構文

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

パラメーター

[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", ...)]

使用例

以下の例はすべて同じテーブル構造を使用しています。最初に次の文を実行して、テーブルとサンプルデータファイルを作成してください。

テーブルの作成

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");

サンプルデータファイル

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

例 1：HDFS から CSV ファイルをインポート

カンマを列区切り文字として使用し、file1.txt を test_table にインポートします。fs.defaultFS プロパティは HDFS クラスターへの接続に必須です。

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

例 2：1 つのジョブで複数のファイルを複数のテーブルにインポート

file2.csv を test_table に、file1.txt を test_table2 に、1 つのアトミックジョブでインポートします。SET 句により、ソースファイルの temp_age 列を使用して age 列の値を 1 増分します。

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"
);

例 3：高可用性（HA）モードの HDFS クラスターからのインポート

ワイルドカードを使用して /example/ ディレクトリ内のすべてのファイルをインポートします。\\x01 デリミタは Hive のデフォルトのフィールド区切り文字です。

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"
);

例 4：行フィルターを使用したインポート

file1.txt から age < 20 の条件を満たす行のみをインポートします。

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"
);

例 5: タイムアウトとエラー許容を伴う MERGE モード

Unique Key モデルを使用する test_table に file1.txt を MERGE モードでインポートします。age < 20 の行は削除対象としてマークされます。このジョブは最大 10% のエラー行を許容し、タイムアウトは 1 時間に設定されています。

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"
);

構文

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

パラメーター

ジョブ状態

出力カラム

SHOW LOAD は各ジョブについて以下のカラムを返します。

ジョブが CANCELLED 状態の場合、再送信前に ErrorMsg および URL フィールドを使用して障害を診断してください。

使用例

ラベルパターンでジョブを検索し、古い順に 10 件取得

SHOW LOAD FROM example_db WHERE LABEL LIKE "2014_01_02" LIMIT 10;

特定のジョブを検索し、開始時刻でソート

SHOW LOAD FROM example_db
WHERE LABEL = "load_example_db_20140102"
ORDER BY LoadStartTime DESC;

LOADING 状態のジョブを検索

SHOW LOAD FROM example_db
WHERE LABEL = "load_example_db_20140102" AND STATE = "loading";

結果をページネーション：先頭 5 件をスキップし、次の 10 件を取得

SHOW LOAD FROM example_db ORDER BY LoadStartTime DESC LIMIT 10 OFFSET 5;

構文

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

パラメーター

使用例

特定のジョブをキャンセル

CANCEL LOAD
FROM example_db
WHERE LABEL = "example_db_test_load_label";

ラベルが `example_` で始まるすべてのジョブをキャンセル

CANCEL LOAD
FROM example_db
WHERE LABEL LIKE "example_";

個々のジョブのデータ量を 100 GB 未満に抑える

1 ジョブあたりのデータ上限はノード数 × 3 GB です。大規模なデータセットの場合は、1 つの大きなジョブではなく複数のジョブに分割してください。これにより、ジョブが失敗した場合のリトライコストを削減し、進行状況の追跡も容易になります。

トレーサビリティを考慮したラベル設計

メタデータをエンコードしたラベルを使用してください（例：etl_orders_20260328_batch01）。これにより、失敗したジョブの特定、特定バッチの再実行、SHOW LOAD を使用したロード履歴の監査が容易になります。

送信前にキュー状況を監視する

SelectDB はクラスターあたりの同時インポートジョブ数を 3～10 に制限しており、キュー上限は 100 です。キュー上限を超えたジョブは即座に拒否されます。新しいジョブを送信する前に、SHOW LOAD を使用してアクティブおよびキュー中のジョブを監視してください。

キュー待ち時間もジョブのタイムアウトに含まれます。キュー内で実行開始までに時間がかかりすぎると、タイムアウトによりジョブがキャンセルされる可能性があります。ワークロードにキュー遅延が含まれる場合は、timeout プロパティを調整してください。

データ品質に基づいて max_filter_ratio を設定する

ジョブが失敗または行をスキップした場合、SHOW LOAD の URL フィールドに記載された URL を開き、どの行が拒否されたのか、その理由を正確に確認してください。

大規模インポートには load_parallelism を使用する

load_parallelism を 1 より大きい値に設定すると、複数の実行計画が同時に実行されます。これにより大規模ファイルのインポートが高速化されますが、メモリ使用量も増加します。クラスターの容量に応じて、2 または 4 から開始し、必要に応じて調整してください。

ラベルによる冪等性の確保

Broker Load はラベル単位で at-most-once セマンティクスを保証します。特定のラベルを持つジョブが成功（FINISHED 状態）した場合、同じラベルを再送信しても効果はありません。ジョブが失敗またはキャンセルされた場合は、そのラベルを再利用できます。再送信の判断を行う前に、必ず SHOW LOAD で状態を確認してください。