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 Broker Load

1.1.1 适用场景

源数据在Broker可以访问的存储系统中，如HDFS。

数据量在几十到百GB级别。

1.1.2 基本原理

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

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

 

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 函数，目标表及分区等信息。下面主要对数据描述类的部分参数详细解释：
3. 多表导入Broker load 支持一次导入任务涉及多张表，每个 Broker load 导入任务可在多个 data_desc 声明多张表来实现多表导入。每个单独的 data_desc 还可以指定属于该表的数据源地址。Broker load 保证了单次导入的多张表之间原子性成功或失败。
4. negativedata_desc中还可以设置数据取反导入。这个功能主要用于，当数据表中聚合列的类型都为 SUM 类型时。如果希望撤销某一批导入的数据。则可以通过 negative 参数导入同一批数据。Doris 会自动为这一批数据在聚合列上数据取反，以达到消除同一批数据的功能。
5. partition在 data_desc 中可以指定待导入表的 partition 信息，如果待导入数据不属于指定的 partition 则不会被导入。同时，不在指定 Partition 的数据会被认为是错误数据。
6. set column mapping在 data_desc 中的 SET 语句负责设置列函数变换，这里的列函数变换支持所有查询的等值表达式变换。如果原始数据的列和表中的列不一一对应，就需要用到这个属性。
7. preceding filter predicate用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。
8. where predicate在 data_desc 中的 WHERE 语句中负责过滤已经完成 transform 的数据，被 filter 的数据不会进入容忍率的统计中。如果多个 data_desc 中声明了同一张表的多个条件的话，则会 merge 同一张表的多个条件，merge 策略是 AND 。
9. 导入作业参数

导入作业参数主要指的是 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_modeBroker 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.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.ConfiguredFailoverProxyProvider"
   )
   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.ConfiguredFailoverProxyProvider"
   )
   PROPERTIES
   (
   "timeout" = "3600"
   
   );

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_4bb00753932c491a-a6da6e2725415317_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/ALOAD 的进度范围为：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.6 取消导入

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

CANCEL LOAD

[FROM db_name]

WHERE LABEL=”load_label”;

1.2 Stream Load

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

1.2.1 适用场景

Stream load 主要适用于导入本地文件，或通过程序导入数据流中的数据。

目前Stream Load支持两个数据格式：CSV（文本）和JSON。

1.2.2 基本原理

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

 

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

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

导入的最终结果由 Coordinator BE 返回给用户。

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 系统会根据签名验证用户身份和导入权限。

导入任务参数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_commitStream 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.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.2.5 取消导入

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

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

1.3 Routine Load

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

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.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.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]
5. 映射列：按顺序指定，源数据中各个列，对应目的表中的哪些列。对于希望跳过的列，可以指定一个不存在的列名。假设目的表有三列 k1, k2, v1。源数据有4列，其中第1、2、4列分别对应 k2, k1, v1。则书写如下：COLUMNS (k2, k1, xxx, v1)
   其中 xxx 为不存在的一列，用于跳过源数据中的第三列。
6. 衍生列：以 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)
7. column_separator:指定列分隔符，如：COLUMNS TERMINATED BY ","这个只在文本数据导入的时候需要指定，JSON格式的数据导入不需要指定这个参数。默认为：\t
8. columns_mapping:指定源数据中列的映射关系，以及定义衍生列的生成方式。
9. where_predicates用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：WHERE k1 > 100 and k2 = 1000
10. partitions指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。示例：PARTITION(p1, p2, p3)
11. delete_on_predicates表示删除条件，仅在 merge type 为MERGE 时有意义，语法与where 相同
12. source_sequence:只适用于UNIQUE_KEYS,相同key列下，保证value列按照source_sequence列进行REPLACE, source_sequence可以是数据源中的列，也可以是表结构中的一列。
13. preceding_predicatesPRECEDING FILTER predicate用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。
14. 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. jsonpathsjsonpaths: 导入json方式分为：简单模式和匹配模式。如果设置了jsonpath则为匹配模式导入，否则为简单模式导入，具体可参考示例
    8. strip_outer_array布尔类型，为true表示json数据以数组对象开始且将数组对象中进行展平，默认值是false
    9. json_rootjson_root为合法的jsonpath字符串，用于指定json document的根节点，默认值为""
    10. send_batch_parallelism整型，用于设置发送批处理数据的并行度，如果并行度的值超过 BE 配置中的 max_send_batch_parallelism_per_job，那么作为协调点的 BE 将使用 max_send_batch_parallelism_per_job 的值
    11. data_source_properties数据源的类型。当前支持：Kafka(
        "key1" = "val1",
        "key2" = "val2"
        )

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.3.5 查看导入作业状态

查看作业状态的具体命令和示例可以通过 HELP SHOW ROUTINE LOAD; 命令查看。

查看任务运行状态的具体命令和示例可以通过 HELP SHOW ROUTINE LOAD TASK; 命令查看。

只能查看当前正在运行中的任务，已结束和未开始的任务无法查看。

1.3.6 修改作业属性

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

1.3.7 作业控制

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

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集群中不存在时：所以，如果用户希望当 kafka topic 不存在的时候，被例行作业自动创建的话，只需要将用户方的kafka集群中的 broker 设置 auto.create.topics.enable = true 即可。

• • 如果用户 kafka 集群的 broker 设置了 auto.create.topics.enable = true，则 kafka_topic 会先被自动创建，自动创建的 partition 个数是由用户方的kafka集群中的 broker 配置 num.partitions 决定的。例行作业会正常的不断读取该 topic 的数据。
  • 如果用户 kafka 集群的 broker 设置了 auto.create.topics.enable = false, 则 topic 不会被自动创建，例行作业会在没有读取任何数据之前就被暂停，状态为 PAUSED。

5.在网络隔离的环境中可能出现的问题在有些环境中存在网段和域名解析的隔离措施，所以需要注意：

• 创建Routine load 任务中指定的 Broker list 必须能够被Doris服务访问
  • 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。

7.STOP和PAUSE的区别

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

1.4 Binlog Load

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

1.4.1 适用场景

INSERT/UPDATE/DELETE支持。

过滤Query。

暂不兼容DDL语句。

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.4.3 配置MySQL端

在MySQL Cluster模式的主从同步中，二进制日志文件（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.4.4 配置Canal端

Canal是属于阿里巴巴otter项目下的一个子项目，主要用途是基于MySQL数据库增量日志解析，提供增量数据订阅和消费，用于解决跨机房同步的业务场景，建议使用canal 1.1.5及以上版本。

下载地址：https://github.com/alibaba/canal/releases

1. 上传并解压canal deployermkdir /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
5. 启动sh bin/startup.sh
6. 验证启动成功cat logs/doris-load/doris-load.log
   注意：canal client和canal instance是一一对应的，Binlog Load已限制多个数据同步作业不能连接到同一个destination。

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.4.6 基本语法

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

CREATE SYNC [db.]job_name

(

channel_desc,

channel_desc

...

)

binlog_desc

• job_namejob_name是数据同步作业在当前数据库内的唯一标识，相同job_name的作业只能有一个在运行。
• channel_descchannel_desc用来定义任务下的数据通道，可表示MySQL源表到doris目标表的映射关系。在设置此项时，如果存在多个映射关系，必须满足MySQL源表应该与doris目标表是一一对应关系，其他的任何映射关系（如一对多关系），检查语法时都被视为不合法。FROM mysql_db.src_tbl INTO des_tbl
  [partitions]
  [columns_mapping]
  column_mapping主要指MySQL源表和doris目标表的列之间的映射关系，如果不指定，FE会默认源表和目标表的列按顺序一一对应。但是我们依然建议显式的指定列的映射关系，这样当目标表的结构发生变化（比如增加一个 nullable 的列），数据同步作业依然可以进行。否则，当发生上述变动后，因为列映射关系不再一一对应，导入将报错。
• binlog_descbinlog_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大小的最大值，默认8192canal.username: instance的用户名canal.password: instance的密码canal.debug: 设置为true时，会将batch和每一行数据的详细信息都打印出来，会影响性能。

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`;
   返回结果集的参数意义如下：
3. State作业当前所处的阶段。作业状态之间的转换如下图所示：

 

作业提交之后状态为PENDING，由FE调度执行启动canal client后状态变成RUNNING，用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止，暂停和恢复，操作后作业状态分别为CANCELLED/PAUSED/RUNNING。作业的最终阶段只有一个CANCELLED，当作业状态变为CANCELLED后，将无法再次恢复。当作业发生了错误时，若错误是不可恢复的，状态会变成CANCELLED，否则会变成PAUSED。

1. Channel作业所有源表到目标表的映射关系。
2. Status当前binlog的消费位置(若设置了GTID模式，会显示GTID)，以及doris端执行时间相比mysql端的延迟时间。
3. JobConfig对接的远端服务器信息，如canal server的地址与连接instance的destination。
4. MySQL表继续插入数据，观察Doris的表
5. 控制作业用户可以通过 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.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.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.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.6 S3 Load

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

2 数据导出

2.1 Export导出

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

2.1.1 基本原理

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

1. 调度方式

 

• PENDING：FE 生成 ExportPendingTask，向 BE 发送 snapshot 命令，对所有涉及到的 Tablet 做一个快照。并生成多个查询计划。
• EXPORTING：FE 生成 ExportExportingTask，开始执行查询计划。
  • 用户提交一个 Export 作业到 FE。
  • FE 的 Export 调度器会通过两阶段来执行一个 Export 作业：

1. 查询计划拆分

Export 作业会生成多个查询计划，每个查询计划负责扫描一部分 Tablet。每个查询计划扫描的 Tablet 个数由 FE 配置参数 export_tablet_num_per_task 指定，默认为 5。即假设一共 100 个 Tablet，则会生成 20 个查询计划。用户也可以在提交作业时，通过作业属性 tablet_num_per_task 指定这个数值。

1. 查询计划执行一个作业的多个查询计划顺序执行。一个查询计划扫描多个分片，将读取的数据以行的形式组织，每 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.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.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.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.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 查询结果导出

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

2.2.1 语法

语法如下

query_stmt

INTO OUTFILE "file_path"

[format_as]

[properties]

• file_pathfile_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 并发导出

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.3 使用示例

1. 示例一：使用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, ...。
2. 示例二：使用broker方式，指定导出格式为 PARQUETSELECT 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。
3. 示例三：使用 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, ...。
4. 示例四：使用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" = ","
   );
5. 其他用法：
6. 不存在的path会自动创建
7. access.key/secret.key/endpoint需要和cos的同学确认。尤其是endpoint的值，不需要填写bucket_name。
   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.ConfiguredFailoverProxyProvider"
      );
      最终生成文件如如果不大于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上验证
   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"
      )

文章转载来源于大数据那些事，作者尚硅谷陈老师