E-MapReduce:Broker Load

How Broker Load works

Broker Load has two operating modes depending on your StarRocks version:

• Broker-based loading (StarRocks earlier than 2.5.8): StarRocks depends on broker processes to connect to external storage. Specify the broker name explicitly with WITH BROKER "<broker_name>" in the load statement.

• Broker-free loading (StarRocks 2.5.8 and later): StarRocks connects to external storage directly. Omit the broker name but keep the WITH BROKER keyword in the load statement.

When you create a StarRocks cluster in E-MapReduce (EMR), broker processes are automatically deployed and started on every core node.

Supported sources and formats

Prerequisites

Before you begin, ensure that you have:

• A running StarRocks cluster in EMR with at least one core node

• Access credentials for HDFS or OSS

• A destination database and table in StarRocks

Query brokers

To confirm brokers are running on your cluster, run:

SHOW PROC "/brokers"\G

The output lists each broker with its IP address, port (default: 8000), alive status, and timestamps:

*************************** 1. row ***************************
          Name: broker
            IP: 10.0.**.**
          Port: 8000
         Alive: true
 LastStartTime: 2022-04-13 11:38:46
LastUpdateTime: 2022-04-13 15:26:44
        ErrMsg:

Create an import job

All import jobs use the LOAD LABEL statement. The syntax differs slightly between StarRocks versions.

LOAD LABEL db_name.label_name
    (data_desc, ...)
WITH BROKER broker_name broker_properties
    [PROPERTIES (key1=value1, ... )]

LOAD LABEL db_name.label_name
    (data_desc, ...)
WITH BROKER broker_properties
    [PROPERTIES (key1=value1, ... )]

In EMR StarRocks clusters, use broker as the broker name.

To view the full syntax reference, run HELP BROKER LOAD.

data_desc:
    DATA INFILE ('file_path', ...)
    [NEGATIVE]
    INTO TABLE tbl_name
    [PARTITION (p1, p2)]
    [COLUMNS TERMINATED BY column_separator]
    [FORMAT AS file_type]
    [(col1, ...)]
    [COLUMNS FROM PATH AS (colx, ...)]
    [SET (k1=f1(xx), k2=f2(xx))]
    [WHERE predicate]

broker_properties:
    (key2=value2, ...)

Load data from OSS

The following example loads a TPC-H lineitem file from OSS into a StarRocks table.

LOAD LABEL tpch.lineitem
(
    DATA INFILE("oss://bucket/tpc_h/sf1/lineitem.tbl")
    INTO TABLE `lineitem`
    COLUMNS TERMINATED BY '|'
    (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice,
     l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate,
     l_receiptdate, l_shipinstruct, l_shipmode, l_comment)
)

LOAD LABEL tpch.lineitem
(
    DATA INFILE("oss://bucket/tpc_h/sf1/lineitem.tbl")
    INTO TABLE `lineitem`
    COLUMNS TERMINATED BY '|'
    (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice,
     l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate,
     l_receiptdate, l_shipinstruct, l_shipmode, l_comment)
)
WITH BROKER broker
(
    "fs.oss.accessKeyId" = "xxx",
    "fs.oss.accessKeySecret" = "xxx",
    "fs.oss.endpoint" = "oss-cn-beijing-internal.aliyuncs.com"
);

Load data from HDFS

(
    "username" = "user",
    "password" = "passwd",
    "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"
)

LOAD LABEL db1.label1
(
    DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/hive/test.db/ml/file1")
    INTO TABLE tbl1
    COLUMNS TERMINATED BY ","
    (tmp_c1, tmp_c2)
    SET
    (
        id = tmp_c2,
        name = tmp_c1
    ),

    DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/hive/test.db/ml/file2")
    INTO TABLE tbl2
    COLUMNS TERMINATED BY ","
    (col1, col2)
    WHERE col1 > 1
)
WITH BROKER 'broker1'
(
    "username" = "hdfs_username",
    "password" = "hdfs_password"
)
PROPERTIES
(
    "timeout" = "3600"
);

Query job status

Broker Load is asynchronous. After submitting a job, use SHOW LOAD to track its progress:

SHOW LOAD WHERE label = 'label1'\G

To view the full SHOW LOAD syntax, run HELP SHOW LOAD.

SHOW LOAD works only with asynchronous import methods. It does not apply to synchronous methods like Stream Load.

The output looks like:

*************************** 1. row ***************************
         JobId: 7****
         Label: label1
         State: FINISHED
      Progress: ETL:N/A; LOAD:100%
          Type: BROKER
       EtlInfo: unselected.rows=4; dpp.abnorm.ALL=15; dpp.norm.ALL=28133376
      TaskInfo: cluster:N/A; timeout(s):10800; max_filter_ratio:5.0E-5
      ErrorMsg: N/A
    CreateTime: 2019-07-27 11:46:42
  EtlStartTime: 2019-07-27 11:46:44
 EtlFinishTime: 2019-07-27 11:46:44
 LoadStartTime: 2019-07-27 11:46:44
LoadFinishTime: 2019-07-27 11:50:16
           URL: http://192.168.**.**:8040/api/_load_error_log?file=__shard_4/error_log_insert_stmt_...
    JobDetails: {"Unfinished backends":{"9c3441027ff948a0-8287923329a2****":[10002]},"ScannedRows":2390016,"TaskNumber":1,"All backends":{"9c3441027ff948a0-8287923329a2****":[10002]},"FileNumber":1,"FileSize":1073741824}

Cancel an import job

Cancel a job that is in PENDING or LOADING state:

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

To view the full syntax, run HELP CANCEL LOAD.

End-to-end example: load from OSS

This example loads TPC-H lineitem data from OSS into StarRocks.

Step 1: Create the destination table.

CREATE TABLE lineitem (
  l_orderkey      bigint,
  l_partkey       bigint,
  l_suppkey       bigint,
  l_linenumber    int,
  l_quantity      double,
  l_extendedprice double,
  l_discount      double,
  l_tax           double,
  l_returnflag    string,
  l_linestatus    string,
  l_shipdate      date,
  l_commitdate    date,
  l_receiptdate   date,
  l_shipinstruct  string,
  l_shipmode      string,
  l_comment       string
)
ENGINE = OLAP
DUPLICATE KEY(l_orderkey)
DISTRIBUTED BY HASH(l_orderkey) BUCKETS 96
PROPERTIES (
  "replication_num" = "1"
);

Step 2: Submit the import job.

LOAD LABEL tpch.lineitem
(
    DATA INFILE("oss://xxx/tpc_h/sf1/lineitem.tbl")
    INTO TABLE `lineitem`
    COLUMNS TERMINATED BY '|'
    (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice,
     l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate,
     l_receiptdate, l_shipinstruct, l_shipmode, l_comment)
)

Step 3: Monitor the job.

SHOW LOAD WHERE label = 'lineitem'\G

When the job succeeds, the output shows State: FINISHED and Progress: ETL:100%; LOAD:100%.

Step 4: Verify the data.

-- Check total row count
SELECT COUNT(*) FROM lineitem;

-- Preview the first two rows
SELECT * FROM lineitem LIMIT 2;

Concurrency

A Broker Load job is divided into tasks, then tasks are divided into instances distributed across backends in parallel.

Task splitting: A job produces one task per data_desc clause that points to a different source address or a different set of partitions.

Instance splitting: Each task is split into instances based on three frontend parameters:

Total instances = min(file_size / min_bytes_per_broker_scanner, max_broker_concurrency, load_parallel_instance_num × number of backends)

In practice, most jobs have a single data_desc clause and therefore a single task. That task is split into as many instances as there are backends, with each instance running on a different backend.

Troubleshooting

What's next

• Query the StarRocks documentation for advanced LOAD LABEL options and usage of HELP BROKER LOAD

• Monitor ongoing jobs with SHOW LOAD and review backend logs for performance tuning

• Adjust min_bytes_per_broker_scanner, max_broker_concurrency, and load_parallel_instance_num on the frontend to tune import concurrency for your cluster