使用Stream Load方式导入数据至Doris集群

基本原理

下图展示了Stream load的主要流程，省略了一些导入细节。

  ^      +
                         |      |
                         |      | 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 |
                       +---+       +---+      +---+

Stream load中，Doris会选定一个节点作为Coordinator节点。该节点负责接数据并分发数据到其他数据节点。您可以通过HTTP协议提交导入命令。如果提交到FE，则FE会通过HTTP redirect指令将请求转发给某一个BE。用户也可以直接提交导入命令给某一指定BE。导入的最终结果由Coordinator BE返回给用户。

基本操作

在执行数据导入Stream Load操作之前，必须确保Doris集群的安全组端口开放，即8030和8040端口，否则Stream Load操作将会连接超时。

• 创建导入

  Stream Load通过HTTP协议提交和传输数据。这里通过curl命令展示如何提交导入。

  用户也可以通过其他HTTP client进行操作。

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

• Header中支持属性见下面的‘导入任务参数’说明。
• 格式为：-H "key1:value1"。
• port：HTTP的端口。

创建导入任务的详细语法可以通过HELP STREAM LOAD命令查看。Stream Load中所有与导入任务相关的参数均设置在Header中。相关参数描述如下表所示。

• 示例1，CSV数据格式导入。
  • 创建Doris表

    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'
    );

  • 准备数据表sequencedata01.csv。

    表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

  • 执行curl命令load数据。

    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

  • 返回结果。

    由于Stream load是一种同步的导入方式，所以导入的结果会通过创建导入的返回值直接返回给用户。

    {
        "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"
    }

    • Stream load导入结果参数如下表。

      表3 参数说明

      参数	说明
      TxnId	导入的事务ID。用户可不感知。
      Label	导入Label。由用户指定或系统自动生成。
      Status	导入完成状态。 Success：表示导入成功。 Publish Timeout：该状态也表示导入已经完成，只是数据可能会延迟可见，无需重试。 Label Already Exists：Label重复，需更换Label。 Fail：导入失败。
      ExistingJobStatus	已存在的Label对应的导入作业的状态。 这个字段只有在当Status为“Label Already Exists”时才会显示。用户可以通过这个状态，知晓已存在Label对应的导入作业的状态。“RUNNING”表示作业还在执行，“FINISHED”表示作业成功。
      Message	导入错误信息。
      NumberTotalRows	导入总处理的行数。
      NumberLoadedRows	成功导入的行数。
      NumberFilteredRows	数据质量不合格的行数。
      NumberUnselectedRows	被where条件过滤的行数。
      LoadBytes	导入的字节数。
      LoadTimeMs	导入完成时间。单位毫秒。
      BeginTxnTimeMs	向Fe请求开始一个事务所花费的时间，单位毫秒。
      StreamLoadPutTimeMs	向Fe请求获取导入数据执行计划所花费的时间，单位毫秒。
      ReadDataTimeMs	读取数据所花费的时间，单位毫秒。
      WriteDataTimeMs	执行写入数据操作所花费的时间，单位毫秒。
      CommitAndPublishTimeMs	向Fe请求提交并且发布事务所花费的时间，单位毫秒。
      ErrorURL	如果有数据质量问题，通过访问这个URL查看具体错误行。

      

      由于Stream load是同步的导入方式，所以并不会在Doris系统中记录导入信息，用户无法异步的通过查看导入命令看到Stream load。使用时需监听创建导入请求的返回值获取导入结果。

• 示例2，json数据格式导入。

  准备json格式数据并保存为testjson.json，并将json数据上传至doris客户端:

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

  • 创建Doris表。

    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'
    );

  • curl命令进行load数据。

    curl --location-trusted -u admin:{doris集群密码} -H 'format: json' -T testjson.json  https://fe_host:http_port/api/{doris数据库}/doris_testjson01/_stream_load -k

  • 查询数据。

    select * from doris_testjson01;

取消导入

用户无法手动取消Stream Load，Stream Load在超时或者导入错误后会被系统自动取消。

查看Stream load

用户可以通过show stream load来查看已经完成的stream load任务。