1、Doris数据的导入和导出

1.1 数据导入

导入（Load）功能就是将用户的原始数据导入到 Doris 中。导入成功后，用户即可通过Mysql 客户端查询数据。为适配不同的数据导入需求，Doris 系统提供了 6 种不同的导入方式。每种导入方式支持不同的数据源，存在不同的使用方式（异步，同步）。
所有导入方式都支持 csv 数据格式。其中 Broker load 还支持 parquet 和 orc 数据格式。
1）Broker load
通过 Broker 进程访问并读取外部数据源（如 HDFS）导入到 Doris。用户通过 Mysql协议提交导入作业后，异步执行。通过 SHOW LOAD 命令查看导入结果。
2）Stream load
用户通过 HTTP 协议提交请求并携带原始数据创建导入。主要用于快速将本地文件或数据流中的数据导入到 Doris。导入命令同步返回导入结果。
3）Insert
类似 MySQL 中的 Insert 语句，Doris 提供 INSERT INTO tbl SELECT …; 的方式从Doris 的表中读取数据并导入到另一张表。或者通过 INSERT INTO tbl VALUES(…); 插入单条数据。
4）Multi load
用户通过 HTTP 协议提交多个导入作业。Multi Load 可以保证多个导入作业的原子生效。
5）Routine load
用户通过 MySQL 协议提交例行导入作业，生成一个常驻线程，不间断的从数据源（如Kafka）中读取数据并导入到 Doris 中。
6）通过 S3 协议直接导入
用户通过 S3 协议直接导入数据，用法和 Broker Load 类似。
Broker load 是一个异步的导入方式，支持的数据源取决于 Broker 进程支持的数据源。
用户需要通过 MySQL 协议创建 Broker load 导入，并通过查看导入命令检查导入结果。

1.1.1 Broker Load

1.1.1.1 适用场景

源数据在 Broker 可以访问的存储系统中，如 HDFS。
数据量在几十到百 GB 级别。

1.1.1.2 基本原理

用户在提交导入任务后，FE 会生成对应的 Plan 并根据目前 BE 的个数和文件的大小，将 Plan 分给多个 BE 执行，每个 BE 执行一部分导入数据。

BE 在执行的过程中会从 Broker 拉取数据，在对数据 transform 之后将数据导入系统。所有 BE 均完成导入，由 FE 最终决定导入是否成功。

1.1.1.3 基本语法

LOAD LABEL db_name.label_name 
(data_desc, ...)
WITH BROKER broker_name broker_properties
[PROPERTIES (key1=value1, ... )]
* data_desc:
 DATA INFILE ('file_path', ...)
 [NEGATIVE]
 INTO TABLE tbl_name
 [PARTITION (p1, p2)]
 [COLUMNS TERMINATED BY separator ]
 [(col1, ...)]
 [PRECEDING FILTER predicate]
 [SET (k1=f1(xx), k2=f2(xx))]
 [WHERE predicate]
* broker_properties: 
 (key1=value1, ...)

创建导入的详细语法执行 HELP BROKER LOAD 查看语法帮助。这里主要介绍 Broker load 的创建导入语法中参数意义和注意事项。
1）Label
导入任务的标识。每个导入任务，都有一个在单 database 内部唯一的 Label。Label 是用户在导入命令中自定义的名称。通过这个 Label，用户可以查看对应导入任务的执行情况。

Label 的另一个作用，是防止用户重复导入相同的数据。强烈推荐用户同一批次数据使用相同的 label。这样同一批次数据的重复请求只会被接受一次，保证了 At-Most-Once 语义。

当 Label 对应的导入作业状态为 CANCELLED 时，可以再次使用该 Label 提交导入作业。

2）数据描述类参数
数据描述类参数主要指的是 Broker load 创建导入语句中的属于 data_desc 部分的参数。
每组 data_desc 主要表述了本次导入涉及到的数据源地址，ETL 函数，目标表及分区等信息。
下面主要对数据描述类的部分参数详细解释：
⚫ 多表导入
Broker load 支持一次导入任务涉及多张表，每个 Broker load 导入任务可在多
个 data_desc 声明多张表来实现多表导入。每个单独的 data_desc 还可以指定属于该表的数据源地址。Broker load 保证了单次导入的多张表之间原子性成功或失败。
⚫ negative
data_desc 中还可以设置数据取反导入。这个功能主要用于，当数据表中聚合列的类型都为 SUM 类型时。如果希望撤销某一批导入的数据。则可以通过 negative 参数导入同一批数据。Doris 会自动为这一批数据在聚合列上数据取反，以达到消除同一批数据的功能。
⚫ partition
在 data_desc 中可以指定待导入表的 partition 信息，如果待导入数据不属于指定的partition 则不会被导入。同时，不在指定 Partition 的数据会被认为是错误数据。
⚫ set column mapping
在 data_desc 中的 SET 语句负责设置列函数变换，这里的列函数变换支持所有查询的等值表达式变换。如果原始数据的列和表中的列不一一对应，就需要用到这个属性。
⚫ preceding filter predicate
用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。
⚫ where predicate
在 data_desc 中的 WHERE 语句中负责过滤已经完成 transform 的数据，被 filter 的数据不会进入容忍率的统计中。如果多个 data_desc 中声明了同一张表的多个条件的话，则会merge 同一张表的多个条件，merge 策略是 AND 。
3）导入作业参数
导入作业参数主要指的是 Broker load 创建导入语句中的属于 opt_properties 部分的参数。导入作业参数是作用于整个导入作业的。
下面主要对导入作业参数的部分参数详细解释：
⚫ timeout
导入作业的超时时间(以秒为单位)，用户可以在 opt_properties 中自行设置每个导入的超时时间。导入任务在设定的 timeout 时间内未完成则会被系统取消，变成 CANCELLED。
Broker load 的默认导入超时时间为 4 小时。
通常情况下，用户不需要手动设置导入任务的超时时间。当在默认超时时间内无法完成导入时，可以手动设置任务的超时时间。
推荐超时时间：
总文件大小（MB） / 用户 Doris 集群最慢导入速度(MB/s) > timeout > （（总文件大小(MB) * 待导入的表及相关 Roll up 表的个数） / (10 * 导入并发数） ）
导入并发数见文档最后的导入系统配置说明，公式中的 10 为目前的导入限速 10MB/s。例如一个 1G 的待导入数据，待导入表包含 3 个 Rollup 表，当前的导入并发数为 3。则 timeout 的 最小值为 (1 * 1024 * 3 ) / (10 * 3) = 102 秒
由于每个 Doris 集群的机器环境不同且集群并发的查询任务也不同，所以用户 Doris 集群的最慢导入速度需要用户自己根据历史的导入任务速度进行推测。
⚫ max_filter_ratio
导入任务的最大容忍率，默认为 0 容忍，取值范围是 0~1。当导入的错误率超过该值，则导入失败。
如果用户希望忽略错误的行，可以通过设置这个参数大于 0，来保证导入可以成功。
计算公式为：
max_filter_ratio = (dpp.abnorm.ALL / (dpp.abnorm.ALL + dpp.norm.ALL ) )
dpp.abnorm.ALL 表示数据质量不合格的行数。如类型不匹配，列数不匹配，长度不匹配等等。
dpp.norm.ALL 指的是导入过程中正确数据的条数。可以通过 SHOW LOAD 命令查询导入任务的正确数据量。
原始文件的行数 = dpp.abnorm.ALL + dpp.norm.ALL
⚫ exec_mem_limit
导入内存限制。默认是 2GB。单位为字节。
⚫ strict_mode
Broker load 导入可以开启 strict mode 模式。开启方式为 properties (“strict_mode” = “true”) 。默认的 strict mode 为关闭。strict mode 模式的意思是：对于导入过程中的列类型转换进行严格过滤。严格过滤的策略如下：
① 对于列类型转换来说，如果 strict mode 为 true，则错误的数据将被 filter。这里的错误数据是指：原始数据并不为空值，在参与列类型转换后结果为空值的这一类数据。
② 对于导入的某列由函数变换生成时，strict mode 对其不产生影响。
③ 对于导入的某列类型包含范围限制的，如果原始数据能正常通过类型转换，但无法通过范围限制的，strict mode 对其也不产生影响。例如：如果类型是 decimal(1,0), 原始数据为 10，则属于可以通过类型转换但不在列声明的范围内。这种数据 strict 对其不产生影响。
⚫ merge_type
数据的合并类型，一共支持三种类型 APPEND、DELETE、MERGE 其中，APPEND 是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据 key 相同的所有行，MERGE 语义 需要与 delete 条件联合使用，表示满足 delete 条件的数据按照DELETE 语义处理其余的按照 APPEND 语义处理

1.1.1.4 导入示例

1）Doris 中创建表

create table student_result
(
id int ,
name varchar(50),
age int ,
score decimal(10,4)
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 10;

2）文件上传 HDFS
启动 HDFS 相关服务

hadoop fs -put student.csv /

3）导入数据
csv 文件导入

LOAD LABEL test_db.student_result
(
DATA INFILE("hdfs://my_cluster/student.csv")
INTO TABLE `student_result`
COLUMNS TERMINATED BY ","
FORMAT AS "csv"
(id, name, age, score)
)
WITH BROKER broker_name
(
#开启了 HA 的写法，其他 HDFS 参数可以在这里指定
"dfs.nameservices" = "my_cluster",
"dfs.ha.namenodes.my_cluster" = "nn1,nn2,nn3",
"dfs.namenode.rpc-address.my_cluster.nn1" = "hadoop1:8020",
"dfs.namenode.rpc-address.my_cluster.nn2" = "hadoop2:8020",
"dfs.namenode.rpc-address.my_cluster.nn3" = "hadoop3:8020",
"dfs.client.failover.proxy.provider" = 
"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProx
yProvider"
)
PROPERTIES
(
 "timeout" = "3600"
);
通用文件格式写法
LOAD LABEL test_db.student_result
(
DATA INFILE("hdfs://hadoop1:8020/student.csv")
INTO TABLE `student_result`
COLUMNS TERMINATED BY ","
(c1, c2, c3, c4)
set(
 id=c1,
 name=c2, 
 age=c3,
 score=c4
 )
)
WITH BROKER broker_name
(
#开启了 HA 的写法，其他 HDFS 参数可以在这里指定
"dfs.nameservices" = "my_cluster",
"dfs.ha.namenodes.my_cluster" = "nn1,nn2,nn3",
"dfs.namenode.rpc-address.my_cluster.nn1" = "hadoop1:8020",
"dfs.namenode.rpc-address.my_cluster.nn2" = "hadoop2:8020",
"dfs.namenode.rpc-address.my_cluster.nn3" = "hadoop3:8020",
"dfs.client.failover.proxy.provider" = 
"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProx
yProvider"
)
PROPERTIES
(
 "timeout" = "3600"
);

1.1.1.5 查看导入

Broker load 导入方式由于是异步的，所以用户必须将创建导入的 Label 记录，并且在查看导入命令中使用 Label 来查看导入结果。查看导入命令在所有导入方式中是通用的，具体语法可执行 HELP SHOW LOAD 查看。

mysql> show load order by createtime desc limit 1\G
*************************** 1. row ***************************
 JobId: 76391
 Label: label1
 State: FINISHED
 Progress: ETL:N/A; LOAD:100%
 Type: BROKER
 EtlInfo: unselected.rows=4; dpp.abnorm.ALL=15; 
dpp.norm.ALL=28133376
 TaskInfo: cluster:N/A; timeout(s):10800; 
max_filter_ratio:5.0E-5
 ErrorMsg: N/A
 CreateTime: 2019-07-27 11:46:42
 EtlStartTime: 2019-07-27 11:46:44
EtlFinishTime: 2019-07-27 11:46:44
LoadStartTime: 2019-07-27 11:46:44
LoadFinishTime: 2019-07-27 11:50:16
 URL: 
http://192.168.1.1:8040/api/_load_error_log?file=__shard_4/error_
log_insert_stmt_4bb00753932c491aa6da6e2725415317_4bb00753932c491a_a6da6e2725415317
 JobDetails: {"Unfinished backends":{"9c3441027ff948a0-
8287923329a2b6a7":[10002]},"ScannedRows":2390016,"TaskNumber":1,"
All backends":{"9c3441027ff948a0-
8287923329a2b6a7":[10002]},"FileNumber":1,"FileSize":1073741824}

下面主要介绍了查看导入命令返回结果集中参数意义：
⚫ JobId
导入任务的唯一 ID，每个导入任务的 JobId 都不同，由系统自动生成。与 Label 不同的是，JobId 永远不会相同，而 Label 则可以在导入任务失败后被复用。
⚫ Label
导入任务的标识。
⚫ State
导入任务当前所处的阶段。在 Broker load 导入过程中主要会出现 PENDING 和
LOADING 这两个导入中的状态。如果 Broker load 处于 PENDING 状态，则说明当前导入任务正在等待被执行；LOADING 状态则表示正在执行中。
导入任务的最终阶段有两个：CANCELLED 和 FINISHED，当 Load job 处于这两个阶
段时，导入完成。其中 CANCELLED 为导入失败，FINISHED 为导入成功。
⚫ Progress
导入任务的进度描述。分为两种进度：ETL 和 LOAD，对应了导入流程的两个阶段 ETL 和 LOADING。目前 Broker load 由于只有 LOADING 阶段，所以 ETL 则会永远显示为 N/A

LOAD 的进度范围为：0~100%。

LOAD 进度 = 当前完成导入的表个数 / 本次导入任务设计的总表个数 * 100%

如果所有导入表均完成导入，此时 LOAD 的进度为 99% 导入进入到最后生效阶段，整个导入完成后，LOAD 的进度才会改为 100%。

导入进度并不是线性的。所以如果一段时间内进度没有变化，并不代表导入没有在执行。
⚫ Type
导入任务的类型。Broker load 的 type 取值只有 BROKER。
⚫ EtlInfo
主要显示了导入的数据量指标 unselected.rows , dpp.norm.ALL 和 dpp.abnorm.ALL。用户可以根据第一个数值判断 where 条件过滤了多少行，后两个指标验证当前导入任务的错误率是否超过 max_filter_ratio。三个指标之和就是原始数据量的总行数。
⚫ TaskInfo
主要显示了当前导入任务参数，也就是创建 Broker load 导入任务时用户指定的导入任务参数，包括：cluster，timeout 和 max_filter_ratio。
⚫ ErrorMsg
在导入任务状态为 CANCELLED，会显示失败的原因，显示分两部分：type 和 msg，如果导入任务成功则显示 N/A。
type 的取值意义：
USER_CANCEL: 用户取消的任务
ETL_RUN_FAIL：在 ETL 阶段失败的导入任务
ETL_QUALITY_UNSATISFIED ： 数 据 质 量 不 合 格 ， 也 就 是 错 误 数 据 率 超 过 了
max_filter_ratio
LOAD_RUN_FAIL：在 LOADING 阶段失败的导入任务
TIMEOUT：导入任务没在超时时间内完成
UNKNOWN：未知的导入错误
CreateTime/EtlStartTime/EtlFinishTime/LoadStartTime/LoadFinishTime
这几个值分别代表导入创建的时间，ETL 阶段开始的时间，ETL 阶段完成的时间，
Loading 阶段开始的时间和整个导入任务完成的时间。
Broker load 导 入 由 于 没 有 ETL 阶 段 ， 所 以 其 EtlStartTime, EtlFinishTime,
LoadStartTime 被设置为同一个值。
导入任务长时间停留在 CreateTime，而 LoadStartTime 为 N/A 则说明目前导入任务堆
积严重。用户可减少导入提交的频率。
LoadFinishTime - CreateTime = 整个导入任务所消耗时间
LoadFinishTime - LoadStartTime = 整个 Broker load 导入任务执行时间 = 整个导入任
务所消耗时间 - 导入任务等待的时间
⚫ URL
导入任务的错误数据样例，访问 URL 地址既可获取本次导入的错误数据样例。当本次导入不存在错误数据时，URL 字段则为 N/A。
⚫ JobDetails
显示一些作业的详细运行状态。包括导入文件的个数、总大小（字节）、子任务个数、已处理的原始行数，运行子任务的 BE 节点 Id，未完成的 BE 节点 Id。
{“Unfinished backends”:{“9c3441027ff948a0-
8287923329a2b6a7”:[10002]},“ScannedRows”:2390016,“TaskNumber”:1,“All
backends”:{“9c3441027ff948a0-
8287923329a2b6a7”:[10002]},“FileNumber”:1,“FileSize”:1073741824}
其中已处理的原始行数，每 5 秒更新一次。该行数仅用于展示当前的进度，不代表最终实际的处理行数。实际处理行数以 EtlInfo 中显示的为准。

1.1.1.6 取消导入

当 Broker load 作业状态不为 CANCELLED 或 FINISHED 时，可以被用户手动取消。取消时需要指定待取消导入任务的 Label 。取消导入命令语法可执行 HELP CANCEL LOAD 查看。

CANCEL LOAD
[FROM db_name]
WHERE LABEL=”load_label”;

1.1.2 Stream Load

Stream load 是一个同步的导入方式，用户通过发送 HTTP 协议发送请求将本地文件或数据流导入到 Doris 中。Stream load 同步执行导入并返回导入结果。用户可直接通过请求的返回体判断本次导入是否成功。

1.1.2.1 适用场景

Stream load 主要适用于导入本地文件，或通过程序导入数据流中的数据。目前 Stream Load 支持两个数据格式：CSV（文本）和 JSON。

1.1.2.2 基本原理

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

Stream load 中，Doris 会选定一个节点作为 Coordinator 节点。该节点负责接数据并分发数据到其他数据节点。

用户通过 HTTP 协议提交导入命令。如果提交到 FE，则 FE 会通过 HTTP redirect 指令将请求转发给某一个 BE。用户也可以直接提交导入命令给某一指定 BE。
导入的最终结果由 Coordinator BE 返回给用户。

1.1.2.3 基本语法

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”
创建导入的详细语法帮助执行 HELP STREAM LOAD 查看, 下面主要介绍创建 Stream load 的部分参数意义。
1）签名参数
user/passwd
Stream load 由于创建导入的协议使用的是 HTTP 协议，通过 Basic access authentication
进行签名。Doris 系统会根据签名验证用户身份和导入权限。
2）导入任务参数
Stream load 由于使用的是 HTTP 协议，所以所有导入任务有关的参数均设置在
Header 中。下面主要介绍了 Stream load 导入任务参数的部分参数意义。
⚫ Label
导入任务的标识
⚫ column_separator
用于指定导入文件中的列分隔符，默认为\t。如果是不可见字符，则需要加\x 作为前缀，使用十六进制来表示分隔符。
如 hive 文件的分隔符\x01，需要指定为-H “column_separator:\x01”。
可以使用多个字符的组合作为列分隔符。
⚫ line_delimiter
用于指定导入文件中的换行符，默认为\n。
可以使用做多个字符的组合作为换行符。
⚫ max_filter_ratio
导入任务的最大容忍率
⚫ where

导入任务指定的过滤条件。Stream load 支持对原始数据指定 where 语句进行过滤。被过滤的数据将不会被导入，也不会参与 filter ratio 的计算，但会被计入num_rows_unselected。
⚫ partition
待导入表的 Partition 信息，如果待导入数据不属于指定的 Partition 则不会被导入。这些数据将计入 dpp.abnorm.ALL
⚫ columns
待导入数据的函数变换配置，目前 Stream load 支持的函数变换方法包含列的顺序变化以及表达式变换，其中表达式变换的方法与查询语句的一致。
列顺序变换例子：原始数据有三列(src_c1,src_c2,src_c3), 目前 doris 表也有三列
（dst_c1,dst_c2,dst_c3）
如果原始表的 src_c1 列对应目标表 dst_c1 列，原始表的 src_c2 列对应目标表 dst_c2 列，原始表的 src_c3 列对应目标表 dst_c3 列，则写法如下：
columns: dst_c1, dst_c2, dst_c3
如果原始表的 src_c1 列对应目标表 dst_c2 列，原始表的 src_c2 列对应目标表 dst_c3 列，原始表的 src_c3 列对应目标表 dst_c1 列，则写法如下：
columns: dst_c2, dst_c3, dst_c1
表达式变换例子：原始文件有两列，目标表也有两列（c1,c2）但是原始文件的两列均需要经过函数变换才能对应目标表的两列，则写法如下：
columns: tmp_c1, tmp_c2, c1 = year(tmp_c1), c2 = month(tmp_c2)
其中 tmp_*是一个占位符，代表的是原始文件中的两个原始列。
⚫ exec_mem_limit
导入内存限制。默认为 2GB，单位为字节。
⚫ strict_mode
⚫ two_phase_commit
Stream load 导入可以开启两阶段事务提交模式。开启方式为在 HEADER 中声明
two_phase_commit=true 。默认的两阶段批量事务提交为关闭。 两阶段批量事务提交模式的意思是：Stream load 过程中，数据写入完成即会返回信息给用户，此时数据不可见，事务状态为 PRECOMMITTED，用户手动触发 commit 操作之后，数据才可见。
用户可以调用如下接口对 stream load 事务触发 commit 操作：

curl -X PUT --location-trusted -u user:passwd -H "txn_id:txnId" -H 
"txn_operation:commit" 
http://fe_host:http_port/api/{db}/_stream_load_2pc
或
curl -X PUT --location-trusted -u user:passwd -H "txn_id:txnId" -H 
"txn_operation:commit" 
http://be_host:webserver_port/api/{db}/_stream_load_2pc

用户可以调用如下接口对 stream load 事务触发 abort 操作：

curl -X PUT --location-trusted -u user:passwd -H "txn_id:txnId" -H 
"txn_operation:abort" 
http://fe_host:http_port/api/{db}/_stream_load_2pc
或
curl -X PUT --location-trusted -u user:passwd -H "txn_id:txnId" -H 
"txn_operation:abort" 
http://be_host:webserver_port/api/{db}/_stream_load_2pc

1.1.2.4 导入示例

curl --location-trusted -u root -H "label:123" -
H"column_separator:," -T student.csv -X PUT 
http://hadoop1:8030/api/test_db/student_result/_stream_load

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

1.1.2.5 取消导入

用户无法手动取消 Stream load，Stream load 在超时或者导入错误后会被系统自动取消。
Stream Load 是一个同步的导入方式，用户通过发送 HTTP 协议将本地文件或数据流导入到Doris 中，Stream load 同步执行导入并返回结果。用户可以直接通过返回判断导入是否成功。

1.1.3 Routine Load

例行导入（Routine Load）功能为用户提供了一种自动从指定数据源进行数据导入的功能。

1.1.3.1 适用场景

当前仅支持从 Kafka 系统进行例行导入，使用限制：
（1）支持无认证的 Kafka 访问，以及通过 SSL 方式认证的 Kafka 集群。
（2）支持的消息格式为 csv, json 文本格式。csv 每一个 message 为一行，且行尾不包含换行符。
（3）默认支持 Kafka 0.10.0.0（含）以上版本。如果要使用 Kafka 0.10.0.0 以下版本(0.9.0, 0.8.2, 0.8.1, 0.8.0)，需要修改 be 的配置，将 kafka_broker_version_fallback 的值设置为要兼容的旧版本，或者在创建 routine load 的时候直接设置 property.broker.version.fallback的值为要兼容的旧版本，使用旧版本的代价是 routine load 的部分新特性可能无法使用，如根据时间设置 kafka 分区的 offset。

1.1.3.2 基本原理

如上图，Client 向 FE 提交一个例行导入作业。
（1）FE 通过 JobScheduler 将一个导入作业拆分成若干个 Task。每个 Task 负责导入指定的一部分数据。Task 被 TaskScheduler 分配到指定的 BE 上执行。
（2）在 BE 上，一个 Task 被视为一个普通的导入任务，通过 Stream Load 的导入机制进行导入。导入完成后，向 FE 汇报。
（3）FE 中的 JobScheduler 根据汇报结果，继续生成后续新的 Task，或者对失败的Task 进行重试。
（4）整个例行导入作业通过不断的产生新的 Task，来完成数据不间断的导入。

1.1.3.3 基本语法

CREATE ROUTINE LOAD [db.]job_name ON tbl_name
[merge_type]
[load_properties]
[job_properties]
FROM data_source
[data_source_properties]

执行 HELP ROUTINE LOAD 可以查看语法帮助，下面是参数说明
1）[db.]job_name
导入作业的名称，在同一个 database 内，相同名称只能有一个 job 在运行。
2）tbl_name
指定需要导入的表的名称。
3）merge_type
数据的合并类型，一共支持三种类型 APPEND、DELETE、MERGE 其中，APPEND 是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据 key 相同的所有行，MERGE 语义 需要与 delete on 条件联合使用，表示满足 delete 条件的数据按照 DELETE 语义处理其余的按照 APPEND 语义处理 , 语法为 [WITH MERGE|APPEND|DELETE]
4）load_properties
用于描述导入数据。语法： [column_separator], [columns_mapping], [where_predicates], [delete_on_predicates], [source_sequence], [partitions], [preceding_predicates]
（1）column_separator:
指定列分隔符，如： COLUMNS TERMINATED BY “,”
这个只在文本数据导入的时候需要指定，JSON 格式的数据导入不需要指定这个参数。默认为：\t
（2）columns_mapping:
指定源数据中列的映射关系，以及定义衍生列的生成方式。
⚫ 映射列：
按顺序指定，源数据中各个列，对应目的表中的哪些列。对于希望跳过的列，可以指定一个不存在的列名。假设目的表有三列 k1, k2, v1。源数据有 4 列，其中第 1、2、4 列分别
对应 k2, k1, v1。则书写如下：
COLUMNS (k2, k1, xxx, v1)
其中 xxx 为不存在的一列，用于跳过源数据中的第三列。
⚫ 衍生列：
以 col_name = expr 的形式表示的列，我们称为衍生列。即支持通过 expr 计算得出目的表中对应列的值。 衍生列通常排列在映射列之后，虽然这不是强制的规定，但是 Doris 总是先解析映射列，再解析衍生列。 接上一个示例，假设目的表还有第 4 列 v2，v2 由 k1 和 k2 的和产生。则可以书写如下：
COLUMNS (k2, k1, xxx, v1, v2 = k1 + k2);
再举例，假设用户需要导入只包含 k1 一列的表，列类型为 int。并且需要将源文件中的对应列进行处理：将负数转换为正数，而将正数乘以 100。这个功能可以通过 case when 函数实现，正确写法应如下：
COLUMNS (xx, k1 = case when xx < 0 then cast(-xx as varchar) else cast((xx + ‘100’) as varchar) end)
（3）where_predicates
用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。 例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：
WHERE k1 > 100 and k2 = 1000
（4）partitions
指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。
示例：
PARTITION(p1, p2, p3)
（5）delete_on_predicates
表示删除条件，仅在 merge type 为 MERGE 时有意义，语法与 where 相同
（6）source_sequence:
只适用于 UNIQUE_KEYS,相同 key 列下，保证 value 列按照 source_sequence 列进行REPLACE, source_sequence 可以是数据源中的列，也可以是表结构中的一列。
（7）preceding_predicates
PRECEDING FILTER predicate
用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。

5）job_properties
用于指定例行导入作业的通用参数。 语法：

PROPERTIES (
 "key1" = "val1",
 "key2" = "val2"
)

目前支持以下参数：
（1）desired_concurrent_number
期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于 0。默认为 3。 这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。
一个作业，最多有多少 task 同时在执行。对于 Kafka 导入而言，当前的实际并发度计算如下：
Min(partition num, desired_concurrent_number, alive_backend_num,
Config.max_routine_load_task_concurrrent_num)
其中 Config.max_routine_load_task_concurrrent_num 是系统的一个默认的最大并发数限制。这是一个 FE 配置，可以通过改配置调整。默认为 5。
其中 partition num 指订阅的 Kafka topic 的 partition 数量。alive_backend_num 是当前正常的 BE 节点数。
（2）max_batch_interval/max_batch_rows/max_batch_size
这三个参数分别表示：
① 每个子任务最大执行时间，单位是秒。范围为 5 到 60。默认为 10。
② 每个子任务最多读取的行数。必须大于等于 200000。默认是 200000。
③ 每个子任务最多读取的字节数。单位是字节，范围是 100MB 到 1GB。默认是
100MB。
这三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。 例如：
“max_batch_interval” = “20”,
“max_batch_rows” = “300000”,
“max_batch_size” = “209715200”
（3）max_error_number
采样窗口内，允许的最大错误行数。必须大于等于 0。默认是 0，即不允许有错误行。

采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，
则会导致例行作业被暂停，需要人工介入检查数据质量问题。 被 where 条件过滤掉的行不算错误行
（4）strict_mode
是否开启严格模式，默认为关闭。如果开启后，非空原始数据的列类型变换如果结果为NULL，则会被过滤。指定方式为 “strict_mode” = “true”
（5）timezone
指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果
（6）format
指定导入数据格式，默认是 csv，支持 json 格式
（7）jsonpaths
jsonpaths: 导入 json 方式分为：简单模式和匹配模式。如果设置了 jsonpath 则为匹配模式导入，否则为简单模式导入，具体可参考示例
（8）strip_outer_array
布尔类型，为 true 表示 json 数据以数组对象开始且将数组对象中进行展平，默认值是false
（9）json_root
json_root 为合法的 jsonpath 字符串，用于指定 json document 的根节点，默认值为""
（10）send_batch_parallelism
整型，用于设置发送批处理数据的并行度，如果并行度的值超过 BE 配置中的
max_send_batch_parallelism_per_job ， 那 么 作 为 协 调 点 的 BE 将 使 用
max_send_batch_parallelism_per_job 的值
6）data_source_properties
数据源的类型。当前支持：Kafka
(
“key1” = “val1”,
“key2” = “val2”
)

1.1.3.4 Kafka 导入示例

1）在 doris 中创建对应表

create table student_kafka
(
id int,
name varchar(50),
age int
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 10;

2）启动 kafka 并准备数据

bin/kafka-topics.sh --create
–zookeeper hadoop1:2181/kafka
–replication-factor 1
–partitions 1
–topic test_doris1 bin/kafka-console-producer.sh
–broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092
–topic test_doris

3）创建导入任务

CREATE ROUTINE LOAD test_db.kafka_test ON student_kafka
COLUMNS TERMINATED BY ",",
COLUMNS(id, name, age)
PROPERTIES
(
"desired_concurrent_number"="3",
"strict_mode" = "false"
)
FROM KAFKA
(
 "kafka_broker_list"= "hadoop1:9092,hadoop2:9092,hadoop3:9092",
 "kafka_topic" = "test_doris1",
 "property.group.id"="test_doris_group",
 "property.kafka_default_offsets" = "OFFSET_BEGINNING",
 "property.enable.auto.commit"="false"
);

4）查看表
select * from student_kafka;
继续往 kafka 发送数据，查看表的变化

1.1.3.5 查看导入作业状态

查看作业状态的具体命令和示例可以通过 HELP SHOW ROUTINE LOAD; 命令查看。
查看任务运行状态的具体命令和示例可以通过 HELP SHOW ROUTINE LOAD TASK;
命令查看。
只能查看当前正在运行中的任务，已结束和未开始的任务无法查看。

1.1.3.6 修改作业属性

用户可以修改已经创建的作业。具体说明可以通过 HELP ALTER ROUTINE LOAD; 命令查看。或参阅 ALTER ROUTINE LOAD。

1.1.3.7 作业控制

用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止，暂停和重启。可以
通过 HELP STOP ROUTINE LOAD; HELP PAUSE ROUTINE LOAD; 以及 HELP RESUME
ROUTINE LOAD; 三个命令查看帮助和示例。

1.1.3.8 其他说明

1）例行导入作业和 ALTER TABLE 操作的关系
例行导入不会阻塞 SCHEMA CHANGE 和 ROLLUP 操作。但是注意如果 SCHEMA CHANGE 完成后，列映射关系无法匹配，则会导致作业的错误数据激增，最终导致作业暂停。建议通过在例行导入作业中显式指定列映射关系，以及通过增加 Nullable 列或带Default 值的列来减少这类问题。
删除表的 Partition 可能会导致导入数据无法找到对应的 Partition，作业进入暂停。
2）例行导入作业和其他导入作业的关系（LOAD, DELETE, INSERT）
例行导入和其他 LOAD 作业以及 INSERT 操作没有冲突。
当执行 DELETE 操作时，对应表分区不能有任何正在执行的导入任务。所以在执行DELETE 操作前，可能需要先暂停例行导入作业，并等待已下发的 task 全部完成后，才可以执行 DELETE。
3）例行导入作业和 DROP DATABASE/TABLE 操作的关系
当例行导入对应的 database 或 table 被删除后，作业会自动 CANCEL。
4）kafka 类型的例行导入作业和 kafka topic 的关系
当用户在创建例行导入声明的 kafka_topic 在 kafka 集群中不存在时：
（1）如果用户 kafka 集群的 broker 设置了 auto.create.topics.enable = true，则
kafka_topic 会先被自动创建，自动创建的 partition 个数是由用户方的 kafka 集群中的broker 配置 num.partitions 决定的。例行作业会正常的不断读取该 topic 的数据。
（2）如果用户 kafka 集群的 broker 设置了 auto.create.topics.enable = false, 则 topic 不会被自动创建，例行作业会在没有读取任何数据之前就被暂停，状态为 PAUSED。所以，如果用户希望当 kafka topic 不存在的时候，被例行作业自动创建的话，只需要将用户方的 kafka 集群中的 broker 设置 auto.create.topics.enable = true 即可。
5）在网络隔离的环境中可能出现的问题
在有些环境中存在网段和域名解析的隔离措施，所以需要注意：
（1）创建 Routine load 任务中指定的 Broker list 必须能够被 Doris 服务访问
（2）Kafka 中如果配置了 advertised.listeners, advertised.listeners 中的地址必须能够被Doris 服务访问
6）关于指定消费的 Partition 和 Offset
Doris 支持指定 Partition 和 Offset 开始消费。新版中还支持了指定时间点进行消费的功能。这里说明下对应参数的配置关系。
有三个相关参数：
kafka_partitions：指定待消费的 partition 列表，如：“0, 1, 2, 3”。
kafka_offsets：指定每个分区的起始 offset，必须和 kafka_partitions 列表个数对应。如：
“1000, 1000, 2000, 2000”
property.kafka_default_offset：指定分区默认的起始 offset。
在创建导入作业时，这三个参数可以有以下组合

组合	kafka_partitions	kafka_offsets	property.kafka_default_offset	行为
1	No	No	No	系统会自动查找 topic 对应的所有分区并从OFFSET_END 开始消费
2	No	No	Yes	系统会自动查找 topic 对应的所有分区并从default offset 指定的位置开始消费
3	Yes	No	No	系统会从指定分区的OFFSET_END 开始消费
4	Yes	Yes	No	系统会从指定分 区 的 指 定offset 处 开 始消费
5	Yes	No	Yes	系统会从指定分区， default offset 指 定 的位置开始消费

7）STOP 和 PAUSE 的区别
FE 会自动定期清理 STOP 状态的 ROUTINE LOAD，而 PAUSE 状态的则可以再次被恢复启用。

1.1.4 Binlog Load

Binlog Load 提供了一种使 Doris 增量同步用户在 Mysql 数据库的对数据更新操作的CDC（Change Data Capture）功能。

1.1.4.1 适用场景

INSERT/UPDATE/DELETE 支持。
过滤 Query。
暂不兼容 DDL 语句。

1.1.4.2 基本原理

在第一期的设计中，Binlog Load 需要依赖 canal 作为中间媒介，让 canal 伪造成一个从节点去获取 Mysql 主节点上的 Binlog 并解析，再由 Doris 去获取 Canal 上解析好的数据，主要涉及 Mysql 端、Canal 端以及 Doris 端，总体数据流向如下：

如上图，用户向 FE 提交一个数据同步作业。
（1）FE 会为每个数据同步作业启动一个 canal client，来向 canal server 端订阅并获取数据。
（2）client 中的 receiver 将负责通过 Get 命令接收数据，每获取到一个数据 batch，都会由 consumer 根据对应表分发到不同的 channel，每个 channel 都会为此数据 batch 产生一个发送数据的子任务 Task。
（3）在 FE 上，一个 Task 是 channel 向 BE 发送数据的子任务，里面包含分发到当前
channel 的同一个 batch 的数据。
（4）channel 控制着单个表事务的开始、提交、终止。一个事务周期内，一般会从 consumer获取到多个 batch 的数据，因此会产生多个向 BE 发送数据的子任务 Task，在提交事务成功前，这些 Task 不会实际生效。
（5）满足一定条件时（比如超过一定时间、达到提交最大数据大小），consumer 将会
阻塞并通知各个 channel 提交事务。
（6）当且仅当所有 channel 都提交成功，才会通过 Ack 命令通知 canal 并继续获取并消
费数据。
（7）如果有任意 channel 提交失败，将会重新从上一次消费成功的位置获取数据并再
次提交（已提交成功的 channel 不会再次提交以保证幂等性）。
（8）整个数据同步作业中，FE 通过以上流程不断的从 canal 获取数据并提交到 BE，来
完成数据同步。

1.1.4.3 配置 MySQL 端

在 MySQLCluster 模式的主从同步中，二进制日志文件（Binlog）记录了主节点上的所
有数据变化，数据在 Cluster 的多个节点间同步、备份都要通过 Binlog 日志进行，从而提高
集群的可用性。架构通常由一个主节点（负责写）和一个或多个从节点（负责读）构成，所
有在主节点上发生的数据变更将会复制给从节点。
注意：目前必须要使用 Mysql 5.7 及以上的版本才能支持 Binlog Load 功能。
1）打开 mysql 的二进制 binlog 日志功能，编辑 my.cnf 配置文件

[mysqld]
log-bin = mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
binlog-do-db=test #指定具体要同步的数据库，也可以不设置

2）开启 GTID 模式 [可选]
一个全局事务 Id(global transaction identifier)标识出了一个曾在主节点上提交过的事务，
在全局都是唯一有效的。开启了 Binlog 后，GTID 会被写入到 Binlog 文件中，与事务一一对应。
编辑 my.cnf 配置文件。

gtid-mode=on // 开启 gtid 模式
enforce-gtid-consistency=1 // 强制 gtid 和事务的一致性

在 GTID 模式下，主服务器可以不需要 Binlog 的文件名和偏移量，就能很方便的追踪
事务、恢复数据、复制副本。
在 GTID 模式下，由于 GTID 的全局有效性，从节点将不再需要通过保存文件名和偏移
量来定位主节点上的 Binlog 位置，而通过数据本身就可以定位了。在进行数据同步中，从节点会跳过执行任意被识别为已执行的 GTID 事务。
GTID 的表现形式为一对坐标, source_id 标识出主节点，transaction_id 表示此事务在主
节点上执行的顺序（最大 263-1）。
3）重启 MySQL 使配置生效

sudo systemctl restart mysqld

4）创建用户并授权

set global validate_password_length=4;
set global validate_password_policy=0;
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 
'canal'@'%' IDENTIFIED BY 'canal' ;

5）准备测试表

CREATE TABLE `test`.`tbl1` (
 `a` int(11) NOT NULL COMMENT "",
 `b` int(11) NOT NULL COMMENT ""
)
insert into test.tbl1 values(1,1),(2,2),(3,3);

1.1.4.4 配置 Canal 端

Canal 是属于阿里巴巴 otter 项目下的一个子项目，主要用途是基于 MySQL 数据库增量
日志解析，提供增量数据订阅和消费，用于解决跨机房同步的业务场景，建议使用 canal 1.1.5及以上版本。
下载地址：https://github.com/alibaba/canal/releases
1）上传并解压 canal deployer

mkdir /opt/module/canal-1.1.5
tar -zxvf canal.deployer-1.1.5.tar.gz -C /opt/module/canal-1.1.5

2）在 conf 文件夹下新建目录并重命名
一个 canal 服务中可以有多个 instance，conf/下的每一个目录即是一个实例，每个实例
下面都有独立的配置文件

mkdir /opt/module/canel-1.1.5/conf/doris-load

拷贝配置文件模板

cp /opt/module/canel-1.1.5/conf/example/instance.properties
/opt/module/canel-1.1.5/conf/doris-load

3）修改 conf/canal.properties 的配置

canal.destinations = doris-load

4）修改 instance 配置文件

vim /opt/module/canel-1.1.5/conf/doris-load/instance.properties
##canal instance serverId
canal.instance.mysql.slaveId = 1234
##mysql address
canal.instance.master.address = hadoop1:3306 
##mysql username/password
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal

4）启动

sh bin/startup.sh

5）验证启动成功

cat logs/doris-load/doris-load.log

注意：canal client 和 canal instance 是一一对应的，Binlog Load 已限制多个数据同步作
业不能连接到同一个 destination。
4.1.4.5 配置目标表
1）Doris 创建与 Mysql 对应的目标表

CREATE TABLE `binlog_test` (
 `a` int(11) NOT NULL COMMENT "",
 `b` int(11) NOT NULL COMMENT ""
) ENGINE=OLAP
UNIQUE KEY(`a`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`a`) BUCKETS 8;

Binlog Load 只能支持 Unique 类型的目标表，且必须激活目标表的 Batch Delete 功能。
2）开启 SYNC 功能
在 fe.conf 中将 enable_create_sync_job 设为 true，不想修改配置文件重启，可以执行如下：

使用 root 账号登陆

ADMIN SET FRONTEND CONFIG (“enable_create_sync_job” = “true”);

1.1.4.6 基本语法

创建数据同步作业的的详细语法可以连接到 Doris 后，执行 HELP CREATE SYNC JOB;
查看语法帮助。

CREATE SYNC [db.]job_name
(
 channel_desc, 
 channel_desc
 ...
)
binlog_desc

⚫ job_name
job_name 是数据同步作业在当前数据库内的唯一标识，相同 job_name 的作业只能有一
个在运行。
⚫ channel_desc
channel_desc 用来定义任务下的数据通道，可表示 MySQL 源表到 doris 目标表的映射关系。在设置此项时，如果存在多个映射关系，必须满足 MySQL 源表应该与 doris 目标表是一一对应关系，其他的任何映射关系（如一对多关系），检查语法时都被视为不合法。

FROM mysql_db.src_tbl INTO des_tbl
[partitions]
[columns_mapping]

column_mapping主要指MySQL源表和doris目标表的列之间的映射关系，如果不指定，
FE 会默认源表和目标表的列按顺序一一对应。但是我们依然建议显式的指定列的映射关系，这样当目标表的结构发生变化（比如增加一个 nullable 的列），数据同步作业依然可以进行。否则，当发生上述变动后，因为列映射关系不再一一对应，导入将报错。
⚫ binlog_desc
binlog_desc 中的属性定义了对接远端 Binlog 地址的一些必要信息，目前可支持的对接
类型只有 canal 方式，所有的配置项前都需要加上 canal 前缀。

FROM BINLOG
(
 "key1" = "value1", 
 "key2" = "value2"
)

canal.server.ip: canal server 的地址
canal.server.port: canal server 的端口
canal.destination: 前文提到的 instance 的字符串标识
canal.batchSize: 每批从 canal server 处获取的 batch 大小的最大值，默认 8192
canal.username: instance 的用户名
canal.password: instance 的密码
canal.debug: 设置为 true 时，会将 batch 和每一行数据的详细信息都打印出来，会影响
性能。

1.1.4.7 示例

1）创建同步作业

CREATE SYNC test_db.job1
(
 FROM test.tbl1 INTO binlog_test
)
FROM BINLOG 
(
 "type" = "canal",
 "canal.server.ip" = "hadoop1",
 "canal.server.port" = "11111",
 "canal.destination" = "doris-load",
 "canal.username" = "canal",
 "canal.password" = "canal"
);

2）查看作业状态
查看作业状态的具体命令和示例可以通过 HELP SHOW SYNC JOB; 命令查看。
展示当前数据库的所有数据同步作业状态。

SHOW SYNC JOB;

展示数据库 test_db 下的所有数据同步作业状态。

SHOW SYNC JOB FROM test_db;

返回结果集的参数意义如下：
⚫ State
作业当前所处的阶段。作业状态之间的转换如下图所示：
作业提交之后状态为PENDING，由FE调度执行启动canal client后状态变成RUNNING，
用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止，暂停和恢复，操作后作业状态分别为 CANCELLED/PAUSED/RUNNING。
作业的最终阶段只有一个 CANCELLED，当作业状态变为 CANCELLED 后，将无法再
次恢复。当作业发生了错误时，若错误是不可恢复的，状态会变成 CANCELLED，否则会变成 PAUSED。
⚫ Channel
作业所有源表到目标表的映射关系。
⚫ Status
当前 binlog 的消费位置(若设置了 GTID 模式，会显示 GTID)，以及 doris 端执行时间相
比 mysql 端的延迟时间。
⚫ JobConfig
对接的远端服务器信息，如 canal server 的地址与连接 instance 的 destination。
3）MySQL 表继续插入数据，观察 Doris 的表
4）控制作业
用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止，暂停和恢复。
停止名称为 job_name 的数据同步作业
STOP SYNC JOB [db.]job_name
暂停名称为 job_name 的数据同步作业
PAUSE SYNC JOB [db.]job_name
恢复名称为 job_name 的数据同步作业
RESUME SYNC JOB job_name

1.1.5 Insert Into

Insert Into 语句的使用方式和 MySQL 等数据库中 Insert Into 语句的使用方式类似。但
在 Doris 中，所有的数据写入都是一个独立的导入作业。所以这里将 Insert Into 也作为一
种导入方式介绍。
主要的 Insert Into 命令包含以下两种；

INSERT INTO tbl SELECT ...
INSERT INTO tbl (col1, col2, ...) VALUES (1, 2, ...), (1,3, ...);

其中第二种命令仅用于 Demo，不要使用在测试或生产环境中。
Insert Into 命令需要通过 MySQL 协议提交，创建导入请求会同步返回导入结果。

1.1.5.1 语法

INSERT INTO table_name [partition_info] [WITH LABEL label] 
[col_list] [query_stmt] [VALUES];

WITH LABEL：
INSERT 操作作为一个导入任务，也可以指定一个 label。如果不指定，则系统会自动
指定一个 UUID 作为 label。
该功能需要 0.11+ 版本。
注意：建议指定 Label 而不是由系统自动分配。如果由系统自动分配，但在 Insert Into
语句执行过程中，因网络错误导致连接断开等，则无法得知 Insert Into 是否成功。而如果指定 Label，则可以再次通过 Label 查看任务结果。
示例：

INSERT INTO tbl2 WITH LABEL label1 SELECT * FROM tbl3;
INSERT INTO tbl1 VALUES ("qweasdzxcqweasdzxc"), ("a");

注意：
当需要使用 CTE(Common Table Expressions) 作为 insert 操作中的查询部分时，必须指
定 WITH LABEL 和 column list 部分。示例

INSERT INTO tbl1 WITH LABEL label1
WITH cte1 AS (SELECT * FROM tbl1), cte2 AS (SELECT * FROM tbl2)
SELECT k1 FROM cte1 JOIN cte2 WHERE cte1.k1 = 1;
INSERT INTO tbl1 (k1)
WITH cte1 AS (SELECT * FROM tbl1), cte2 AS (SELECT * FROM tbl2)
SELECT k1 FROM cte1 JOIN cte2 WHERE cte1.k1 = 1;

1.1.5.2 SHOW LAST INSERT

一些语言的 MySQL 类库中很难获取返回结果的中的 json 字符串。因此，Doris 还提供
了 SHOW LAST INSERT 命令来显式的获取最近一次 insert 操作的结果。
当执行完一个 insert 操作后，可以在同一 session 连接中执行 SHOW LAST INSERT。该命令会返回最近一次 insert 操作的结果，如：

mysql> show last insert\G
*************************** 1. row ***************************
 TransactionId: 64067
 Label: insert_ba8f33aea9544866-8ed77e2844d0cc9b
 Database: default_cluster:db1
 Table: t1
TransactionStatus: VISIBLE
 LoadedRows: 2
 FilteredRows: 0

该命令会返回 insert 以及对应事务的详细信息。因此，用户可以在每次执行完 insert 操
作后，继续执行 show last insert 命令来获取 insert 的结果。
注意：该命令只会返回在同一 session 连接中，最近一次 insert 操作的结果。如果连接断
开或更换了新的连接，则将返回空集。

1.1.6 S3 Load

参考官网：https://doris.apache.org/zh-CN/administrator-guide/load-data/s3-load-manual.html

2.2 数据导出

2.2.1 Export 导出

数据导出是 Doris 提供的一种将数据导出的功能。该功能可以将用户指定的表或分区的
数据以文本的格式，通过 Broker 进程导出到远端存储上，如 HDFS/BOS 等。

2.2.1.1 基本原理

用户提交一个 Export 作业后。Doris 会统计这个作业涉及的所有 Tablet。然后对这些
Tablet 进行分组，每组生成一个特殊的查询计划。该查询计划会读取所包含的 Tablet 上的数据，然后通过 Broker 将数据写到远端存储指定的路径中，也可以通过 S3 协议直接导出到支持 S3 协议的远端存储上。
1）调度方式

（1）用户提交一个 Export 作业到 FE。
（2）FE 的 Export 调度器会通过两阶段来执行一个 Export 作业：
① PENDING：FE 生成 ExportPendingTask，向 BE 发送 snapshot 命令，对所有
涉及到的 Tablet 做一个快照。并生成多个查询计划。
② EXPORTING：FE 生成 ExportExportingTask，开始执行查询计划。
2）查询计划拆分
Export 作业会生成多个查询计划，每个查询计划负责扫描一部分 Tablet。每个查询计划
扫描的 Tablet 个数由 FE 配置参数 export_tablet_num_per_task 指定，默认为 5。即假设一共 100 个 Tablet，则会生成 20 个查询计划。用户也可以在提交作业时，通过作业属性tablet_num_per_task 指定这个数值。
3）查询计划执行
一个作业的多个查询计划顺序执行。
一个查询计划扫描多个分片，将读取的数据以行的形式组织，每 1024 行为一个 batch，
调用 Broker 写入到远端存储上。
查询计划遇到错误会整体自动重试 3 次。如果一个查询计划重试 3 次依然失败，则整
个作业失败。
Doris 会首先在指定的远端存储的路径中，建立一个名为 __doris_export_tmp_12345 的
临时目录（其中 12345 为作业 id）。导出的数据首先会写入这个临时目录。每个查询计划会生成一个文件，文件名示例：©
export-data-c69fcf2b6db5420f-a96b94c1ff8bccef-1561453713822
其中 c69fcf2b6db5420f-a96b94c1ff8bccef 为查询计划的 query id。1561453713822 为文件生成的时间戳。
当所有数据都导出后，Doris 会将这些文件 rename 到用户指定的路径中。

2.2.1.2 基本语法

Export 的详细命令可以通过 HELP EXPORT 查看：

EXPORT TABLE db1.tbl1 
PARTITION (p1,p2)
[WHERE [expr]]
TO "hdfs://host/path/to/export/" 
PROPERTIES
(
"label" = "mylabel",
 "column_separator"=",",
 "columns" = "col1,col2",
 "exec_mem_limit"="2147483648",
 "timeout" = "3600"
)
WITH BROKER "hdfs"
(
"username" = "user",
"password" = "passwd"
);

➢ label：本次导出作业的标识。后续可以使用这个标识查看作业状态。
➢ column_separator：列分隔符。默认为 \t。支持不可见字符，比如 ‘\x07’。
➢ columns：要导出的列，使用英文状态逗号隔开，如果不填这个参数默认是导出表
的所有列。
➢ line_delimiter：行分隔符。默认为 \n。支持不可见字符，比如 ‘\x07’。
➢ exec_mem_limit： 表示 Export 作业中，一个查询计划在单个 BE 上的内存使用限
制。默认 2GB。单位字节。
➢ timeout：作业超时时间。默认 2 小时。单位秒。
➢ tablet_num_per_task：每个查询计划分配的最大分片数。默认为 5。

2.2.1.3 导出示例

1）启动 hadoop 集群
2）执行导出

export table example_site_visit2
to "hdfs://mycluster/doris-export"
PROPERTIES
(
"label" = "mylabel",
 "column_separator"="|",
 "timeout" = "3600"
)
WITH BROKER "broker_name"
(
 #HDFS 开启 HA 需要指定，还指定其他参数
 "dfs.nameservices"="mycluster",
 "dfs.ha.namenodes.mycluster"="nn1,nn2,nn3",
 "dfs.namenode.rpc-address.mycluster.nn1"= "hadoop1:8020",
 "dfs.namenode.rpc-address.mycluster.nn2"= "hadoop2:8020",
 "dfs.namenode.rpc-address.mycluster.nn3"="hadoop3:8020",
"dfs.client.failover.proxy.provider.mycluster"="org.apache.hadoop
.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider" 
);

3）导出之后查看 hdfs 对应路径，会多出许多文件

2.2.1.4 查询导出作业状态

提交作业后，可以通过 SHOW EXPORT 命令查询导出作业状态。结果举例如下：

JobId: 14008
 Label: mylabel
 State: FINISHED
 Progress: 100%
 TaskInfo: {"partitions":["*"],"exec mem 
limit":2147483648,"column separator":",","line 
delimiter":"\n","tablet num":1,"broker":"hdfs","coord 
num":1,"db":"default_cluster:db1","tbl":"tbl3"}
 Path: bos://bj-test-cmy/export/
CreateTime: 2019-06-25 17:08:24
StartTime: 2019-06-25 17:08:28
FinishTime: 2019-06-25 17:08:34
 Timeout: 3600
 ErrorMsg: N/A

➢ JobId：作业的唯一 ID
➢ Label：自定义作业标识
➢ State：作业状态：
PENDING：作业待调度
EXPORTING：数据导出中
FINISHED：作业成功
CANCELLED：作业失败
➢ Progress：作业进度。该进度以查询计划为单位。假设一共 10 个查询计划，当前
已完成 3 个，则进度为 30%。
➢ TaskInfo：以 Json 格式展示的作业信息：
db：数据库名
tbl：表名
partitions：指定导出的分区。* 表示所有分区。
exec mem limit：查询计划内存使用限制。单位字节。
column separator：导出文件的列分隔符。
line delimiter：导出文件的行分隔符。
tablet num：涉及的总 Tablet 数量。
broker：使用的 broker 的名称。
coord num：查询计划的个数。
➢ Path：远端存储上的导出路径。
➢ CreateTime/StartTime/FinishTime：作业的创建时间、开始调度时间和结束时间。
➢ Timeout：作业超时时间。单位是秒。该时间从 CreateTime 开始计算。
➢ ErrorMsg：如果作业出现错误，这里会显示错误原因

2.2.1.5 注意事项

（1）不建议一次性导出大量数据。一个 Export 作业建议的导出数据量最大在几十 GB。
过大的导出会导致更多的垃圾文件和更高的重试成本。
（2）如果表数据量过大，建议按照分区导出。
（3）在 Export 作业运行过程中，如果 FE 发生重启或切主，则 Export 作业会失败，
需要用户重新提交。
（4）如果 Export 作业运行失败，在远端存储中产生的 __doris_export_tmp_xxx 临时
目录，以及已经生成的文件不会被删除，需要用户手动删除。
（5）如果 Export 作业运行成功，在远端存储中产生的 __doris_export_tmp_xxx 目录，
根据远端存储的文件系统语义，可能会保留，也可能会被清除。比如在百度对象存储（BOS）中，通过 rename 操作将一个目录中的最后一个文件移走后，该目录也会被删除。如果该目录没有被清除，用户可以手动清除。
（6）当 Export 运行完成后（成功或失败），FE 发生重启或切主，则 SHOW EXPORT
展示的作业的部分信息会丢失，无法查看。
（7）Export 作业只会导出 Base 表的数据，不会导出 Rollup Index 的数据。
（8）Export 作业会扫描数据，占用 IO 资源，可能会影响系统的查询延迟

2.2.2 查询结果导出

SELECT INTO OUTFILE 语句可以将查询结果导出到文件中。目前支持通过 Broker 进
程, 通过 S3 协议, 或直接通过 HDFS 协议，导出到远端存储，如 HDFS，S3，BOS，COS（腾讯云）上。

2.2.2.1 语法

语法如下

query_stmt
INTO OUTFILE "file_path"
[format_as]
[properties]

⚫ file_path
file_path 指向文件存储的路径以及文件前缀。如 hdfs://path/to/my_file_。
最终的文件名将由 my_file_，文件序号以及文件格式后缀组成。其中文件序号由 0 开
始，数量为文件被分割的数量。如：
my_file_abcdefg_0.csv
my_file_abcdefg_1.csv
my_file_abcdegf_2.csv
⚫ [format_as]
FORMAT AS CSV
指定导出格式。默认为 CSV。
⚫ [properties]
指定相关属性。目前支持通过 Broker 进程, 或通过 S3 协议进行导出。
Broker 相关属性需加前缀 broker.。具体参阅 Broker 文档。
HDFS 相关属性需加前缀 hdfs. 其中 hdfs.fs.defaultFS 用于填写 namenode 地址和端
口。属于必填项。
S3 协议则直接执行 S3 协议配置即可。
示例：

("broker.prop_key" = "broker.prop_val", ...)
or
("hdfs.fs.defaultFS" = "xxx", "hdfs.hdfs_user" = "xxx")
or 
("AWS_ENDPOINT" = "xxx", ...)

其他属性：
(“key1” = “val1”, “key2” = “val2”, …)
目前支持以下属性：
➢ column_separator：列分隔符，仅对 CSV 格式适用。默认为 \t。
➢ line_delimiter：行分隔符，仅对 CSV 格式适用。默认为 \n。
➢ max_file_size：单个文件的最大大小。默认为 1GB。取值范围在 5MB 到 2GB 之
间。超过这个大小的文件将会被切分。
➢ schema：PARQUET 文件 schema 信息。仅对 PARQUET 格式适用。导出文件格式
为 PARQUET 时，必须指定 schema。

2.2.2.2 并发导出

1）并发导出的条件
默认情况下，查询结果集的导出是非并发的，也就是单点导出。如果用户希望查询结果
集可以并发导出，需要满足以下条件：
（1）session variable ‘enable_parallel_outfile’ 开启并发导出：

set enable_parallel_outfile = true;

（2）导出方式为 S3 , 或者 HDFS，而不是使用 broker
（3）查询可以满足并发导出的需求，比如顶层不包含 sort 等单点节点。（后面会举例
说明，哪种属于不可并发导出结果集的查询）
满足以上三个条件，就能触发并发导出查询结果集了。
并发度 = be_instacne_num * parallel_fragment_exec_instance_num
2）验证结果集被并发导出。
用户通过 session 变量设置开启并发导出后，如果想验证当前查询是否能进行并发导出，
则可以通过下面这个方法。
explain select xxx from xxx where xxx into outfile “s3://xxx”
format as csv properties (“AWS_ENDPOINT” = “xxx”, …);
对查询进行 explain 后，Doris 会返回该查询的规划，如果发现 RESULT FILE SINK 出现
在 PLAN FRAGMENT 1 中，就说明导出并发开启成功了。如果 RESULT FILE SINK 出现在PLAN FRAGMENT 0 中，则说明当前查询不能进行并发导出（当前查询不同时满足并发导出的三个条件）。

2.2.2.3 使用示例

示例一：使用 broker 方式，将简单查询结果导出

SELECT * FROM example_site_visit
INTO OUTFILE "hdfs://hadoop1:8020/doris-out/broker_a_"
FORMAT AS CSV
PROPERTIES
(
 "broker.name" = "broker_name",
 "column_separator" = ",",
 "line_delimiter" = "\n",
 "max_file_size" = "100MB"
);

最终生成文件如如果不大于 100MB，则为：result_0.csv。
如果大于 100MB，则可能为 result_0.csv, result_1.csv, ...。
示例二：使用 broker 方式，指定导出格式为 PARQUET

SELECT city, age FROM example_site_visit
INTO OUTFILE "hdfs://hadoop1:8020/doris-out/broker_b_"
FORMAT AS PARQUET
PROPERTIES
(
"broker.name" = "broker_name",
"schema"="required,byte_array,city;required,int32,age"
);

查询结果导出到 parquet 文件需要明确指定schema。
示例三：使用 HDFS 方式导出

SELECT * FROM example_site_visit
INTO OUTFILE "hdfs://doris-out/hdfs_"
FORMAT AS CSV
PROPERTIES
(
 "hdfs.fs.defaultFS" = "hdfs://hadoop1:8020",
"hdfs.hdfs_user" = "atguigu",
"column_separator" = ","
);

最终生成文件如如果不大于 100MB，则为：result_0.csv。
如果大于 100MB，则可能为 result_0.csv, result_1.csv, ...。
示例四：使用 HDFS 方式导出，开启并发导出

set enable_parallel_outfile = true;
EXPLAIN SELECT * FROM example_site_visit
INTO OUTFILE "hdfs://doris-out/hdfs_"
FORMAT AS CSV
PROPERTIES
(
 "hdfs.fs.defaultFS" = "hdfs://hadoop1:8020",
"hdfs.hdfs_user" = "atguigu",
"column_separator" = ","
);

其他用法：
（1） 将 CTE 语句的查询结果导出到文件 hdfs://path/to/result.txt。默认导出格式为
CSV。使用my_broker并设置 HDFS 高可用信息。使用默认的行列分隔符。

WITH
x1 AS
(SELECT k1, k2 FROM tbl1),
x2 AS
(SELECT k3 FROM tbl2)
SELEC k1 FROM x1 UNION SELECT k3 FROM x2
INTO OUTFILE "hdfs://path/to/result_"
PROPERTIES
(
 "broker.name" = "my_broker",
 "broker.username"="user",
 "broker.password"="passwd",
 "broker.dfs.nameservices" = "my_ha",
 "broker.dfs.ha.namenodes.my_ha" = "my_namenode1, my_namenode2",
 "broker.dfs.namenode.rpc-address.my_ha.my_namenode1" = 
"nn1_host:rpc_port",
 "broker.dfs.namenode.rpc-address.my_ha.my_namenode2" = 
"nn2_host:rpc_port",
 "broker.dfs.client.failover.proxy.provider" = 
"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProx
yProvider"
);

最终生成文件如如果不大于 1GB，则为：result_0.csv。
如果大于 1GB，则可能为 result_0.csv, result_1.csv, ...。
（2）将 UNION 语句的查询结果导出到文件bos://bucket/result.txt。指定导出格式为
PARQUET。使用my_broker并设置 HDFS 高可用信息。PARQUET 格式无需指定列分割符。
导出完成后，生成一个标识文件。

SELECT k1 FROM tbl1 UNION SELECT k2 FROM tbl1
INTO OUTFILE "bos://bucket/result_"
FORMAT AS PARQUET
PROPERTIES
(
 "broker.name" = "my_broker",
 "broker.bos_endpoint" = "http://bj.bcebos.com",
 "broker.bos_accesskey" = "xxxxxxxxxxxxxxxxxxxxxxxxxx",
 "broker.bos_secret_accesskey" = "yyyyyyyyyyyyyyyyyyyyyyyyyy",
 "schema"="required,int32,k1;required,byte_array,k2"
);

（3）将 select 语句的查询结果导出到文件 cos://${bucket_name}/path/result.txt。指定
导出格式为 csv。
导出完成后，生成一个标识文件。

select k1,k2,v1 from tbl1 limit 100000
into outfile "s3a://my_bucket/export/my_file_"
FORMAT AS CSV
PROPERTIES
(
 "broker.name" = "hdfs_broker",
 "broker.fs.s3a.access.key" = "xxx",
 "broker.fs.s3a.secret.key" = "xxxx",
 "broker.fs.s3a.endpoint" = "https://cos.xxxxxx.myqcloud.com/",
 "column_separator" = ",",
 "line_delimiter" = "\n",
 "max_file_size" = "1024MB",
 "success_file_name" = "SUCCESS"
)

最终生成文件如如果不大于 1GB，则为：my_file_0.csv。
如果大于 1GB，则可能为 my_file_0.csv, result_1.csv, ...。
在 cos 上验证
① 不存在的 path 会自动创建
② access.key/secret.key/endpoint 需要和 cos 的同学确认。尤其是 endpoint 的值，不需要填写 bucket_name。
（4）使用 s3 协议导出到 bos，并且并发导出开启。

set enable_parallel_outfile = true;
select k1 from tb1 limit 1000
into outfile "s3://my_bucket/export/my_file_"
format as csv
properties
(
 "AWS_ENDPOINT" = "http://s3.bd.bcebos.com",
 "AWS_ACCESS_KEY" = "xxxx",
 "AWS_SECRET_KEY" = "xxx",
 "AWS_REGION" = "bd"
)

最终生成的文件前缀为 my_file_{fragment_instance_id}_。
（5）使用 s3 协议导出到 bos，并且并发导出 session 变量开启。
注意：但由于查询语句带了一个顶层的排序节点，所以这个查询即使开启并发导出的
session 变量，也是无法并发导出的。

set enable_parallel_outfile = true;
select k1 from tb1 order by k1 limit 1000
into outfile "s3://my_bucket/export/my_file_"
format as csv
properties
(
 "AWS_ENDPOINT" = "http://s3.bd.bcebos.com",
 "AWS_ACCESS_KEY" = "xxxx",
 "AWS_SECRET_KEY" = "xxx",
 "AWS_REGION" = "bd"
)