Doris-04-数据导入和导出&数据备份和恢复_doris备份

数据导入和导出

数据导入

导入（Load）功能就是将用户的原始数据导入到 Doris 中。导入成功后，用户即可通过Mysql 客户端查询数据。为适配不同的数据导入需求，Doris 系统提供了 6 种不同的导入方式。每种导入方式支持不同的数据源，存在不同的使用方式（异步，同步）。

所有导入方式都支持 csv 数据格式。其中 Broker load 还支持 parquet 和 orc 数据格式。

• Broker load：通过 Broker 进程访问并读取外部数据源（如 HDFS）导入到 Doris。用户通过 Mysql协议提交导入作业后，异步执行。通过 SHOW LOAD 命令查看导入结果。
• Stream load：用户通过 HTTP 协议提交请求并携带原始数据创建导入。主要用于快速将本地文件或数据流中的数据导入到 Doris。导入命令同步返回导入结果。目前 Stream Load 支持两个数据格式：CSV（文本）和 JSON。
• Insert：类似 MySQL 中的 Insert 语句，Doris 提供 INSERT INTO tbl SELECT …; 的方式从Doris 的表中读取数据并导入到另一张表。或者通过 INSERT INTO tbl VALUES(…); 插入单条数据。
• Multi load：用户通过 HTTP 协议提交多个导入作业。Multi Load 可以保证多个导入作业的原子生效。
• Routine load：用户通过 MySQL 协议提交例行导入作业，生成一个常驻线程，不间断的从数据源（如Kafka）中读取数据并导入到 Doris 中。
• 通过S3 协议直接导入：用户通过 S3 协议直接导入数据，用法和 Broker Load 类似。Broker load 是一个异步的导入方式，支持的数据源取决于 Broker 进程支持的数据源。用户需要通过 MySQL 协议创建 Broker load 导入，并通过查看导入命令检查导入结果。
• Binlog Load：提供了一种使Doris增量同步用户在Mysql数据库的对数据更新操作的CDC(Change Data Capture)功能。需要依赖canal作为中间媒介。

Broker Load

（1）适用场景

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

（2）基本原理

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

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

（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, ...)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

创建导入的详细语法执行 HELP BROKER LOAD 查看语法帮助。这里主要介绍 Broker load 的创建导入语法中参数意义和注意事项。

• Label：导入任务的标识。每个导入任务，都有一个在单 database 内部唯一的 Label。Label 是用户在导入命令中自定义的名称。通过这个 Label，用户可以查看对应导入任务的执行情况。

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

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

• 数据描述类参数：数据描述类参数主要指的是 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 的数据会被认为是错误数据。
  • preceding filter predicate：用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。
  • set column mapping：在 data_desc 中的 SET 语句负责设置列函数变换，这里的列函数变换支持所有查询的等值表达式变换。如果原始数据的列和表中的列不一一对应，就需要用到这个属性。
  • where predicate：在 data_desc 中的 WHERE 语句中负责过滤已经完成 transform 的数据，被 filter 的数据不会进入容忍率的统计中。如果多个 data_desc 中声明了同一张表的多个条件的话，则会merge 同一张表的多个条件，merge 策略是 AND 。

• 导入作业参数：导入作业参数主要指的是 Broker load 创建导入语句中的属于 opt_properties 部分的参数。导入作业参数是作用于整个导入作业的。

  下面主要对导入作业参数的部分参数详细解释：

  • timeout：导入作业的超时时间(以秒为单位)，用户可以在 opt_properties 中自行设置每个导入的超时时间。导入任务在设定的 timeout 时间内未完成则会被系统取消，变成 CANCELLED。Broker load 的默认导入超时时间为 4 小时。

    通常情况下，用户不需要手动设置导入任务的超时时间。当在默认超时时间内无法完成导入时，可以手动设置任务的超时时间。

    推荐超时时间：总文件大小（MB） / 用户 Doris 集群最慢导入速度(MB/s) > timeout > （（总文件大小(MB) * 待导入的表及相关 Roll up 表的个数） / (10 * 导入并发数） ）

  • max_filter_ratio：导入任务的最大容忍率，默认为 0 容忍，取值范围是 0~1。当导入的错误率超过该值，则导入失败。

    如果用户希望忽略错误的行，可以通过设置这个参数大于 0，来保证导入可以成功。

    计算公式为：

    max_filter_ratio = (dpp.abnorm.ALL / (dpp.abnorm.ALL + dpp.norm.ALL ) )
    1

    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 语义处理。

（4）导入示例

• 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;
  1
  2
  3
  4
  5
  6
  7
  8
  9

• 文件上传 HDFS

  启动 HDFS 相关服务：

  hadoop fs -put student.csv /
  1

• 导入数据

  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"
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21

• 通用文件格式写法：

  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
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27

（5）查看导入

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

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}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

下面主要介绍了查看导入命令返回结果集中参数意义：

• 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。

  其中已处理的原始行数，每 5 秒更新一次。该行数仅用于展示当前的进度，不代表最终实际的处理行数。实际处理行数以 EtlInfo 中显示的为准。

（6）取消导入

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

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

Stream Load

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

（1）适用场景

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

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

（2）基本原理

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

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

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

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

（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
1

创建导入的详细语法帮助执行 HELP STREAM LOAD 查看, 下面主要介绍创建 Stream load 的部分参数意义。

• 签名参数：user/passwd

  Stream load 由于创建导入的协议使用的是 HTTP 协议，通过 Basic access authentication 进行签名。Doris 系统会根据签名验证用户身份和导入权限。

• 导入任务参数

  Stream load 由于使用的是 HTTP 协议，所以所有导入任务有关的参数均设置在Header 中。格式为： -H “key1:value1”。下面主要介绍了 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
    1
    2
    3
    4
    5
    6

    用户可以调用如下接口对 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
    3
    4
    5

（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
1
2

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

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

（5）取消导入

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

Routine Load

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

（1）适用场景

当前仅支持从 Kafka 系统进行例行导入，使用限制：

• 支持无认证的 Kafka 访问，以及通过 SSL 方式认证的 Kafka 集群。
• 支持的消息格式为 csv, json 文本格式。csv 每一个 message 为一行，且行尾不包含换行符。
• 默认支持 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。

（2）基本原理

如上图，Client 向 FE 提交一个例行导入作业。

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

**（3）基本语法 **

CREATE ROUTINE LOAD [db.]job_name ON tbl_name
[merge_type]
[load_properties]
[job_properties]
FROM data_source
[data_source_properties]
1
2
3
4
5
6

• [db.]job_name：导入作业的名称，在同一个 database 内，相同名称只能有一个 job 在运行。

• tbl_name：指定需要导入的表的名称。

• merge_type：数据的合并类型，一共支持三种类型 APPEND、DELETE、MERGE 其中，APPEND 是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据 key 相同的所有行，MERGE 语义 需要与 delete on 条件联合使用，表示满足 delete 条件的数据按照 DELETE 语义处理其余的按照 APPEND 语义处理 , 语法为 [WITH MERGE|APPEND|DELETE]

• load_properties：用于描述导入数据。语法： [column_separator], [columns_mapping], [where_predicates], [delete_on_predicates], [source_sequence], [partitions], [preceding_predicates]

  • column_separator：指定列分隔符，如： COLUMNS TERMINATED BY ","这个只在文本数据导入的时候需要指定，JSON 格式的数据导入不需要指定这个参数。默认为：\t ；

  • 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)

  • where_predicates：用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。 例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：WHERE k1 > 100 and k2 = 1000

  • partitions：指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。

    示例：PARTITION(p1, p2, p3)

  • delete_on_predicates：表示删除条件，仅在 merge type 为 MERGE 时有意义，语法与 where 相同

  • source_sequence:只适用于 UNIQUE_KEYS,相同 key 列下，保证 value 列按照 source_sequence 列进行REPLACE, source_sequence 可以是数据源中的列，也可以是表结构中的一列。

  • preceding_predicates：PRECEDING FILTER predicate。用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。

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

  PROPERTIES (
  "key1" = "val1",
  "key2" = "val2"
  )
  1
  2
  3
  4

  目前支持以下参数：

  • desired_concurrent_number：期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于 0。默认为 3。 这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。

    一个作业，最多有多少 task 同时在执行。对于 Kafka 导入而言，当前的实际并发度计算如下：

    Min(partition num, desired_concurrent_number, alive_backend_num, 
    Config.max_routine_load_task_concurrrent_num)
    1
    2

    其中 Config.max_routine_load_task_concurrrent_num 是系统的一个默认的最大并发数限制。这是一个 FE 配置，可以通过改配置调整。默认为 5。

    其中 partition num 指订阅的 Kafka topic 的 partition 数量。alive_backend_num 是当前正常的 BE 节点数。

  • 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"
    1
    2
    3

  • max_error_number：采样窗口内，允许的最大错误行数。必须大于等于 0。默认是 0，即不允许有错误行。

    采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被暂停，需要人工介入检查数据质量问题。 被 where 条件过滤掉的行不算错误行

  • strict_mode：是否开启严格模式，默认为关闭。如果开启后，非空原始数据的列类型变换如果结果为NULL，则会被过滤。指定方式为 “strict_mode” = “true”

  • timezone：指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果

  • format：指定导入数据格式，默认是 csv，支持 json 格式

  • jsonpaths：导入 json 方式分为：简单模式和匹配模式。如果设置了 jsonpath 则为匹配模式导入，否则为简单模式导入，具体可参考示例

  • strip_outer_array：布尔类型，为 true 表示 json 数据以数组对象开始且将数组对象中进行展平，默认值是false

  • json_root：json_root 为合法的 jsonpath 字符串，用于指定 json document 的根节点，默认值为""

  • send_batch_parallelism：整型，用于设置发送批处理数据的并行度，如果并行度的值超过 BE 配置中的max_send_batch_parallelism_per_job ， 那么作为协调点的BE 将 使 用max_send_batch_parallelism_per_job 的值

  • data_source_properties：数据源的类型。当前支持：Kafka

    (
    "key1" = "val1",
    "key2" = "val2"
    )
    1
    2
    3
    4

（4）Kafka 导入示例

• 在 doris 中创建对应表

  create table student_kafka
  (
      id int,
      name varchar(50),
      age int
  )
  DUPLICATE KEY(id)
  DISTRIBUTED BY HASH(id) BUCKETS 10;
  1
  2
  3
  4
  5
  6
  7
  8

• 启动 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
  1
  2
  3
  4
  5
  6
  7
  8

• 创建导入任务

  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"
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15

• 查看表

  select * from student_kafka;
  1

  继续往 kafka 发送数据，查看表的变化

（5）查看导入作业状态

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

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

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

（6）修改作业属性

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

（7）作业控制

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

（8）其他说明

• 例行导入作业和 ALTER TABLE 操作的关系

  例行导入不会阻塞 SCHEMA CHANGE 和 ROLLUP 操作。但是注意如果 SCHEMA CHANGE 完成后，列映射关系无法匹配，则会导致作业的错误数据激增，最终导致作业暂停。建议通过在例行导入作业中显式指定列映射关系，以及通过增加 Nullable 列或带Default 值的列来减少这类问题。

  删除表的 Partition 可能会导致导入数据无法找到对应的 Partition，作业进入暂停。

• 例行导入作业和其他导入作业的关系（LOAD, DELETE, INSERT）

  例行导入和其他 LOAD 作业以及 INSERT 操作没有冲突。当执行 DELETE 操作时，对应表分区不能有任何正在执行的导入任务。所以在执行DELETE 操作前，可能需要先暂停例行导入作业，并等待已下发的 task 全部完成后，才可以执行 DELETE。

• 例行导入作业和DROP DATABASE/TABLE 操作的关系

  当例行导入对应的 database 或 table 被删除后，作业会自动 CANCEL。

• kafka类型的例行导入作业和 kafka topic 的关系

  当用户在创建例行导入声明的 kafka_topic 在 kafka 集群中不存在时：

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

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

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

  • 创建 Routine load 任务中指定的 Broker list 必须能够被 Doris 服务访问
  • Kafka 中如果配置了 advertised.listeners, advertised.listeners 中的地址必须能够被Doris 服务访问

• 关于指定消费的 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 指定的位置开始消费

• STOP 和 PAUSE 的区别

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

Binlog Load

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

（1）适用场景

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

（2）基本原理

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

如上图，用户向 FE 提交一个数据同步作业。

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

（3）配置 MySQL 端

在 MySQLCluster 模式的主从同步中，二进制日志文件（Binlog）记录了主节点上的所有数据变化，数据在 Cluster 的多个节点间同步、备份都要通过 Binlog 日志进行，从而提高集群的可用性。架构通常由一个主节点（负责写）和一个或多个从节点（负责读）构成，所有在主节点上发生的数据变更将会复制给从节点。

注意：目前必须要使用 Mysql 5.7 及以上的版本才能支持 Binlog Load 功能。

• 打开 mysql 的二进制 binlog 日志功能，编辑 my.cnf 配置文件

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

• 开启 GTID 模式 [可选]

  一个全局事务 Id(global transaction identifier)标识出了一个曾在主节点上提交过的事务，在全局都是唯一有效的。开启了 Binlog 后，GTID 会被写入到 Binlog 文件中，与事务一一对应。

  编辑 my.cnf 配置文件。

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

  在 GTID 模式下，主服务器可以不需要 Binlog 的文件名和偏移量，就能很方便的追踪事务、恢复数据、复制副本。

  在 GTID 模式下，由于 GTID 的全局有效性，从节点将不再需要通过保存文件名和偏移量来定位主节点上的 Binlog 位置，而通过数据本身就可以定位了。在进行数据同步中，从节点会跳过执行任意被识别为已执行的 GTID 事务。

  GTID 的表现形式为一对坐标, source_id 标识出主节点，transaction_id 表示此事务在主节点上执行的顺序（最大 263-1）。

• 重启 MySQL 使配置生效

  sudo systemctl restart mysqld
  1

• 创建用户并授权

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

• 准备测试表

  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
  2
  3
  4
  5

（4）配置 Canal 端

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

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

• 上传并解压 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
  1

• 在 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
  1
  2
  3

• 修改 conf/canal.properties 的配置

  canal.destinations = doris-load
  1

• 修改 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
  1
  2
  3
  4
  5
  6
  7
  8

• 启动

  sh bin/startup.sh
  1

• 验证启动成功

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

  注意：canal client 和 canal instance 是一一对应的，Binlog Load 已限制多个数据同步作业不能连接到同一个 destination。

（5）配置目标表

• 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;
  1
  2
  3
  4
  5
  6
  7

  Binlog Load 只能支持 Unique 类型的目标表，且必须激活目标表的 Batch Delete 功能。

• 开启 SYNC 功能

  在 fe.conf 中将 enable_create_sync_job 设为 true，不想修改配置文件重启，可以执行如下：

  使用 root 账号登陆
  ADMIN SET FRONTEND CONFIG ("enable_create_sync_job" = "true");
  1
  2

（6）基本语法

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

CREATE SYNC [db.]job_name
(
 channel_desc, 
 channel_desc
 ...
)
binlog_desc
1
2
3
4
5
6
7

• 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]
  1
  2
  3

  column_mapping主要指MySQL源表和doris目标表的列之间的映射关系，如果不指定，FE 会默认源表和目标表的列按顺序一一对应。但是我们依然建议显式的指定列的映射关系，这样当目标表的结构发生变化（比如增加一个 nullable 的列），数据同步作业依然可以进行。否则，当发生上述变动后，因为列映射关系不再一一对应，导入将报错。

• binlog_desc：binlog_desc 中的属性定义了对接远端 Binlog 地址的一些必要信息，目前可支持的对接类型只有 canal 方式，所有的配置项前都需要加上 canal 前缀。

  FROM BINLOG
  (
   "key1" = "value1", 
   "key2" = "value2"
  )
  1
  2
  3
  4
  5

  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 和每一行数据的详细信息都打印出来，会影响性能。

（7）示例

• 创建同步作业：

  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"
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13

• 查看作业状态：查看作业状态的具体命令和示例可以通过 HELP SHOW SYNC JOB; 命令查看。

  # 展示当前数据库的所有数据同步作业状态。
  SHOW SYNC JOB;
  # 展示数据库 `test_db` 下的所有数据同步作业状态。
  SHOW SYNC JOB FROM `test_db`;
  1
  2
  3
  4

  返回结果集的参数意义如下：

  • 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。

• MySQL 表继续插入数据，观察 Doris 的表

• 控制作业：用户可以通过 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
  2
  3
  4
  5
  6

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, ...);
1
2

其中第二种命令仅用于 Demo，不要使用在测试或生产环境中。

Insert Into 命令需要通过 MySQL 协议提交，创建导入请求会同步返回导入结果。

（1）语法

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

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");
1
2

注意：

当需要使用 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
2
3
4
5
6

（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
1
2
3
4
5
6
7
8
9

该命令会返回 insert 以及对应事务的详细信息。因此，用户可以在每次执行完 insert 操作后，继续执行 show last insert 命令来获取 insert 的结果。

注意：该命令只会返回在同一 session 连接中，最近一次 insert 操作的结果。如果连接断开或更换了新的连接，则将返回空集。

S3 Load

从0.14 版本开始，Doris 支持通过 S3 协议直接从支持 S3 协议的在线存储系统导入数据。

本文档主要介绍如何导入 AWS S3 中存储的数据。也支持导入其他支持 S3 协议的对象存储系统导入，如百度云的 BOS、阿里云的OSS和腾讯云的 COS 等。

（1）适用场景

• 源数据在 支持 S3 协议的存储系统中，如 S3,BOS 等。
• 数据量在 几十到百 GB 级别。

（2）准备工作

• 准备AK 和 SK 首先需要找到或者重新生成 AWS Access keys，可以在 AWS console 的 My Security Credentials 找到生成方式；
• 准备 REGION 和 ENDPOINT REGION 可以在创建桶的时候选择也可以在桶列表中查看到。ENDPOINT 可以通过如下页面通过 REGION 查到 AWS 文档。

（3）示例

导入方式和 Broker Load 基本相同，只需要将 WITH BROKER broker_name () 语句替换成如下部分

WITH S3
(
    "AWS_ENDPOINT" = "AWS_ENDPOINT",
    "AWS_ACCESS_KEY" = "AWS_ACCESS_KEY",
    "AWS_SECRET_KEY"="AWS_SECRET_KEY",
    "AWS_REGION" = "AWS_REGION"
)
1
2
3
4
5
6
7

完整示例如下：

LOAD LABEL example_db.exmpale_label_1
(
    DATA INFILE("s3://your_bucket_name/your_file.txt")
    INTO TABLE load_test
    COLUMNS TERMINATED BY ","
)
WITH S3
(
    "AWS_ENDPOINT" = "AWS_ENDPOINT",
    "AWS_ACCESS_KEY" = "AWS_ACCESS_KEY",
    "AWS_SECRET_KEY"="AWS_SECRET_KEY",
    "AWS_REGION" = "AWS_REGION"
)
PROPERTIES
(
    "timeout" = "3600"
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

数据导出

Export导出

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

（1）基本原理

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

a. 调度方式：

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

b. 查询计划拆分

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

c. 查询计划执行

一个作业的多个查询计划顺序执行。

一个查询计划扫描多个分片，将读取的数据以行的形式组织，每 1024 行为一个 batch，调用 Broker 写入到远端存储上。

查询计划遇到错误会整体自动重试 3 次。如果一个查询计划重试 3 次依然失败，则整个作业失败。

Doris 会首先在指定的远端存储的路径中，建立一个名为 __doris_export_tmp_12345 的临时目录（其中 12345 为作业 id）。导出的数据首先会写入这个临时目录。每个查询计划会生成一个文件，文件名示例：

export-data-c69fcf2b6db5420f-a96b94c1ff8bccef-1561453713822
1

其中 c69fcf2b6db5420f-a96b94c1ff8bccef 为查询计划的 query id。1561453713822 为文件生成的时间戳。

当所有数据都导出后，Doris 会将这些文件 rename 到用户指定的路径中。

（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"
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

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

（3）导出示例

• 启动 hadoop 集群

• 执行导出

  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" 
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19

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

（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
1
2
3
4
5
6
7
8
9
10
11
12
13
14

• 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：如果作业出现错误，这里会显示错误原因

（5）注意事项

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

查询结果导出

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

（1）语法

语法如下：

query_stmt
INTO OUTFILE "file_path"
[format_as]
[properties]
1
2
3
4

• 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
  1

  指定导出格式。默认为 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", ...)
  1
  2
  3
  4
  5

  其它属性：

  ("key1" = "val1", "key2" = "val2", ...)
  1

  目前支持以下属性：

  • column_separator：列分隔符，仅对 CSV 格式适用。默认为 \t。
  • line_delimiter：行分隔符，仅对 CSV 格式适用。默认为 \n。
  • max_file_size：单个文件的最大大小。默认为 1GB。取值范围在 5MB 到 2GB 之间。超过这个大小的文件将会被切分。
  • schema：PARQUET 文件 schema 信息。仅对 PARQUET 格式适用。导出文件格式为 PARQUET 时，必须指定 schema。

（2）并发导出

• 并发导出的条件：默认情况下，查询结果集的导出是非并发的，也就是单点导出。如果用户希望查询结果集可以并发导出，需要满足以下条件：

  • session variable ‘enable_parallel_outfile’ 开启并发导出：

    set enable_parallel_outfile = true;

  • 导出方式为 S3 , 或者 HDFS，而不是使用 broker；

  • 查询可以满足并发导出的需求，比如顶层不包含 sort 等单点节点。（后面会举例说明，哪种属于不可并发导出结果集的查询）

  满足以上三个条件，就能触发并发导出查询结果集了。

  并发度 = be_instacne_num * parallel_fragment_exec_instance_num

• 验证结果集被并发导出。

  用户通过 session 变量设置开启并发导出后，如果想验证当前查询是否能进行并发导出，则可以通过下面这个方法。

  explain select xxx from xxx where xxx into outfile "s3://xxx" 
  format as csv properties ("AWS_ENDPOINT" = "xxx", ...);
  1
  2

  对查询进行 explain 后，Doris 会返回该查询的规划，如果发现 RESULT FILE SINK 出现在 PLAN FRAGMENT 1 中，就说明导出并发开启成功了。如果 RESULT FILE SINK 出现在PLAN FRAGMENT 0 中，则说明当前查询不能进行并发导出（当前查询不同时满足并发导出的三个条件）。

  

（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"
);
1
2
3
4
5
6
7
8
9
10

最终生成文件如如果不大于 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"
);
1
2
3
4
5
6
7
8

查询结果导出到 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" = ","
);
1
2
3
4
5
6
7
8
9

最终生成文件如如果不大于 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
2
3
4
5
6
7
8
9
10

**示例五：**将 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"
);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

最终生成文件如如果不大于 1GB，则为：result_0.csv。

如果大于 1GB，则可能为 result_0.csv, result_1.csv, ...。

**示例六：**将 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"
);
1
2
3
4
5
6
7
8
9
10
11

**示例七：**将 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"
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

最终生成文件如如果不大于 1GB，则为：my_file_0.csv。

如果大于 1GB，则可能为 my_file_0.csv, result_1.csv, ...。

在 cos 上验证：

① 不存在的 path 会自动创建

② access.key/secret.key/endpoint 需要和 cos 的同学确认。尤其是 endpoint 的值，不需要填写 bucket_name。

**示例八：**使用 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"
)
1
2
3
4
5
6
7
8
9
10
11

最终生成的文件前缀为 my_file_{fragment_instance_id}_。

**示例九：**使用 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"
)
1
2
3
4
5
6
7
8
9
10
11

mysqldump 导出

Doris 1.0 支持通过 mysqldump 工具导出数据或者表结构，下面几种操作：

• 导出 test 数据库中的 user 表：

  mysqldump -h127.0.0.1 -P9030 -uroot --no-tablespaces --databases test_db --tables user > dump1.sql
  1

• 导出 test_db 数据库中的 user 表结构：

  mysqldump -h127.0.0.1 -P9030 -uroot --no-tablespaces --databases test_db --tables user --no-data > dump2.sql
  1

• 导出 test_db 数据库中所有表：

  mysqldump -h127.0.0.1 -P9030 -uroot --no-tablespaces --databases test_db
  1

• 导出所有数据库和表

  mysqldump -h127.0.0.1 -P9030 -uroot --no-tablespaces --all-databases
  1

• 导出的结果可以重定向到文件中，之后可以通过 source 命令导入到 Doris 中

  source /opt/module/doris-1.0.0/dump1.sql
  1

数据备份及恢复

Doris 支持将当前数据以文件的形式，通过 broker 备份到远端存储系统中。之后可以通过 恢复 命令，从远端存储系统中将数据恢复到任意 Doris 集群。通过这个功能，Doris 可以支持将数据定期的进行快照备份。也可以通过这个功能，在不同集群间进行数据迁移。

该功能需要 Doris 版本 0.8.2+使用该功能，需要部署对应远端存储的 broker。如 BOS、HDFS 等。可以通过 SHOW BROKER; 查看当前部署的 broker。

简要原理说明

备份（Backup）

备份操作是将指定表或分区的数据，直接以 Doris 存储的文件的形式，上传到远端仓库中进行存储。当用户提交 Backup 请求后，系统内部会做如下操作：

（1）快照及快照上传

快照阶段会对指定的表或分区数据文件进行快照。之后，备份都是对快照进行操作。在快照之后，对表进行的更改、导入等操作都不再影响备份的结果。快照只是对当前数据文件产生一个硬链，耗时很少。快照完成后，会开始对这些快照文件进行逐一上传。快照上传由各个 Backend 并发完成。

（2）元数据准备及上传

数据文件快照上传完成后，Frontend 会首先将对应元数据写成本地文件，然后通过broker 将本地元数据文件上传到远端仓库。完成最终备份作业。

恢复（Restore）

恢复操作需要指定一个远端仓库中已存在的备份，然后将这个备份的内容恢复到本地集群中。当用户提交 Restore 请求后，系统内部会做如下操作：

（1）在本地创建对应的元数据

这一步首先会在本地集群中，创建恢复对应的表分区等结构。创建完成后，该表可见，但是不可访问。

（2）本地 snapshot

这一步是将上一步创建的表做一个快照。这其实是一个空快照（因为刚创建的表是没有数据的），其目的主要是在 Backend 上产生对应的快照目录，用于之后接收从远端仓库下载的快照文件。

（3）下载快照

远端仓库中的快照文件，会被下载到对应的上一步生成的快照目录中。这一步由各个Backend 并发完成。

（4）生效快照

快照下载完成后，我们要将各个快照映射为当前本地表的元数据。然后重新加载这些快照，使之生效，完成最终的恢复作业。

最佳实践

（1）备份

当前支持最小分区（Partition）粒度的全量备份（增量备份有可能在未来版本支持）。如果需要对数据进行定期备份，首先需要在建表时，合理的规划表的分区及分桶，比如按时间进行分区。然后在之后的运行过程中，按照分区粒度进行定期的数据备份。

（2）数据迁移

用户可以先将数据备份到远端仓库，再通过远端仓库将数据恢复到另一个集群，完成数据迁移。因为数据备份是通过快照的形式完成的，所以，在备份作业的快照阶段之后的新的导入数据，是不会备份的。因此，在快照完成后，到恢复作业完成这期间，在原集群上导入的数据，都需要在新集群上同样导入一遍。

建议在迁移完成后，对新旧两个集群并行导入一段时间。完成数据和业务正确性校验后，再将业务迁移到新的集群。

（3）重点说明

• 备份恢复相关的操作目前只允许拥有 ADMIN 权限的用户执行。
• 一个 Database 内，只允许有一个正在执行的备份或恢复作业。
• 备份和恢复都支持最小分区（Partition）级别的操作，当表的数据量很大时，建议按分区分别执行，以降低失败重试的代价。
• 因为备份恢复操作，操作的都是实际的数据文件。所以当一个表的分片过多，或者一个分片有过多的小版本时，可能即使总数据量很小，依然需要备份或恢复很长时间。用户可以通过 SHOW PARTITIONS FROM table_name; 和 SHOW TABLET FROM table_name; 来查看各个分区的分片数量，以及各个分片的文件版本数量，来预估作业执行时间。文件数量对作业执行的时间影响非常大，所以建议在建表时，合理规划分区分桶，以避免过多的分片。
• 当通过 SHOW BACKUP 或者 SHOW RESTORE 命令查看作业状态时。有可能会在 TaskErrMsg 一列中看到错误信息。但只要 State 列不为 CANCELLED，则说明作业依然在继续。这些 Task 有可能会重试成功。当然，有些 Task 错误，也会直接导致作业失败。
• 如果恢复作业是一次覆盖操作（指定恢复数据到已经存在的表或分区中），那么从恢复作业的 COMMIT 阶段开始，当前集群上被覆盖的数据有可能不能再被还原。此时如果恢复作业失败或被取消，有可能造成之前的数据已损坏且无法访问。这种情况下，只能通过再次执行恢复操作，并等待作业完成。因此，我们建议，如无必要，尽量不要使用覆盖的方式恢复数据，除非确认当前数据已不再使用。

备份

（1）创建一个远端仓库路径

CREATE REPOSITORY `hdfs_ods_dw_backup`
WITH BROKER `broker_name`
ON LOCATION "hdfs://hadoop1:8020/tmp/doris_backup"
PROPERTIES (
    "username" = "",
    "password" = ""
)
1
2
3
4
5
6
7

（2）执行备份

BACKUP SNAPSHOT [db_name].{snapshot_name}
TO `repository_name`
ON (
    `table_name` [PARTITION (`p1`, ...)],
    ...
)
PROPERTIES ("key"="value", ...);
1
2
3
4
5
6
7

示例：

BACKUP SNAPSHOT test_db.backup1
 TO hdfs_ods_dw_backup
 ON
 (
     table1
 );
1
2
3
4
5
6

（3）查看备份任务

SHOW BACKUP [FROM db_name]
1

（4）查看远端仓库镜像

语法：

SHOW SNAPSHOT ON `repo_name` [WHERE SNAPSHOT = "snapshot" [AND TIMESTAMP = "backup_timestamp"]];
1

示例一：查看仓库 hdfs_ods_dw_backup 中已有的备份：

SHOW SNAPSHOT ON hdfs_ods_dw_backup;    
1

示例二：仅查看仓库 hdfs_ods_dw_backup 中名称为 backup1 的备份：

SHOW SNAPSHOT ON hdfs_ods_dw_backup WHERE SNAPSHOT = "backup1";
1

示例三：查看仓库 hdfs_ods_dw_backup 中名称为 backup1 的备份，时间版本为 “2021-05-05-15-34-26” 的详细信息：

SHOW SNAPSHOT ON hdfs_ods_dw_backup WHERE SNAPSHOT = "backup1" AND TIMESTAMP = "2021-05-05-15-34-26";
1

（5）取消备份

取消一个正在执行的备份作业语法：

CANCEL BACKUP FROM db_name;
1

示例：取消 test_db 下的 BACKUP 任务

CANCEL BACKUP FROM test_db;
1

恢复

将之前通过 BACKUP 命令备份的数据，恢复到指定数据库下。该命令为异步操作。提交成功后，需通过 SHOW RESTORE 命令查看进度。

• 仅支持恢复 OLAP 类型的表
• 支持一次恢复多张表，这个需要和你对应的备份里的表一致

（1）使用用法

RESTORE SNAPSHOT [db_name].{snapshot_name}
FROM `repository_name`
ON (
    `table_name` [PARTITION (`p1`, ...)] [AS `tbl_alias`],
    ...
)
PROPERTIES ("key"="value", ...);
1
2
3
4
5
6
7

说明：

• 同一数据库下只能有一个正在执行的 BACKUP 或 RESTORE 任务。

• ON 子句中标识需要恢复的表和分区。如果不指定分区，则默认恢复该表的所有分区。所指定的表和分区必须已存在于仓库备份中

• 可以通过 AS 语句将仓库中备份的表名恢复为新的表。但新表名不能已存在于数据库中。分区名称不能修改。

• 可以将仓库中备份的表恢复替换数据库中已有的同名表，但须保证两张表的表结构完全一致。表结构包括：表名、列、分区、Rollup 等等。

• 可以指定恢复表的部分分区，系统会检查分区 Range 或者 List 是否能够匹配。

• PROPERTIES 目前支持以下属性：

  “backup_timestamp” = “2018-05-04-16-45-08”：指定了恢复对应备份的哪个时间版本，必填。该信息可以通过 SHOW SNAPSHOT ON repo; 语句获得。

  “replication_num” = “3”：指定恢复的表或分区的副本数。默认为 3。若恢复已存在的表或分区，则副本数必须和已存在表或分区的副本数相同。同时，必须有足够的host 容纳多个副本。

  “timeout” = “3600”：任务超时时间，默认为一天。单位秒。

  “meta_version” = 40：使用指定的 meta_version 来读取之前备份的元数据。注意，该参数作为临时方案，仅用于恢复老版本 Doris 备份的数据。最新版本的备份数据中已经包含 meta version，无需再指定。

（2）使用示例

• 示例一

  从 example_repo 中恢复备份 snapshot_1 中的表 backup_tbl 到数据库 example_db1，时间版本为 “2021-05-04-16-45-08”。恢复为 1 个副本：

  RESTORE SNAPSHOT example_db1.`snapshot_1`
  FROM `example_repo`
  ON ( `backup_tbl` )
  PROPERTIES
  (
      "backup_timestamp"="2021-05-04-16-45-08",
      "replication_num" = "1"
  );
  1
  2
  3
  4
  5
  6
  7
  8

• 示例二

  从 example_repo 中恢复备份 snapshot_2 中的表 backup_tbl 的分区 p1,p2，以及表backup_tbl2 到数据库 example_db1，并重命名为 new_tbl，时间版本为 “2021-05-04-17-11-01”。默认恢复为 3 个副本：

  RESTORE SNAPSHOT example_db1.`snapshot_2`
  FROM `example_repo`
  ON
  (
   `backup_tbl` PARTITION (`p1`, `p2`),
   `backup_tbl2` AS `new_tbl`
  )
  PROPERTIES
  (
   "backup_timestamp"="2021-05-04-17-11-01"
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

• 演示

  RESTORE SNAPSHOT test_db.backup1 
  FROM `hdfs_ods_dw_backup` 
  ON 
  (
      table1 AS table_restore
  )
  PROPERTIES 
  (
      "backup_timestamp"="2022-04-01-16-45-19" 
  );
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

（3）查看恢复任务

可以通过下面的语句查看数据恢复的情况：

SHOW RESTORE [FROM db_name]
1

（4）取消恢复

下面的语句用于取消一个正在执行数据恢复的作业：

CANCEL RESTORE FROM db_name;
1

当取消处于 COMMIT 或之后阶段的恢复左右时，可能导致被恢复的表无法访问。此时只能通过再次执行恢复作业进行数据恢复。

删除远端仓库

该语句用于删除一个已创建的仓库。仅 root 或 superuser 用户可以删除仓库。这里的用户是指 Doris 的用户 语法：

DROP REPOSITORY `repo_name`;
1

说明：

删除仓库，仅仅是删除该仓库在 Doris 中的映射，不会删除实际的仓库数据。删除后，可以再次通过指定相同的 broker 和 LOCATION 映射到该仓库。