Broker Load

因为 Doris 表里的数据是有序的，所以 Broker load 在导入数据的时是要利用doris 集群资源对数据进行排序，相对于 Spark load 来完成海量历史数据迁移，对 Doris 的集群资源占用要比较大，这种方式是在用户没有 Spark 这种计算资源的情况下使用，如果有 Spark 计算资源建议使用 Spark load。

用户需要通过 MySQL 协议创建导入，并通过查看导入命令检查导入结果。

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

基本原理

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

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

开始导入

下面我们通过几个实际的场景示例来看的使用

创建 Hive 表


CREATE TABLE `ods_demo_detail`(
  `id` string, 
  `store_id` string, 
  `company_id` string, 
  `tower_id` string, 
  `commodity_id` string, 
  `commodity_name` string, 
  `commodity_price` double, 
  `member_price` double, 
  `cost_price` double, 
  `unit` string, 
  `quantity` double, 
  `actual_price` double
)
PARTITIONED BY (day string)
row format delimited fields terminated by ','
lines terminated by '\n'

然后使用 Hive 的 Load 命令将你的数据导入到 Hive 表中

load data local inpath '/opt/custorm' into table ods_demo_detail;

创建 Doris 表，具体建表语法参照：

CREATE TABLE `doris_ods_test_detail` (
  `rq` date NULL,
  `id` varchar(32) NOT NULL,
  `store_id` varchar(32) NULL,
  `company_id` varchar(32) NULL,
  `tower_id` varchar(32) NULL,
  `commodity_id` varchar(32) NULL,
  `commodity_name` varchar(500) NULL,
  `commodity_price` decimal(10, 2) NULL,
  `member_price` decimal(10, 2) NULL,
  `cost_price` decimal(10, 2) NULL,
  `unit` varchar(50) NULL,
  `quantity` int(11) NULL,
  `actual_price` decimal(10, 2) NULL
) ENGINE=OLAP
UNIQUE KEY(`rq`, `id`, `store_id`)
PARTITION BY RANGE(`rq`)
(
PARTITION P_202204 VALUES [('2022-04-01'), ('2022-05-01')))
PROPERTIES (
"replication_allocation" = "tag.location.default: 3",
"dynamic_partition.enable" = "true",
"dynamic_partition.time_unit" = "MONTH",
"dynamic_partition.start" = "-2147483648",
"dynamic_partition.end" = "2",
"dynamic_partition.prefix" = "P_",
"dynamic_partition.buckets" = "1",
"in_memory" = "false",
"storage_format" = "V2"
);

开始导入数据

具体语法参照： Broker Load

LOAD LABEL broker_load_2022_03_23
(
    DATA INFILE("hdfs://192.168.20.123:8020/user/hive/warehouse/ods.db/ods_demo_detail/*/*")
    INTO TABLE doris_ods_test_detail
    COLUMNS TERMINATED BY ","
    COLUMNS FROM PATH AS (`day`)
   SET 
   (rq = str_to_date(`day`,'%Y-%m-%d'),id=id,store_id=store_id,company_id=company_id,tower_id=tower_id,commodity_id=commodity_id,commodity_name=commodity_name,commodity_price=commodity_price,member_price=member_price,cost_price=cost_price,unit=unit,quantity=quantity,actual_price=actual_price)
    )
WITH BROKER "broker_name_1" 
    ( 
      "username" = "hdfs", 
      "password" = "" 
    )
PROPERTIES
(
    "timeout"="1200",
    "max_filter_ratio"="0.1"
);

Hive 分区表导入(ORC格式)

创建Hive分区表，ORC格式

创建Doris表，这里的建表语句和上面的Doris建表语句一样，请参考上面的.

使用 Broker Load 导入数据

LOAD LABEL dish_2022_03_23
(
    DATA INFILE("hdfs://10.220.147.151:8020/user/hive/warehouse/ods.db/ods_demo_orc_detail/*/*")
    INTO TABLE doris_ods_test_detail
    COLUMNS TERMINATED BY ","
    FORMAT AS "orc"
(id,store_id,company_id,tower_id,commodity_id,commodity_name,commodity_price,member_price,cost_price,unit,quantity,actual_price) 
    COLUMNS FROM PATH AS (`day`)
   SET 
   (rq = str_to_date(`day`,'%Y-%m-%d'),id=id,store_id=store_id,company_id=company_id,tower_id=tower_id,commodity_id=commodity_id,commodity_name=commodity_name,commodity_price=commodity_price,member_price=member_price,cost_price=cost_price,unit=unit,quantity=quantity,actual_price=actual_price)
    )
WITH BROKER "broker_name_1" 
    ( 
      "username" = "hdfs", 
      "password" = "" 
    )
PROPERTIES
(
    "timeout"="1200",
    "max_filter_ratio"="0.1"
);

注意：

FORMAT AS "orc" : 这里我们指定了要导入的数据格式
SET : 这里我们定义了 Hive 表和 Doris 表之间的字段映射关系及字段转换的一些操作

HDFS文件系统数据导入

我们继续以上面创建好的 Doris 表为例，演示通过 Broker Load 从 HDFS 上导入数据。

导入作业的语句如下：

LOAD LABEL demo.label_20220402
        (
            INTO TABLE `ods_dish_detail_test`
            COLUMNS TERMINATED BY "\t"            (id,store_id,company_id,tower_id,commodity_id,commodity_name,commodity_price,member_price,cost_price,unit,quantity,actual_price)
        ) 
        with HDFS (
            "fs.defaultFS"="hdfs://10.220.147.151:8020",
            "hadoop.username"="root"
        )
        PROPERTIES
        (
            "timeout"="1200",
            "max_filter_ratio"="0.1"
        );

这里的具体参数可以参照：及 Broker Load 文档

我们可以通过下面的命令查看上面导入任务的状态信息，

具体的查看导入状态的语法参考 SHOW LOAD

mysql> show load order by createtime desc limit 1\G;
*************************** 1. row ***************************
         Label: broker_load_2022_03_23
         State: FINISHED
      Progress: ETL:100%; LOAD:100%
          Type: BROKER
       EtlInfo: unselected.rows=0; dpp.abnorm.ALL=0; dpp.norm.ALL=27
      TaskInfo: cluster:N/A; timeout(s):1200; max_filter_ratio:0.1
      ErrorMsg: NULL
    CreateTime: 2022-04-01 18:59:06
  EtlStartTime: 2022-04-01 18:59:11
 EtlFinishTime: 2022-04-01 18:59:11
 LoadStartTime: 2022-04-01 18:59:11
LoadFinishTime: 2022-04-01 18:59:11
           URL: NULL
    JobDetails: {"Unfinished backends":{"5072bde59b74b65-8d2c0ee5b029adc0":[]},"ScannedRows":27,"TaskNumber":1,"All backends":{"5072bde59b74b65-8d2c0ee5b029adc0":[36728051]},"FileNumber":1,"FileSize":5540}
1 row in set (0.01 sec)

取消导入

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

例如：撤销数据库 demo 上， label 为 broker_load_2022_03_23 的导入作业

CANCEL LOAD FROM demo WHERE LABEL = "broker_load_2022_03_23";

常见问题

导入报错：Scan bytes per broker scanner exceed limit:xxx

请参照文档中最佳实践部分，修改 FE 配置项 max_bytes_per_broker_scanner 和 max_broker_concurrency
导入报错：failed to send batch 或 TabletWriter add batch with unknown id

适当修改 query_timeout 和 streaming_load_rpc_max_alive_time_sec。

streaming_load_rpc_max_alive_time_sec：

在导入过程中，Doris 会为每一个 Tablet 开启一个 Writer，用于接收数据并写入。这个参数指定了 Writer 的等待超时时间。如果在这个时间内，Writer 没有收到任何数据，则 Writer 会被自动销毁。当系统处理速度较慢时，Writer 可能长时间接收不到下一批数据，导致导入报错：TabletWriter add batch with unknown id。此时可适当增大这个配置。默认为 600 秒
导入报错：

如果是PARQUET或者ORC格式的数据,需要再文件头的列名与doris表中的列名一致，如 :

代表获取在 parquet 或 orc 中以(tmp_c1, tmp_c2)为列名的列，映射到 doris 表中的(id, name)列。如果没有设置set, 则以column中的列作为映射。

注：如果使用某些 hive 版本直接生成的 orc 文件，orc 文件中的表头并非 hive meta 数据，而是（_col0, _col1, _col2, …）, 可能导致 Invalid Column Name 错误，那么则需要使用 set 进行映射

Broker Load

Broker Load

基本原理

开始导入

Hive 分区表导入(ORC格式)

HDFS文件系统数据导入

取消导入

相关系统配置

FE 配置

应用场景

作业调度

性能分析

常见问题

更多帮助