Importing Data to a Doris Cluster with Stream Load

Basic Principles

The following shows the main flow of Stream Load, omitting some import details.

  ^      +
                         |      |
                         |      | 1A. User submit load to FE
                         |      |
                         |   +--v-----------+
                         |   | FE           |
5. Return result to user |   +--+-----------+
                         |      |
                         |      | 2. Redirect to BE
                         |      |
                         |   +--v-----------+
                         +---+Coordinator BE| 1B. User submit load to BE
                             +-+-----+----+-+
                               |     |    |
                         +-----+     |    +-----+
                         |           |          | 3. Distribute data
                         |           |          |
                       +-v-+       +-v-+      +-v-+
                       |BE |       |BE |      |BE |
                       +---+       +---+      +---+

In Stream Load, Doris selects a node as the Coordinator node. This node receives data and distributes data to other data nodes. You can submit the import command through HTTP. If the command is submitted to the FE node, the FE node forwards the request to a BE node through the HTTP redirect command. You can also directly submit the import command to a specified BE node. The Coordinator BE node returns the final import result.

Basic Operations

Before using Stream Load to import data, ensure that the security group ports of the Doris cluster (ports 8030 and 8040) are enabled. Otherwise, the connection to Stream Load will time out.

• Creating an import job

  Stream Load submits and transfers data through HTTP. Here, the curl command shows how to submit an import.

  You can also perform operations through other HTTP clients.

  curl --location-trusted -u user:passwd [-H ""...] -T data.file -XPUT http://fe_host:http_port/api/{db}/{table}/_stream_load

• The properties supported in the header are described in the import parameters below.
• The format is -H "Key 1:Value 1".
• port specifies the HTTP port number.

You can run HELP STREAM LOAD to view the detailed syntax for creating an import job. All parameters related to Stream Load import jobs are set in the header. The following table describes related parameters.

• Example 1: Import data in CSV format.
  • Create a Doris table.

    CREATE TABLE cloudtable0327.doris_streameload_test01
    (
    user_id bigint,
    date date,
    group_id bigint,
    modify_date date,
    keyword VARCHAR(128)
    )
    UNIQUE KEY(user_id, date, group_id)
    DISTRIBUTED BY HASH (user_id) BUCKETS 32
    PROPERTIES(
    'function_column.sequence_col' = 'modify_date',
    'replication_num' = '3',
    'in_memory' = 'false'
    );

  • Prepare the data table sequencedata01.csv.

    Table 2 sequencedata01.csv

    1	2020-02-22	1	2020-02-21	a
    1	2020-02-22	1	2020-02-22	b
    1	2020-02-22	1	2020-03-05	c
    1	2020-02-22	1	2020-02-26	d
    1	2020-02-22	1	2020-02-23	e
    1	2020-02-22	1	2020-02-24	b

  • Run the curl command to load data.

    curl -k --location-trusted -u admin:passwd -T sequencedata01.csv -H 'column_separator:,' https://{fe_host}:{http_port}/api/cloudtable0327/doris_streameload_test01/_stream_load

  • View the returned result.

    Since Stream Load imports data synchronously, the result of the import is directly returned to the user through the return value of the import.

    {
        "TxnId": 1003,
        "Label": "b6f3bc78-0d2c-45d9-9e4c-faa0a0149bee",
        "Status": "Success",
        "ExistingJobStatus": "FINISHED", // optional
        "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://fe_host:http_port/api/_load_error_log?file=__shard_0/error_log_insert_stmt_db18266d4d9b4ee5-abb00ddd64bdf005_db18266d4d9b4ee5_abb00ddd64bdf005"
    }

    • The following table describes the parameters of the Stream Load import result.

      Table 3 Parameters

      Parameter	Description
      TxnId	Import transaction ID, which is invisible to users.
      Label	Import label, which is user-defined or automatically generated by the system.
      Status	Import completion status. Success: indicates that the import is successful. Publish Timeout: indicates that the import has been completed but data visibility may be delayed. You do not need to retry. Label Already Exists: indicates that the label is duplicate and needs to be replaced. Fail: indicates that the import fails.
      ExistingJobStatus	Status of the import job corresponding to the existing label. This field is displayed only when the status is Label Already Exists. You can view the status of the import job corresponding to the existing label. RUNNING indicates that the job is still being executed, and FINISHED indicates that the job is successful.
      Message	Import error information.
      NumberTotalRows	Total number of rows processed during the import.
      NumberLoadedRows	Number of rows successfully imported.
      NumberFilteredRows	Number of rows whose data quality is unqualified.
      NumberUnselectedRows	Number of rows filtered by the where condition.
      LoadBytes	Number of imported bytes.
      LoadTimeMs	Time when the import is complete, in milliseconds.
      BeginTxnTimeMs	Time taken to request the FE node to start a transaction, in milliseconds.
      StreamLoadPutTimeMs	Time taken to request the FE node to obtain the execution plan for importing data, in milliseconds.
      ReadDataTimeMs	Time taken to read data, in milliseconds.
      WriteDataTimeMs	Time taken to write data, in milliseconds.
      CommitAndPublishTimeMs	Time taken to submit and publish a transaction to the FE node, in milliseconds.
      ErrorURL	URL to view a specific error line if there are data quality problems.

      

      Since Stream Load is a synchronous import mode, import information will not be recorded in Doris. You cannot see Stream Load asynchronously by checking import commands. You need to view the return value of the import request to obtain the import result.

• Example 2: Import data in JSON format.

  Prepare data in JSON format and save the data in the testjson.json file. Upload the JSON data to the Doris client.

  {"id": 100, "city": "B", "code" : 1}

  • Create a Doris table.

    CREATE TABLE `doris_testjson01` (
      `id` varchar(32) NOT NULL,
      `city` ARRAY<int(11)>,
      `code` int(11)
    ) ENGINE=OLAP
    DUPLICATE KEY(`id`)
    COMMENT "OLAP"
    DISTRIBUTED BY HASH(`id`) BUCKETS 1
    PROPERTIES (
    'replication_allocation' = 'tag.location.default: 3',
    'in_memory' = 'false',
    'storage_format' = 'V2'
    );

  • Run the curl command to load data.

    curl --location-trusted -u admin:{Doris cluster password} -H 'format: json' -T testjson.json https://fe_host:http_port/api/{Doris database}/doris_testjson01/_stream_load -k

  • Query data.

    select * from doris_testjson01;

Canceling Data Import

You cannot manually cancel Stream Load jobs. It will be automatically canceled by the system upon a timeout or import error.

Viewing Stream Load Jobs

You can run the show stream load command to view completed Stream Load jobs.