ApsaraDB for SelectDB:Use Broker Load to import data

Syntax

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

Parameters

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

Examples

The following examples all use the same table structure. Run these statements first to create the tables and sample data files.

Create tables

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

Sample data files

file1.txt (comma-separated):

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 (comma-separated):

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

Example 1: Import a CSV file from HDFS

Imports file1.txt into test_table using comma as the column delimiter. The fs.defaultFS property is required to connect to the HDFS cluster.

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

Example 2: Import multiple files into multiple tables in one job

Imports file2.csv into test_table and file1.txt into test_table2 in a single atomic job. The SET clause increments the age column by 1 using the temp_age column from the source file.

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

Example 3: Import from an HDFS cluster in high availability (HA) mode

Uses a wildcard to import all files in the /example/ directory. The \\x01 delimiter is the default Hive field separator.

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

Example 4: Import with a row filter

Imports only rows where age < 20 from 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"
);

Example 5: MERGE mode with timeout and error tolerance

Imports file1.txt in MERGE mode into test_table (which uses the Unique Key model). Rows where age < 20 are marked for deletion. The job allows up to 10% error rows and has a 1-hour timeout.

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

Syntax

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

Parameters

Job states

Output columns

SHOW LOAD returns the following columns for each job:

When a job is in CANCELLED state, use the ErrorMsg and URL fields to diagnose the failure before resubmitting.

Examples

Query jobs by label pattern, return 10 oldest

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

Query a specific job, sorted by start time

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

Query a job in the LOADING state

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

Paginate results: skip the first 5, return the next 10

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

Syntax

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

Parameters

Examples

Cancel a specific job

CANCEL LOAD
FROM example_db
WHERE LABEL = "example_db_test_load_label";

Cancel all jobs whose label starts with `example_`

CANCEL LOAD
FROM example_db
WHERE LABEL LIKE "example_";

Keep individual jobs under 100 GB

The per-job data limit is the number of nodes multiplied by 3 GB. For large datasets, split the data into multiple jobs rather than a single large job. This reduces retry costs if a job fails and makes it easier to track progress.

Design labels for traceability

Use labels that encode metadata — for example, etl_orders_20260328_batch01. This makes it easy to identify failed jobs, rerun specific batches, and audit load history with SHOW LOAD.

Monitor the queue before submitting

SelectDB limits concurrent import jobs per cluster to between 3 and 10, with a queue limit of 100. Jobs beyond the queue limit are rejected immediately. Monitor active and queued jobs with SHOW LOAD before submitting new ones.

Queued time counts toward a job's timeout. If a job waits too long in the queue before executing, it may time out and be cancelled. Adjust the timeout property if your workload includes queue delays.

Choose max_filter_ratio based on data quality

When a job fails or skips rows, open the URL from the URL field in SHOW LOAD to see exactly which rows were rejected and why.

Use load_parallelism for large imports

Setting load_parallelism to a value greater than 1 runs multiple execution plans concurrently. This accelerates imports for large files but increases memory usage. Start with 2 or 4 and adjust based on cluster capacity.

Ensure idempotency with labels

Broker Load enforces at-most-once semantics per label. If a job with a given label succeeds (FINISHED state), resubmitting the same label has no effect. If the job failed or was cancelled, the label is available for reuse. Always check SHOW LOAD before deciding whether to resubmit.