Ingest Real-Time Data into Doris via Stream Load HTTP PUT - E-MapReduce

Stream Load is a synchronous HTTP-based method for importing local files or data streams into Apache Doris. Submit data via an HTTP PUT request and get the import result immediately — no polling required.

Supported data formats: CSV and JSON.

How it works

When you submit an import job:

If you send the request to the frontend (FE), the FE redirects it to a coordinator backend (BE) via HTTP redirect.
If you send the request directly to a BE, that node acts as the coordinator BE.
The coordinator BE receives the data and distributes it to other BEs.
After the import completes, the coordinator BE returns the result in the HTTP response.

Important

Stream Load is synchronous. Import results are not recorded in Doris, so SHOW LOAD does not display Stream Load jobs. Check the HTTP response directly to determine success or failure.

Quick start

The following example imports a local CSV file into a Doris table. Replace the placeholders and run the command directly.

curl --location-trusted \
  -u <user>:<passwd> \
  -H "label:my-first-load" \
  -H "column_separator:," \
  -T /path/to/data.csv \
  -XPUT http://<fe-host>:<http-port>/api/<db>/<table>/_stream_load

A successful response looks like this:

{
    "TxnId": 1003,
    "Label": "my-first-load",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 1000000,
    "NumberLoadedRows": 1000000,
    "NumberFilteredRows": 1,
    "NumberUnselectedRows": 0,
    "LoadBytes": 40888898,
    "LoadTimeMs": 2144,
    "BeginTxnTimeMs": 1,
    "StreamLoadPutTimeMs": 2,
    "ReadDataTimeMs": 325,
    "WriteDataTimeMs": 1933,
    "CommitAndPublishTimeMs": 106,
    "ErrorURL": "http://192.168.x.x:8042/api/_load_error_log?file=__shard_0/error_log_insert_stmt_****"
}

Submit an import job

Prerequisites

Before you begin, ensure that you have:

A running Doris cluster with at least one FE and one BE
The host and HTTP port of an FE or BE node
A database and table already created in Doris
Credentials with import permissions for the target database

Command syntax

curl --location-trusted -u <user>:<passwd> \
  [-H "<key>:<value>" ...] \
  -T <data-file> \
  -XPUT http://<fe-host>:<http-port>/api/<db>/<table>/_stream_load

Pass all import parameters as HTTP headers using -H "key:value". To view the full syntax, run HELP STREAM LOAD in Doris.

Request parameters

Authentication

Parameter	Description
`user:passwd`	Credentials for basic access authentication. Doris verifies your identity and import permissions based on these credentials.

Job control

Parameter	Default	Description
`label`	Auto-generated	A unique identifier for the import job within a database. Use the same label for the same batch of data to enforce at-most-once semantics — duplicate requests with an existing label are accepted only once. A label in `CANCELLED` state can be reused.
`max_filter_ratio`	`0`	The maximum filter ratio allowed before the job fails. Valid values: `0`–`1`. Set to a value greater than `0` to allow erroneous rows to be skipped. Calculated as: `dpp.abnorm.ALL / (dpp.abnorm.ALL + dpp.norm.ALL) > max_filter_ratio`.
`where`	—	A WHERE clause to filter source rows. Filtered rows are counted in `NumberUnselectedRows` and are not included in the error rate calculation.
`exec_mem_limit`	`2147483648` (2 GB)	Memory limit for the import job, in bytes.
`strict_mode`	Disabled	Set to `true` to enable strict mode. In strict mode, rows where a column value is non-null in the source but converts to null after column type conversion are filtered out. Strict mode does not apply to columns whose null values come from function evaluation, or to columns where the converted value is out of range for the target type (for example, a source value of `10` for a `DECIMAL(1,0)` column).

Data mapping

Parameter	Default	Description
`column_separator`	`\t`	The column delimiter. For non-printable characters, use hexadecimal format with the `\x` prefix (for example, `\x01` for a Hive file). Multi-character delimiters are supported.
`line_delimiter`	`\n`	The row delimiter. Multi-character delimiters are supported.
`columns`	—	Column mapping and expression transformation. Supports column reordering and function-based transformation — the same expression syntax as a query statement.
`Partitions`	—	Target partitions. Rows that do not belong to the specified partitions are excluded; excluded rows are counted in `dpp.abnorm.ALL`.

Data merging

Parameter	Default	Values
`merge_type`	`APPEND`	`APPEND`: adds this batch to existing data. `DELETE`: deletes rows whose keys match this batch. `MERGE`: applies `DELETE` semantics to rows that match the DELETE condition, and `APPEND` semantics to the rest.

Two-phase commit

Two-phase commit lets you write data and make it visible in two separate steps. Data is written but remains invisible (transaction state: PRECOMMITTED) until you manually trigger a commit.

To enable two-phase commit:

Set disable_stream_load_2pc=false in be.conf.
Set two_phase_commit:true in the request header.

Step 1: Submit the import job with two-phase commit enabled.

curl --location-trusted -u <user>:<passwd> \
  -H "two_phase_commit:true" \
  -T test.txt \
  http://<fe-host>:<http-port>/api/<db>/<table>/_stream_load

The response includes the transaction ID (TxnId) and confirms "TwoPhaseCommit": "true". Data is written but not yet visible.

Step 2: Commit or abort the transaction.

Use the TxnId from the response to manually trigger a commit operation (makes data visible) or an abort operation (discards the written data).

View import results

Return value fields

Field	Description
`TxnId`	Transaction ID for the import job. The transaction ID can be fully managed by Alibaba Cloud.
`Label`	The label of the import job.
`Status`	`Success`: import succeeded. `Publish Timeout`: import is complete but data visibility may be delayed — no retry needed. `Label Already Exists`: change the label and retry. `Fail`: import failed.
`ExistingJobStatus`	Status of the job associated with the existing label. Shown only when `Status` is `Label Already Exists`. Values: `RUNNING` or `FINISHED`.
`Message`	Error message, if any.
`NumberTotalRows`	Total rows processed.
`NumberLoadedRows`	Rows successfully imported.
`NumberFilteredRows`	Rows that failed to import.
`NumberUnselectedRows`	Rows filtered by the WHERE clause.
`LoadBytes`	Bytes imported.
`LoadTimeMs`	Total import duration, in milliseconds.
`BeginTxnTimeMs`	Time to start the FE transaction, in milliseconds.
`StreamLoadPutTimeMs`	Time to get the FE execution plan, in milliseconds.
`ReadDataTimeMs`	Time to read data, in milliseconds.
`WriteDataTimeMs`	Time to write data, in milliseconds.
`CommitAndPublishTimeMs`	Time for the FE to commit and publish the transaction, in milliseconds.
`ErrorURL`	URL to inspect rows that failed to import.

View historical import jobs

Run SHOW STREAM LOAD to query completed import jobs.

By default, BEs do not record Stream Load job history. To enable recording, set enable_stream_load_record to true in the BE configuration. For details, see Configuration items of backend nodes.

Cancel an import job

Stream Load jobs cannot be cancelled manually. The system automatically cancels a job if a timeout or import error occurs.

System configuration

Stream Load behavior is controlled at two levels. Operations teams set cluster-wide defaults in FE and BE configuration files. Individual jobs can override the per-job timeout by setting a separate timeout period in the HTTP request header.

FE configuration

Parameter	Default	Description
`stream_load_default_timeout_second`	`600`	Global default timeout for import jobs, in seconds. If a job does not complete within this period, the system cancels it and sets the status to `CANCELLED`. Override per job by setting a separate timeout period in the HTTP request header, or update this parameter to change the cluster-wide default.

BE configuration

Parameter	Default	Description
`streaming_load_max_mb`	`10240` (10 GB)	Maximum data size per import job, in MB. If your source file exceeds this limit, increase this parameter before submitting the job.

Best practices

When to use Stream Load

Stream Load works best when the source file is already in memory or on a local disk and you need the import result synchronously.

Recommended batch size

Import 1 GB to 10 GB per job. The default maximum is 10 GB (streaming_load_max_mb = 10240). For larger files, increase streaming_load_max_mb in be.conf before submitting.

Concurrent Stream Load jobs are not limited by cluster size.

Timeout planning

Use this formula to estimate the timeout you need:

Timeout (seconds) = File size (MB) / 10 MB/s

The actual throughput depends on your cluster. Adjust the formula based on observed performance.

If the estimated timeout exceeds the default 600 seconds, set a longer value — either per job by specifying a timeout period in the HTTP request header, or globally via stream_load_default_timeout_second in fe.conf.

Complete example

Scenario: Import a 15 GB file from /home/store_sales on a local disk into the store_sales table in the bj_sales database.

Step 1: Increase the BE size limit.

The file exceeds the 10 GB default. Add the following line to be.conf and restart the BE:

streaming_load_max_mb = 16000

Step 2: Calculate the required timeout.

15,000 MB / 10 MB/s = 1,500 seconds

Since 1,500 seconds exceeds the 600-second default, update fe.conf:

stream_load_default_timeout_second = 1500

Step 3: Submit the import job.

curl --location-trusted \
  -u user:password \
  -H "label:abc" \
  -T /home/store_sales \
  -XPUT http://abc.com:8030/api/bj_sales/store_sales/_stream_load