Broker Load

Step 1: Create a table

CREATE DATABASE IF NOT EXISTS load_test;
USE load_test;
CREATE TABLE IF NOT EXISTS customer (
  c_customer_sk         BIGINT,
  c_customer_id         CHAR(16),
  c_current_cdemo_sk    BIGINT,
  c_current_hdemo_sk    BIGINT,
  c_current_addr_sk     BIGINT,
  c_first_shipto_date_sk BIGINT,
  c_first_sales_date_sk  BIGINT,
  c_salutation          CHAR(10),
  c_first_name          CHAR(20),
  c_last_name           CHAR(30),
  c_preferred_cust_flag CHAR(1),
  c_birth_day           INT,
  c_birth_month         INT,
  c_birth_year          INT,
  c_birth_country       VARCHAR(20),
  c_login               CHAR(13),
  c_email_address       CHAR(50),
  c_last_review_date_sk  BIGINT
)
DUPLICATE KEY (c_customer_sk)
DISTRIBUTED BY HASH(c_customer_sk) BUCKETS 5
PROPERTIES (
  "replication_num" = "1"
);

Step 2: Submit a load job

Download customer.orc and upload it to your OSS bucket. Then submit the load job:

LOAD LABEL load_test.customer_label
(
  DATA INFILE("<file_path>")
  INTO TABLE customer
  FORMAT AS "orc"
)
WITH BROKER
(
  "fs.oss.accessKeyId"     = "<your_access_key_id>",
  "fs.oss.accessKeySecret" = "<your_access_key_secret>",
  "fs.oss.endpoint"        = "oss-cn-<region>-internal.aliyuncs.com"
);

Replace the placeholders with your actual values:

Step 3: Check the job status

USE load_test;
SHOW LOAD WHERE LABEL = 'customer_label'\G

For a description of all fields, see Monitor a load job.

Step 4: Verify the imported data

-- Count total rows
SELECT COUNT(1) FROM load_test.customer;

-- Preview the first two rows
SELECT * FROM load_test.customer LIMIT 2;

Prerequisites

• The HDFS cluster and the StarRocks instance must be in the same VPC and the same zone.

• Enable the DataNode ports in the security group to allow HDFS data access. For details, see Manage security groups.

Step 1: Create a database and a table

CREATE DATABASE IF NOT EXISTS mydatabase;

CREATE TABLE IF NOT EXISTS mydatabase.userdata_broker_load (
    userId           INT,
    userName         VARCHAR(20),
    registrationDate DATE
)
ENGINE = OLAP
DUPLICATE KEY(userId)
DISTRIBUTED BY HASH(userId);

Step 2: Submit a load job

Download user_data.parquet and upload it to the /data directory in HDFS. Then submit the load job:

LOAD LABEL mydatabase.userdata_broker_load_label
(
  DATA INFILE("hdfs://<hdfs_ip>:<hdfs_port>/data/user_data.parquet")
  INTO TABLE userdata_broker_load
  FORMAT AS "parquet"
)
WITH BROKER
PROPERTIES
(
  "timeout" = "72000"
);

Step 3: Check the job status and verify data

USE mydatabase;
SHOW LOAD WHERE LABEL = 'userdata_broker_load_label'\G

After the job reaches FINISHED state, verify the imported data:

SELECT * FROM mydatabase.userdata_broker_load;

HDFS authentication

The community edition of HDFS supports simple authentication and Kerberos authentication.

"hadoop.security.authentication" = "simple",
"username" = "<hdfs_username>",
"password" = "<hdfs_password>"

"hadoop.security.authentication" = "kerberos",
"kerberos_principal"             = "nn/zelda1@ZELDA.COM",
"kerberos_keytab"                = "/keytab/hive.keytab",
"kerberos_keytab_content"        = "YWFhYWFh"

HDFS high availability (HA) configuration

When NameNode high availability (HA) is configured, StarRocks automatically identifies the active NameNode after a failover. Add the following to hdfs-site.xml on the Instance Configuration tab:

"dfs.nameservices"                                      = "ha_cluster",
"dfs.ha.namenodes.ha_cluster"                           = "ha_n1,ha_n2",
"dfs.namenode.rpc-address.ha_cluster.ha_n1"             = "<hdfs_host>:<hdfs_port>",
"dfs.namenode.rpc-address.ha_cluster.ha_n2"             = "<hdfs_host>:<hdfs_port>",
"dfs.client.failover.proxy.provider.ha_cluster"         = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"

LOAD LABEL syntax

LOAD LABEL [<database_name>.]<label_name>
(
    data_desc[, data_desc ...]
)
WITH BROKER
(
    StorageCredentialParams
)
[PROPERTIES
(
    opt_properties
)
]

data_desc parameters

DATA INFILE ("<file_path>"[, "<file_path>" ...])
[NEGATIVE]
INTO TABLE <table_name>
[PARTITION (<partition1_name>[, <partition2_name> ...])]
[TEMPORARY PARTITION (<temporary_partition1_name>[, <temporary_partition2_name> ...])]
[COLUMNS TERMINATED BY "<column_separator>"]
[ROWS TERMINATED BY "<row_separator>"]
[FORMAT AS "CSV | Parquet | ORC"]
[(format_type_options)]
[(column_list)]
[COLUMNS FROM PATH AS (<partition_field_name>[, <partition_field_name> ...])]
[SET <k1=f1(v1)>[, <k2=f2(v2)> ...]]
[WHERE predicate]

Label behavior

Each import job has a unique label within the database:

• After `FINISHED`: the label becomes invalid and cannot be reused.

• After `CANCELLED`: the label can be reused to submit a new import job.

Labels enforce exactly-once semantics: reusing a label for the same data prevents duplicate imports.

Atomicity

A single LOAD statement is atomic. When loading multiple data files, either all files are loaded successfully or none are.

Job properties (opt_properties)

PROPERTIES ("<key1>" = "<value1>"[, "<key2>" = "<value2>" ...])

Monitor a load job

Run SHOW LOAD to check the status of all import jobs in the current database:

USE <database_name>;
SHOW LOAD;

To filter by label:

SHOW LOAD WHERE LABEL = "<label_name>"\G

Cancel a load job

Cancel a job that is not in CANCELLED or FINISHED state:

CANCEL LOAD FROM <database_name> WHERE LABEL = "<label_name>";

Concurrency

A load job consists of one or more tasks that run in parallel. A job is split into multiple tasks when:

• Multiple data_desc clauses point to different source addresses.

• Multiple data_desc clauses point to different partitions of the same source table.

Each task is further split into instances distributed across backends (BEs) for parallel execution. The number of instances per task is determined by the following frontend (FE) configuration parameters:

Total instances = min(total_file_size / min_bytes_per_broker_scanner, max_broker_concurrency, load_parallel_instance_num × number_of_backends)