E-MapReduce：Broker Load

查看Broker資訊

阿里雲EMR Serverless StarRocks執行個體在建立時已經自動搭建並啟動Broker服務。您可以使用以下SQL命令查看當前執行個體中所有Broker的詳細資料。

SHOW PROC "/brokers";

建立匯入任務

• 文法

  LOAD LABEL [<database_name>.]<label_name>
  (
      data_desc[, data_desc ...]
  )
  WITH BROKER
  (
      StorageCredentialParams
  )
  [PROPERTIES
  (
      opt_properties
  )
  ]

• 參數描述

  • <database_name>：可選，目標StarRocks表所在的資料庫。

  • <label_name>：匯入任務的標籤。

    每個匯入任務在該資料庫內均具有唯一的標籤。通過該標籤，可以查看相應匯入作業的執行情況，並防止匯入重複的資料。當匯入任務狀態為FINISHED時，該標籤不可再用於其他匯入作業；而當匯入任務狀態為CANCELLED時，該標籤可以被複用至其他匯入作業，但通常情況下是用於重試同一匯入作業（即使用相同標籤匯入相同資料），以實現資料“精確一次（Exactly-Once）”的語義。

  • data_desc：用於描述一批次待匯入的資料。

    Broker Load 支援一次匯入多個資料檔案。在一個匯入作業中，您可以使用多個 data_desc 來聲明匯入多個資料檔案，也可以使用一個 data_desc 來聲明匯入一個路徑下的所有資料檔案。Broker Load 還支援保證單次匯入事務的原子性，即單次匯入的多個資料檔案都成功或者都失敗，而不會出現部分匯入成功、部分匯入失敗的情況。

    DATA INFILE ("<file_path>"[, "<file_path>" ...])
    [NEGATIVE]
    INTO TABLE <table_name>
    [PARTITION (<partition1_name>[, <partition2_name> ...])]
    [TEMPORARY PARTITION (<temporary_partition1_name>[, <temporary_partition2_name> ...])]
    [COLUMNS TERMINATED BY "<column_separator>"]
    [ROWS TERMINATED BY "<row_separator>"]
    [FORMAT AS "CSV | Parquet | ORC"]
    [(format_type_options)]
    [(column_list)]
    [COLUMNS FROM PATH AS (<partition_field_name>[, <partition_field_name> ...])]
    [SET <k1=f1(v1)>[, <k2=f2(v2)> ...]]
    [WHERE predicate]

    data_desc 中部分參數描述如下表所示。

    參數	描述
    file_path	指定來源資料檔案所在的路徑。檔案路徑可以指定到檔案，也可以用星號（*）萬用字元指定某個目錄下的所有檔案。中間的目錄也可以使用萬用字元匹配。 可以使用的萬用字元有?、*、[]、{}和^，使用規則請參見FileSystem。 例如， 通過指定oss://bucket/data/tablename/*/*路徑可以匹配data/tablename下所有分區內的所有檔案。通過指定oss://bucket/data/tablename/dt=202104*/*路徑可以匹配data/tablename目錄下所有 202104 分區內的資料檔案。
    NEGATIVE	用於撤銷某一批已經成功匯入的資料。如果想要撤銷某一批已經成功匯入的資料，可以通過指定 NEGATIVE 關鍵字來匯入同一批資料。 說明 該參數僅適用於目標StarRocks表使用彙總表、並且所有Value列的彙總函式均為sum的情況。
    PARTITION	指定待匯入表的Partition資訊。 指定要把資料匯入哪些分區。如果不指定該參數，則預設匯入到StarRocks表所在的所有分區中。
    COLUMNS TERMINATED BY	用於指定匯入檔案中的資料行分隔符號。如果不指定該參數，則預設資料行分隔符號為\t，即Tab。 Broker Load通過MySQL協議提交匯入請求，除了StarRocks會做轉義處理以外，MySQL協議也會做轉義處理。因此，如果資料行分隔符號是Tab等不可見字元，則需要在列分隔字元前面多加一個反斜線（\）。例如，如果資料行分隔符號是 \t，這裡必須輸入 \\t；如果資料行分隔符號是 \n，這裡必須輸入 \\n。Apache Hive™ 檔案的資料行分隔符號為 \x01，因此，如果來源資料檔案是 Hive 檔案，這裡必須傳入 \\x01。
    FORMAT AS	用於指定匯入檔案的格式。取值包括 CSV、Parquet和ORC。如果不指定該參數，則預設通過file_path參數中指定的副檔名（.csv、.parquet和.orc）來判斷檔案格式。
    COLUMNS FROM PATH AS	用於從指定的檔案路徑中提取一個或多個分區欄位的資訊。該參數僅當指定的檔案路徑中存在分區欄位時有效。 例如，匯入檔案路徑為/path/col_name=col_value/file1，其中col_name可以對應到StarRocks表中的列。這時候，您可以設定參數為col_name。匯入時，StarRocks會將col_value落入col_name對應的列中。 說明 該參數只有在從HDFS匯入資料時可用。
    SET	用於將來源資料檔案的某一列依據指定的函數進行轉換，並將轉換後的結果插入到 StarRocks 表中。文法為column_name = expression。
    WHERE	用於指定過濾條件，對已完成轉換的資料進行篩選。只有符合WHERE子句中所指定的過濾條件的資料，才能匯入到StarRocks表中。

  • opt_properties：用於指定一些匯入相關的選擇性參數，指定的參數設定作用於整個匯入作業。文法如下所示。

    PROPERTIES ("<key1>" = "<value1>"[, "<key2>" = "<value2>" ...])

    部分參數描述如下表所示。

    參數	描述
    timeout	匯入作業的逾時時間（以秒為單位）。 您可以在opt_properties中自行設定每個匯入的逾時時間。匯入任務在設定的時限內未完成則會被系統取消，變為CANCELLED。Broker Load的預設匯入逾時時間為4小時。 重要 通常情況下，不需要您手動設定匯入任務的逾時時間。當在預設逾時時間內無法完成匯入時，可以手動設定任務的逾時時間。 推薦逾時時間的計算方式為：逾時時間 >（匯入檔案的總大小 x匯入檔案及相關物化視圖的個數）/平均匯入速度。 例如，要匯入一個1 GB的資料檔案，待匯入表包含2個Rollup表，當前StarRocks執行個體的平均匯入速度為10 MB/s。在這種情況下，根據公式計算出來時間長度為(1 x 1024 x 3)/10 = 307.2（秒） 因此，建議匯入作業的逾時時間大於308秒。 說明 由於每個StarRocks執行個體的機器環境不同，且執行個體並發的查詢任務也不同，所以StarRocks執行個體的最慢匯入速度需要您根據歷史的匯入任務速度進行推測。
    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 的總和就等於待匯入的總行數。
    load_mem_limit	匯入作業的記憶體限制，最大不超過BE或CN的記憶體限制。單位：位元組。
    strict_mode	是否開啟strict 模式。取值範圍： true： 表示開啟。 false（預設值）：表示關閉。 strict 模式對匯入處理程序中的列類型轉換實施了嚴格的過濾。strict 模式的過濾策略如下： 如果開啟strict 模式，StarRocks會把錯誤的資料行過濾掉，只匯入正確的資料行，並返回錯誤資料詳情。 如果關閉strict 模式，StarRocks會把轉換失敗的錯誤欄位轉換成NULL值，並把這些包含NULL值的錯誤資料行跟正確的資料行一起匯入。

查看匯入任務狀態

use <database_name>;
SHOW LOAD;

返回參數的描述如下表所示。

取消匯入任務

當匯入任務狀態不為CANCELLED或FINISHED時，可以通過CANCEL LOAD語句來取消該匯入任務。

CANCEL LOAD FROM <database_name> WHERE LABEL = "<label_name>";

匯入任務並發度

一個作業可以拆成一個或者多個任務，任務之間並存執行。拆分由LOAD語句中的DataDescription來決定。例如：

• 多個DataDescription對應匯入多個不同的表，每個會拆成一個任務。

• 多個DataDescription對應匯入同一個表的不同分區，每個也會拆成一個任務。

每個任務還會拆分成一個或者多個執行個體，然後將這些執行個體平均分配到BE上並存執行。執行個體的拆分由以下FE配置決定：

• min_bytes_per_broker_scanner：單個執行個體處理的最小資料量，預設值為64 MB。

• max_broker_concurrency：單個任務最大並發執行個體數，預設值為100。

• load_parallel_instance_num：單個BE上並發執行個體數，預設值為1個。

執行個體總數的計算公式為執行個體的總數 = min（匯入檔案總大小/單個執行個體處理的最小資料量，單個任務最大並發執行個體數，單個BE上並發執行個體數 * BE數）。

通常情況下，一個作業只有一個DataDescription，只會拆分成一個任務。任務會拆成與BE數相等的執行個體，然後分配到所有BE上並存執行。

從阿里雲OSS匯入

1. 建立測試表。

   create database if not exists load_test;
   use load_test;
   create table if not exists customer(
     c_customer_sk bigint,
     c_customer_id char(16),
     c_current_cdemo_sk bigint,
     c_current_hdemo_sk bigint,
     c_current_addr_sk bigint,
     c_first_shipto_date_sk bigint,
     c_first_sales_date_sk bigint,
     c_salutation char(10),
     c_first_name char(20),
     c_last_name char(30),
     c_preferred_cust_flag char(1),
     c_birth_day int,
     c_birth_month int,
     c_birth_year int,
     c_birth_country varchar(20),
     c_login char(13),
     c_email_address char(50),
     c_last_review_date_sk bigint
   )
   duplicate key (c_customer_sk)
   distributed by hash(c_customer_sk) buckets 5
   properties(
     "replication_num"="1"
   );

2. 建立匯入任務。

   請下載並上傳資料檔案customer.orc至OSS，然後執行以下命令建立匯入任務。

   LOAD LABEL load_test.customer_label
   (
     DATA INFILE("<file_path>")
     INTO TABLE customer
     format as "orc"
   )
   WITH BROKER 'broker'
   (
     "fs.oss.accessKeyId" = "xxxxx",
     "fs.oss.accessKeySecret" = "xxxxx",
     "fs.oss.endpoint" = "oss-cn-xxx-internal.aliyuncs.com"
   );

   涉及參數如下表所示，請根據實際情況進行替換。

   參數	說明
   <file_path>	customer.orc檔案所在的路徑。例如，oss://<yourBucketName>/data/customer.orc。
   fs.oss.accessKeyId	阿里雲帳號或RAM使用者的AccessKey ID。您可以進入AccessKey管理頁面擷取AccessKey ID。
   fs.oss.accessKeySecret	AccessKey ID對應的 AccessKey Secret。
   fs.oss.endpoint	訪問OSS的Endpoint。例如，oss-cn-hangzhou-internal.aliyuncs.com。 如果StarRocks與OSS位於同一地區，則使用VPC網路Endpoint，否則使用公網Endpoint。擷取方法請參見OSS地區和訪問網域名稱。

3. 查看匯入任務狀態。

   use load_test;
   show load where label='customer_label';

4. 查詢表資訊。

   • 樣本1：統計load_test資料庫中customer表的總行數。

     select count(1) from load_test.customer;

   • 樣本2：展示load_test資料庫中customer表的前兩條完整記錄。

     select * from load_test.customer limit 2;

從HDFS匯入

"dfs.nameservices" = "ha_cluster",
"dfs.ha.namenodes.ha_cluster" = "ha_n1,ha_n2",
"dfs.namenode.rpc-address.ha_cluster.ha_n1" = "<hdfs_host>:<hdfs_port>",
"dfs.namenode.rpc-address.ha_cluster.ha_n2" = "<hdfs_host>:<hdfs_port>",
"dfs.client.failover.proxy.provider.ha_cluster" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"