E-MapReduce:Broker Load

パーティション化された Hive テーブルからのデータインポート

この例では、デフォルトの CSV 形式を使用して、パーティション化された Hive テーブルからデータをインポートします。

1. day パーティションフィールドを持つ Hive テーブルを作成します。

   -- 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. Hive テーブルにデータをロードします。

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

3. ターゲット 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')))
   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. インポートジョブを送信します。SET 句は、Hive パーティションフィールド day を str_to_date を使用して Doris の日付列 rq にマッピングします。

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

ORC 形式のパーティション化された Hive テーブルからのデータインポート

この例では、Optimized Row Columnar (ORC) 形式で保存されている Hive テーブルからインポートします。CSV の例との主な違いは、FORMAT AS "orc" 句です。

1. ORC 形式でパーティション化された Hive テーブルを作成します。

   -- 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. 前の例と同じスキーマを使用して、ターゲット 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')))
   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. 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"
   );

HDFS からの直接データインポート

Hive テーブルを経由せずに HDFS からタブ区切りテキストファイルを直接インポートするには、WITH BROKER 句の代わりに WITH HDFS 句を使用します。

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 構成

fe.conf のこれらのパラメーターは、クラスター内のすべてのブローカーロードジョブに適用され、FE が BE 間で作業をどのように分散するかを制御します。

FE は、次の数式を使用して各インポートジョブの同時実行数を計算します。

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

Data per BE = Source file size / Concurrency

単一ジョブでインポート可能な最大データは、max_bytes_per_broker_scanner x Number of BEs と等しくなります。この制限を超えるデータをインポートするには、max_bytes_per_broker_scanner を増やします。

各ブローカーロードジョブは、1 つの保留中タスクと 1 つ以上のロード中タスクで構成されます。ロード中タスクの数は、LOAD LABEL ステートメントの DATA INFILE 句の数と等しくなります。

ブローカーパラメーター

ブローカーごとに、リモートストレージにアクセスするために異なる認証パラメーターが必要です。必要な接続プロパティについては、ブローカードキュメントをご参照ください。

ジョブごとのデータ量選択

以下のしきい値は、単一 BE クラスターを前提としています。マルチ BE クラスターの場合、BE の数で乗算します (例: 3 BE クラスターでは、3 GB 以下のしきい値が 9 GB 以下に引き上げられます)。

大容量ファイルの構成 (> 3 GB)

3 GB を超えるファイルの場合、ジョブを送信する前に次の手順に従います。

Step 1: fe.conf で同時実行数と BE ごとのデータ制限を設定します。

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

例: 100 GB ファイル、10 BE。

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

これらの設定により、すべての 10 BE がジョブを並行して処理し、それぞれ 10 GB を処理します。

Step 2: ジョブのタイムアウトを設定します。

クラスターのインポート速度を使用してタイムアウトを計算します。控えめな下限として 10 MB/秒 を使用します。

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

例: BE ごとに 10 GB、タイムアウト >= 1,000 秒。

ジョブを送信する際に、PROPERTIES ブロックでタイムアウトを設定します。

Step 3: 4 時間以上かかるファイルを分割します。

デフォルトの最大タイムアウトは 4 時間です。計算がこれを超過する場合でも、最大タイムアウトを増やさず、代わりにソースファイルを分割してください。4 時間以上実行された失敗したジョブは、再試行にも同じ時間がかかります。

次の数式を使用して、4 時間以内に収まる最大ファイルサイズを見つけます。

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

例: 10 BE、バッチあたりの最大値 = 14,400 秒 x 10 MB/秒 x 10 = 1,440 GB。

実際には、インポート速度が 10 MB/秒に達することはめったにありません。500 GB を超えるファイルは、より小さなバッチに分割してください。

ジョブスケジューリングの管理

desired_max_waiting_jobs パラメーターは、PENDING または LOADING 状態にあるジョブ数を同時に制限します (デフォルト: 100)。このしきい値を超えて送信された新しいジョブは拒否されます。

async_pending_load_task_pool_size パラメーター (デフォルト: 10) は、LOADING 状態にアクティブに入るジョブ数を制限します。たとえば、100 個のジョブを送信した場合、同時に実行されるのは 10 個のみで、残りは PENDING 状態で待機します。