Stream load - E-MapReduce - Alibaba Cloud Documentation Center

Background

Stream Load is a synchronous loading method that uses an HTTP request to load data from a local file or a data stream into StarRocks. Because the operation is synchronous, the HTTP response immediately returns the load result, letting you verify its success. Stream Load supports CSV and JSON formats, and each load must be smaller than 10 GB.

Creating a load task

Stream Load imports data via HTTP. This topic shows how to submit an import job with the curl command. You can also use other HTTP clients.

Syntax

curl --location-trusted -u <username>:<password> -XPUT <url>
(
    data_desc
)
[opt_properties]

Note

In the Expect request header, we recommend specifying 100-continue, i.e., "Expect:100-continue". This prevents unnecessary data transfer and reduces resource overhead if the server rejects the import task request.
In StarRocks, some words are SQL reserved keywords and cannot be used directly in SQL statements. To use a reserved keyword in an SQL statement, you must enclose it in backticks (). For more information about keywords, see Keywords.

Parameters

<username>:<password>: The username and password for your StarRocks cluster. This parameter is required. If the account has no password, use <username>:.
PUT: The HTTP request method. This method is required. Stream Load only supports PUT.

<url>: The URL of the destination StarRocks table. This parameter is required. Use the following format: http://<fe_host>:<fe_http_port>/api/<database_name>/<table_name>/_stream_load.

Parameter	Required	Description
`<fe_host>`	Yes	The IP address of an FE in the StarRocks cluster.
`<fe_http_port>`	Yes	The HTTP port of the FE. Default: 18030. You can find the port number by searching for the http_port parameter on the Configure tab of the StarRocks Services page. Additionally, you can run the `SHOW FRONTENDS` command to view the IP addresses and HTTP ports of the FE nodes.
`<database_name>`	Yes	The name of the database containing the destination table.
`<table_name>`	Yes	The name of the destination StarRocks table.

desc: Describes the properties of the source data file, such as the file name, format, column separator, row delimiter, target partition, and the column mapping to the StarRocks table. The format is as follows.

-T <file_path>
-H "format: CSV | JSON"
-H "column_separator: <column_separator>"
-H "row_delimiter: <row_delimiter>"
-H "columns: <column1_name>[, <column2_name>, ... ]"
-H "partitions: <partition1_name>[, <partition2_name>, ...]"
-H "temporary_partitions: <temporary_partition1_name>[, <temporary_partition2_name>, ...]"
-H "jsonpaths: [ \"<json_path1>\"[, \"<json_path2>\", ...] ]"
-H "strip_outer_array: true | false"
-H "json_root: <json_path>"
-H "ignore_json_size: true | false"
-H "compression: <compression_algorithm> | Content-Encoding: <compression_algorithm>"

The parameters in data_desc can be divided into three types: common parameters, parameters for CSV, and parameters for JSON.

Common parameters

Parameter	Required	Description
`<file_path>`	Yes	The path to the source data file. The filename can optionally include an extension.
`format`	No	The data format. Valid values: `CSV` and `JSON`. Default: `CSV`.
`partitions`	No	The partitions to load data into. If this parameter is not specified, data is loaded into all table partitions.
`temporary_partitions`	No	The temporary partitions to load data into.
`columns`	No	Specifies the column mapping between the source data file and the StarRocks table. This parameter is not required if the columns in the source data file match the order of the columns in the destination table. If source file columns do not match the table schema, you must use this parameter to define the mapping. You can map columns in two ways: Example 1: The table has columns `c1, c2, c3`, but the source file columns are in the order `c3, c2, c1`. You must specify `-H "columns: c3, c2, c1"`. Example 2: The table has three columns `c1, c2, c3`, and the source file's first three columns map to them, but the file has a fourth, unneeded column. In this case, specify `-H "columns: c1, c2, c3, temp"`, using a placeholder for the extra column. Example 3: The table has three columns `year, month, day`, but the source file has only a single timestamp column in the `2018-06-01 01:02:03` format. You can derive the table columns by specifying `-H "columns: col, year = year(col), month=month(col), day=day(col)"`.

CSV-specific parameters

Parameter	Required	Description
`column_separator`	No	The column separator in the source data file. Default: `\t`. For non-printable characters, prefix with `\x` and use a hexadecimal value. For example, for the Hive separator `\x01`, specify `-H "column_separator:\x01"`.
`row_delimiter`	No	The row delimiter in the source data file. Default: `\n`. Important When using `curl`, the shell might interpret \n as a literal string instead of a newline character. In Bash, use the `$'string'` syntax to ensure escape sequences like `\n` and `\t` are interpreted correctly. For example: `-H $'row_delimiter:\n'`.
`skip_header`	No	The number of header rows to skip at the beginning of the CSV file. The value must be an integer. Default: 0. Use this parameter to ignore metadata rows, such as column names or types, in the CSV file. For example, setting this parameter to 1 skips the first row. The row delimiter used here must match the one specified in your load command.
`where`	No	A condition to filter which rows to load. For example, to load only rows where the k1 column equals 20180601, specify `-H "where: k1 = 20180601"`.
`max_filter_ratio`	No	The maximum ratio of rows that can be discarded due to data quality errors. Default: `0` (zero tolerance). Note This does not apply to rows filtered by the where condition.
`partitions`	No	The destination partitions for the load job. We recommend specifying this parameter if you can determine the target partitions. Data for other partitions will be discarded. For example, to load data into partitions p1 and p2, specify `-H "partitions: p1, p2"`.
`timeout`	No	The timeout for the load job, in seconds. Default: 600. The value must be between 1 and 259200.
`strict_mode`	No	Enables or disables strict mode for this load job. Valid values: `false` (default): Strict mode is disabled. `true`: Strict mode is enabled. When enabled, strict filtering is applied to column type conversions during loading.
`timezone`	No	The time zone for the load job. Default: `Asia/Shanghai` (UTC+08:00). This parameter affects all time-zone-related functions used in the load.
`exec_mem_limit`	No	The memory limit for the load job. Default: 2 GB.

JSON-specific parameters

Parameter	Required	Description
`jsonpaths`	No	Specifies the fields to extract from the JSON data. This is required only when you need to map specific source fields to table columns. The value must be in JSON format.
`strip_outer_array`	No	Specifies whether to strip the outermost array structure. Valid values: `false` (default): Retains the original JSON structure. The entire JSON array is imported as a single value. For example, for the data `[{"k1" : 1, "k2" : 2},{"k1" : 3, "k2" : 4}]`, setting `strip_outer_array` to `false` imports the data as a single array. true: If the data to be loaded is a JSON array, you need to set `strip_outer_array` to true. For example, for the data `[{"k1" : 1, "k2" : 2},{"k1" : 3, "k2" : 4}]`, setting `strip_outer_array` to `true` imports the data as two separate rows.
`json_root`	No	Specifies the root element of the JSON data to import. This parameter is required only when you load JSON data by matching fields. The value must be a valid JsonPath string. Default: empty, which means the entire JSON file is imported.
`ignore_json_size`	No	Specifies whether to skip the size check for the JSON body in the HTTP request. Note By default, the JSON body size in an HTTP request cannot exceed 100 MB. If the size exceeds this limit, an error is returned: `"The size of this batch exceed the max size [104857600] of json type data data [8617627793]. Set ignore_json_size to skip check, although it may lead huge memory consuming."` To load larger files, add the header `"ignore_json_size:true"` to your request to bypass this check.
`compression, Content-Encoding`	No	The compression algorithm used for the request body. Supported algorithms: GZIP, BZIP2, LZ4_FRAME, and ZSTD. For example: `curl --location-trusted -u root: -v 'http://127.0.0.1:18030/api/db0/tbl_simple/_stream_load' \-X PUT -H "expect:100-continue" \-H 'format: json' -H 'compression: lz4_frame' -T ./b.json.lz4`.

opt_properties: Specifies optional parameters for the import. The specified parameters apply to the entire import task.

The format is as follows:

-H "label: <label_name>"
-H "where: <condition1>[, <condition2>, ...]"
-H "max_filter_ratio: <num>"
-H "timeout: <num>"
-H "strict_mode: true | false"
-H "timezone: <string>"
-H "load_mem_limit: <num>"
-H "partial_update: true | false"
-H "partial_update_mode: row | column"
-H "merge_condition: <column_name>"

Parameter	Required	Description
`label`	No	A unique label for the load job. StarRocks prevents using the same label for more than one successful job. You can use labels to prevent duplicate loads. StarRocks retains labels from successful jobs for 30 minutes.
`where`	No	Specifies a filter condition. StarRocks applies this filter to the data after any transformations from the `columns` parameter. Only data that satisfies the `where` clause is imported. For example, to import only the rows where the k1 column is 20180601, specify `-H "where: k1 = 20180601"`.
`max_filter_ratio`	No	The maximum ratio of rows that can be discarded due to data quality errors. Default: `0` (zero tolerance). Note This does not apply to rows filtered by the `where` condition.
`log_rejected_record_num`	No	Specifies the maximum number of rejected rows to log due to data quality issues. Supported since v3.1. Valid values: `0`, `-1`, or a positive integer. Default: `0`. `0`: No rejected rows are logged. `-1`: All rejected rows are logged. A positive integer, such as `n`: A maximum of `n` rejected rows are logged on each BE (or CN) node.
`timeout`	No	Load job timeout in seconds. Default: 600. The value must be between 1 and 259200.
`strict_mode`	No	Specifies whether to enable strict mode for this load job. `false` (default): Strict mode is disabled. `true`: Strict mode is enabled.
`timezone`	No	The time zone for the load job. Default: `Asia/Shanghai` (UTC+08:00). This parameter affects all time-zone-related functions used in the load.
`load_mem_limit`	No	The memory limit for the load job. Default: 2 GB.
`partial_update`	No	Enables partial column updates. Valid values: `true` and `false`. Default: `false`.
`partial_update_mode`	No	Specifies the mode for partial updates. Valid values: `row` and `column`. `row` (default): Row mode. Best for real-time updates with many columns in small batches. `column`: Column mode. Best for batch updates of a few columns across many rows. This mode can significantly improve performance. For example, if you update 10 out of 100 columns (10%) for all rows in a table, column mode can improve performance by up to 10x.
`merge_condition`	No	Specifies the column name to use as the update condition. An update to a row only occurs if the new data's value for this column is greater than or equal to the existing value. Note The specified column must not be part of the primary key. This feature is available only for primary key tables.

Example

This example loads data.csv into the example_table table in the load_test database of a StarRocks cluster. For a complete example, see Complete data loading example.

curl --location-trusted -u "root:" \
     -H "Expect:100-continue" \
     -H "label:label2" \
     -H "column_separator: ," \
     -T data.csv -XPUT \
     http://172.17.**.**:18030/api/load_test/example_table/_stream_load

Return value

When a load task completes, it returns the result in JSON format. The following is a sample.

{
    "TxnId": 9,
    "Label": "label2",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 4,
    "NumberLoadedRows": 4,
    "NumberFilteredRows": 0,
    "NumberUnselectedRows": 0,
    "LoadBytes": 45,
    "LoadTimeMs": 235,
    "BeginTxnTimeMs": 101,
    "StreamLoadPlanTimeMs": 102,
    "ReadDataTimeMs": 0,
    "WriteDataTimeMs": 11,
    "CommitAndPublishTimeMs": 19
}

The return value includes the following parameters:

Parameter	Description
`TxnId`	The transaction ID of the load task.
`Label`	The label of the load task.
`Status`	The load status. Valid values are: `Success`: The data is successfully loaded and visible. `Publish Timeout`: The load task is committed, but data visibility might be delayed. Do not retry the task. `Label Already Exists`: Indicates that the `Label` already exists. You must change the `Label`. Fail: The load failed.
`ExistingJobStatus`	This field indicates the status of an import task that corresponds to a label that already exists. The field is displayed only when the `Status` is `Label Already Exists`. You can check this status to determine the state of the import task. `RUNNING`: The task is running. `FINISHED`: The task completed.
`Message`	Details about the load status. If a load fails, this parameter provides the reason.
`NumberTotalRows`	The total number of rows read from the data stream.
`NumberLoadedRows`	The number of rows loaded. This parameter applies only when `Status` is `Success`.
`NumberFilteredRows`	The number of rows filtered out due to poor data quality.
`NumberUnselectedRows`	The number of rows filtered out by the `where` condition.
`LoadBytes`	The size of the source file in bytes.
`LoadTimeMs`	The duration of the load task, in milliseconds.
`BeginTxnTimeMs`	The time taken to start the transaction, in milliseconds.
`StreamLoadPlanTimeMs`	The time taken to generate the execution plan, in milliseconds.
`ReadDataTimeMs`	The time taken to read data, in milliseconds.
`WriteDataTimeMs`	The time taken to write data, in milliseconds.
`ErrorURL`	Returned only when the load fails. You can use `ErrorURL` to view the details of error rows that are filtered out during the import process because they fail data quality checks. When you submit an import task, you can use the optional parameter `log_rejected_record_num` to specify the maximum number of error rows to log. You can run the `curl "url"` command to directly view the information about error rows, or run the `wget "url"` command to export the information. For example, to download the error row information: `wget "http://172.17..:18040/api/_load_error_log?file=error_log_b74dccdcf0ceb4de_e82b2709c6c013ad"` The exported error rows are saved to a local file named `_load_error_log?file=error_log_b74dccdcf0ceb4de_e82b2709c6c013ad`. You can use the `cat _load_error_log?file=error_log_b74dccdcf0ceb4de_e82b2709c6c013ad` command to view the contents of the file. Adjust the load task based on the error information, then retry it.

Cancel load task

You cannot manually cancel a Stream Load task. The system automatically cancels the task if it times out or an import error occurs. Use the ErrorURL in the response to download the error details for troubleshooting.

Data loading example

This example shows how to submit a load task using the curl command.

Create a destination table.

Log in to the master node of the StarRocks cluster using SSH. For more information, see Log in to a cluster.
Run the following command to connect to the StarRocks cluster using a MySQL client.
```
mysql -h127.0.0.1  -P 9030 -uroot
```

Run the following commands to create the database and table.

CREATE DATABASE IF NOT EXISTS load_test;
USE load_test;

CREATE TABLE IF NOT EXISTS example_table (
    id INT,
    name VARCHAR(50),
    age INT
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 3
PROPERTIES (
    "replication_num" = "1"  -- Sets the replication number to 1.
);

Press Ctrl+D to exit the MySQL client.

Prepare test data.

CSV data

For example, create a file named data.csv with the following content:

id,name,age
1,Alice,25
2,Bob,30
3,Charlie,35

JSON data

For example, create a file named json.data with the following content:

{"id":1,"name":"Emily","age":25}
{"id":2,"name":"Benjamin","age":35}
{"id":3,"name":"Olivia","age":28}
{"id":4,"name":"Alexander","age":60}
{"id":5,"name":"Ava","age":17}

Run the following command to submit a load task.

CSV data

curl --location-trusted -u "root:" \
     -H "Expect:100-continue" \
     -H "label:label1" \
     -H "column_separator: ," \
     -T data.csv -XPUT \
     http://172.17.**.**:18030/api/load_test/example_table/_stream_load

JSON data

curl --location-trusted -u "root:" \
     -H "Expect:100-continue" \
     -H "label:label2" \
     -H "format:json" \
     -T json.data -XPUT \
     http://172.17.**.**:18030/api/load_test/example_table/_stream_load

Code integration examples

Developing Stream Load jobs in Java. See stream_load for details.
Integrating Stream Load with Spark. See 01_sparkStreaming2StarRocks for details.