ApsaraDB for SelectDB:Gunakan Broker Load untuk mengimpor data

Sintaks

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

Parameter

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

Contoh

Semua contoh berikut menggunakan struktur tabel yang sama. Jalankan pernyataan berikut terlebih dahulu untuk membuat tabel dan file data sampel.

Buat tabel

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

File data sampel

file1.txt (dipisahkan koma):

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 (dipisahkan koma):

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

Contoh 1: Impor file CSV dari HDFS

Mengimpor file1.txt ke test_table menggunakan koma sebagai pemisah kolom. Properti fs.defaultFS diperlukan untuk terhubung ke kluster 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"
);

Contoh 2: Impor beberapa file ke beberapa tabel dalam satu pekerjaan

Mengimpor file2.csv ke test_table dan file1.txt ke test_table2 dalam satu pekerjaan atomik. Klausul SET menambahkan kolom age sebesar 1 menggunakan kolom temp_age dari file sumber.

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

Contoh 3: Impor dari kluster HDFS dalam mode high availability (HA)

Menggunakan wildcard untuk mengimpor semua file dalam direktori /example/. Pemisah \\x01 adalah pemisah bidang Hive default.

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

Contoh 4: Impor dengan filter baris

Hanya mengimpor baris di mana age < 20 dari 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"
);

Contoh 5: Mode MERGE dengan timeout dan toleransi error

Mengimpor file1.txt dalam mode MERGE ke test_table (yang menggunakan model Unique Key). Baris di mana age < 20 ditandai untuk dihapus. Pekerjaan ini mengizinkan hingga 10% baris error dan memiliki timeout 1 jam.

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

Sintaks

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

Parameter

Status pekerjaan

Kolom output

SHOW LOAD mengembalikan kolom berikut untuk setiap pekerjaan:

Jika pekerjaan berada dalam status CANCELLED, gunakan kolom ErrorMsg dan URL untuk mendiagnosis kegagalan sebelum mengirim ulang.

Contoh

Kueri pekerjaan berdasarkan pola label, kembalikan 10 yang tertua

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

Kueri pekerjaan tertentu, diurutkan berdasarkan waktu mulai

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

Kueri pekerjaan dalam status LOADING

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

Paginasi hasil: lewati 5 pertama, kembalikan 10 berikutnya

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

Sintaks

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

Parameter

Contoh

Batalkan pekerjaan tertentu

CANCEL LOAD
FROM example_db
WHERE LABEL = "example_db_test_load_label";

Batalkan semua pekerjaan yang labelnya diawali dengan `example_`

CANCEL LOAD
FROM example_db
WHERE LABEL LIKE "example_";

Batasi ukuran pekerjaan individu di bawah 100 GB

Batas data per pekerjaan adalah jumlah node dikalikan 3 GB. Untuk dataset besar, bagi data menjadi beberapa pekerjaan daripada satu pekerjaan besar. Hal ini mengurangi biaya pengulangan jika pekerjaan gagal dan mempermudah pelacakan progres.

Rancang label untuk kemudahan pelacakan

Gunakan label yang menyertakan metadata—misalnya, etl_orders_20260328_batch01. Hal ini mempermudah identifikasi pekerjaan yang gagal, menjalankan ulang batch tertentu, dan audit riwayat impor dengan SHOW LOAD.

Monitor antrean sebelum mengirim

SelectDB membatasi pekerjaan impor konkuren per kluster antara 3 hingga 10, dengan batas antrean 100. Pekerjaan yang melebihi batas antrean akan ditolak segera. Monitor pekerjaan aktif dan dalam antrean dengan SHOW LOAD sebelum mengirim pekerjaan baru.

Waktu dalam antrean dihitung terhadap timeout pekerjaan. Jika pekerjaan menunggu terlalu lama dalam antrean sebelum dieksekusi, pekerjaan tersebut mungkin timeout dan dibatalkan. Sesuaikan properti timeout jika workload Anda mencakup penundaan antrean.

Pilih max_filter_ratio berdasarkan kualitas data

Jika pekerjaan gagal atau melewati baris, buka URL dari kolom URL dalam SHOW LOAD untuk melihat secara tepat baris mana yang ditolak dan alasannya.

Gunakan load_parallelism untuk impor besar

Mengatur load_parallelism ke nilai lebih besar dari 1 menjalankan beberapa rencana eksekusi secara konkuren. Ini mempercepat impor untuk file besar tetapi meningkatkan penggunaan memori. Mulailah dengan 2 atau 4 dan sesuaikan berdasarkan kapasitas kluster.

Jamin idempotensi dengan label

Broker Load menerapkan semantik at-most-once per label. Jika pekerjaan dengan label tertentu berhasil (status FINISHED), pengiriman ulang dengan label yang sama tidak berpengaruh. Jika pekerjaan gagal atau dibatalkan, label tersebut tersedia untuk digunakan kembali. Selalu periksa SHOW LOAD sebelum memutuskan apakah akan mengirim ulang.