Stream Load

背景信息

Stream Load是属于同步接口的导入方式，用户通过发送HTTP请求将本地文件或数据流导入到云数据库 SelectDB 版实例中。Stream load执行并返回导入结果，用户可直接通过请求的返回体判断本次导入是否成功。

Stream Load主要适用于导入本地文件或通过程序导入数据流中的数据，支持的数据格式包括：CSV（文本）、JSON、PARQUET和ORC。

创建导入

Stream Load通过HTTP协议提交和传输数据，这里通过curl命令展示如何提交导入。用户也可以通过其他HTTP Client进行操作。

语法

# Header中支持的属性，请参见下面的参数说明。
# 格式为: -H "key1:value1"。
curl --location-trusted -u user:passwd [-H ""...] -T data.file -XPUT http://host:port/api/{db}/{table}/_stream_load 

参数说明

由于Stream Load使用HTTP协议，所以导入任务有关的参数主要设置在Header中。常用的导入参数如下。

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

{
    "TxnId": 17,
    "Label": "707717c0-271a-44c5-be0b-4e71bfeacaa5",
    "Status": "Success",
    "Message": "OK",
    "NumberTotalRows": 5,
    "NumberLoadedRows": 5,
    "NumberFilteredRows": 0,
    "NumberUnselectedRows": 0,
    "LoadBytes": 28,
    "LoadTimeMs": 27,
    "BeginTxnTimeMs": 0,
    "StreamLoadPutTimeMs": 2,
    "ReadDataTimeMs": 0,
    "WriteDataTimeMs": 3,
    "CommitAndPublishTimeMs": 18
}

返回结果参数说明如下。

基础的使用样例如下：

curl --location-trusted -u admin:admin_123 -T data.csv -H "label:123" http://selectdb-cn-h033cjs****-fe.selectdbfe.pre.rds.aliyuncs.com:8080/api/test_db/test_table/_stream_load

取消导入

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

查看Stream Load

您可以通过show stream load来查看已经完成的Stream load任务。默认BE（BackEnd）不保留Stream Load的启用记录，如果您要查看则需要在BE上启用记录，配置参数为：enable_stream_load_record=true，具体操作请参见BE配置项。​

使用示例

• 将本地文件test.data中的数据导入到数据库test_db中的test_table表，使用Label用于去重，指定超时时间为100秒。

   curl --location-trusted -u root -H "label:123" -H "timeout:100" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 将本地文件test.data中的数据导入到数据库test_db中的test_table表，使用label用于去重，并且只导入k1等于20180601的数据。

  curl --location-trusted -u root -H "label:123" -H "where: k1=20180601" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 将本地文件test.data中的数据导入到数据库test_db中的test_table表，允许20%的错误率（用户是defalut_cluster中的）。

  curl --location-trusted -u root -H "label:123" -H "max_filter_ratio:0.2" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 将本地文件test.data中的数据导入到数据库test_db中的test_table表，允许20%的错误率，并且指定文件的列名。

  curl --location-trusted -u root -H "label:123" -H "max_filter_ratio:0.2" -H "columns: k2, k1, v1" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 将本地文件test.data中的数据导入到数据库test_db中的test_table表中的p1、p2分区，允许20%的错误率。

  curl --location-trusted -u root -H "label:123" -H "max_filter_ratio:0.2" -H "partitions: p1, p2" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 使用流式方式（streaming）导入。

  seq 1 10 | awk '{OFS="\t"}{print $1, $1 * 10}' | curl --location-trusted -u root -T - http://host:port/api/test_db/test_table/_stream_load

• 导入含有HLL列的表，可以是表中的列或者数据中的列用于生成HLL列，也可使用hll_empty补充数据中没有的列。

  curl --location-trusted -u root -H "columns: k1, k2, v1=hll_hash(k1), v2=hll_empty()" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 导入数据进行严格模式过滤，并设置时区为Africa/Abidjan。

  curl --location-trusted -u root -H "strict_mode: true" -H "timezone: Africa/Abidjan" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 删除与这批导入key相同的数据。

  curl --location-trusted -u root -H "merge_type: DELETE" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 将这批数据中与flag列为1的数据相匹配的列删除，其他行正常追加。

  curl --location-trusted -u root: -H "column_separator:," -H "columns: siteid, citycode, username, pv, flag" -H "merge_type: MERGE" -H "delete: flag=1" -T test.data http://host:port/api/test_db/test_table/_stream_load

• 导入数据到含有sequence列的UNIQUE_KEYS表中。

  curl --location-trusted -u root -H "columns: k1,k2,source_sequence,v1,v2" -H "function_column.sequence_col: source_sequence" -T test.data http://host:port/api/test_db/test_table/_stream_load

相关系统配置

FE配置

stream_load_default_timeout_second：导入任务的超时时间，单位：秒。默认的timeout时间为600秒，导入任务在设定的timeout时间内未完成则会被系统取消，变成CANCELLED。如果导入的源文件无法在规定时间内完成导入，您可以在Stream load请求中设置单独的超时时间或者调整FE的参数stream_load_default_timeout_second来设置全局的默认超时时间。

BE配置

streaming_load_max_mb：Stream load的最大导入大小，单位：MB，默认值为10240 MB。如果您的原始文件超过该值，则需要调整BE参数streaming_load_max_mb。