E-MapReduce:Broker Load

Import data from a partitioned Hive table

This example imports data from a partitioned Hive table using the default CSV format.

1. Create a Hive table with a day partition field.

   -- Default data format (CSV), partition field: day
   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'

2. Load data into the Hive table.

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

3. Create the target Doris table.

   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')))
   DISTRIBUTED BY HASH(`store_id`) BUCKETS 1
   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"
   );

4. Submit the import job. The SET clause maps the Hive partition field day to the Doris date column rq using str_to_date.

   LOAD LABEL broker_load_2022_03_23
   (
       DATA INFILE("hdfs://192.168.**.**:8020/user/hive/warehouse/ods.db/ods_demo_detail/*/*")
       INTO TABLE doris_ods_test_detail
       COLUMNS TERMINATED BY ","
     (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"
   );

Import data from a partitioned Hive table in ORC format

This example imports from a Hive table stored in Optimized Row Columnar (ORC) format. The key difference from the CSV example is the FORMAT AS "orc" clause.

1. Create a partitioned Hive table in ORC format.

   -- Data format: ORC, partition field: day
   CREATE TABLE `ods_demo_orc_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'
   STORED AS ORC

2. Create the target Doris table using the same schema as in the previous example.

   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')))
   DISTRIBUTED BY HASH(`store_id`) BUCKETS 1
   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"
   );

3. Submit the import job with FORMAT AS "orc".

   LOAD LABEL dish_2022_03_23
   (
       DATA INFILE("hdfs://10.220.**.**: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"
   );

Import data directly from HDFS

To import a tab-delimited text file from HDFS without going through a Hive table, use the WITH HDFS clause instead of WITH BROKER.

LOAD LABEL demo.label_20220402
        (
            DATA INFILE("hdfs://10.220.**.**:8020/tmp/test_hdfs.txt")
            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.**.**:8020",
            "hadoop.username"="root"
        )
        PROPERTIES
        (
            "timeout"="1200",
            "max_filter_ratio"="0.1"
        );

FE configurations

These parameters in fe.conf apply to all Broker Load jobs in the cluster and control how the FE distributes work across BEs.

The FE calculates concurrency for each import job using these formulas:

Concurrency = Math.min(
    Source file size / min_bytes_per_broker_scanner,
    max_broker_concurrency,
    Number of BEs
)

Data per BE = Source file size / Concurrency

The maximum data importable in a single job equals max_bytes_per_broker_scanner x Number of BEs. To import more data than this limit, increase max_bytes_per_broker_scanner.

Each Broker Load job consists of one pending task and one or more loading tasks. The number of loading tasks equals the number of DATA INFILE clauses in the LOAD LABEL statement.

Broker parameters

Different brokers require different authentication parameters to access remote storage. Refer to your broker documentation for the required connection properties.

Choosing data volume per job

The following thresholds assume a single-BE cluster. For multi-BE clusters, multiply by the number of BEs (for example, a 3-BE cluster raises the <= 3 GB threshold to <= 9 GB).

Configuring for large files (> 3 GB)

For a file that exceeds 3 GB, follow these steps before submitting the job.

Step 1: Set concurrency and per-BE data limit in `fe.conf`.

max_broker_concurrency = <number of BEs>
max_bytes_per_broker_scanner >= <source file size> / max_broker_concurrency

Example: 100 GB file, 10 BEs.

max_broker_concurrency = 10
max_bytes_per_broker_scanner >= 10 GB   (100 GB / 10)

With these settings, all 10 BEs process the job in parallel, each handling 10 GB.

Step 2: Set the job timeout.

Calculate timeout using the import speed of your cluster. Use 10 MB/s as a conservative lower bound.

Data per BE / Slowest import speed of your cluster (MB/s) >= Timeout >= Data per BE / 10 MB/s

Example: 10 GB per BE, timeout >= 1,000 s.

Set the timeout in the PROPERTIES block when submitting the job.

Step 3: Split files that require more than 4 hours.

The default maximum timeout is 4 hours. If your calculation exceeds this, do not increase the maximum timeout—split the source file instead. A failed job that ran for more than 4 hours takes equally long to retry.

Use this formula to find the largest file size that fits within 4 hours:

Max data per batch = 14,400 s x 10 MB/s x Number of BEs

Example: 10 BEs, max per batch = 14,400 s x 10 MB/s x 10 = 1,440 GB.

In practice, import speed rarely reaches 10 MB/s. Split any file larger than 500 GB into smaller batches.

Managing job scheduling

The desired_max_waiting_jobs parameter limits how many jobs can be in PENDING or LOADING state at once (default: 100). New jobs submitted beyond this threshold are rejected.

The async_pending_load_task_pool_size parameter (default: 10) caps the number of jobs that actively enter LOADING state. For example, if you submit 100 jobs, only 10 run concurrently—the rest wait in PENDING state.