E-MapReduce：Spark Load

基本概念

• Spark ETL：在匯入流程中主要負責資料的ETL工作，包括全域字典構建（BITMAP類型）、分區、排序和彙總等。

• Broker：是一個獨立的無狀態進程。封裝了檔案系統介面，提供StarRocks讀取遠端儲存系統中檔案的能力。

• 全域字典：儲存了資料從原始值到編碼值對應的資料結構，原始值可以是任意資料類型，而編碼後的值為整型。全域字典主要應用於精確去重預計算的情境。

基本原理

使用者通過MySQL用戶端提交Spark類型匯入任務，FE記錄中繼資料並返回使用者提交成功。

全域字典

適用情境

目前StarRocks中BITMAP列是使用類庫Roaringbitmap實現的，而Roaringbitmap的輸入資料類型只能是整型，因此如果要在匯入流程中實現對於BITMAP列的預計算，則需要將輸入資料的類型轉換成整型。在StarRocks現有的匯入流程中，全域字典的資料結構是基於Hive表實現的，儲存了原始值到編碼值的映射。

構建流程

1. 讀取上遊資料來源的資料，產生一張Hive暫存資料表，記為hive-table。

2. 從hive-table中抽取待去重欄位的去重值，產生一張新的Hive表，記為distinct-value-table。

3. 建立一張全域字典表，記為dict-table。一列為原始值，一列為編碼後的值。

4. 將distinct-value-table與dict-table進行LEFT JOIN，計算出新增的去重值集合，然後對這個集合使用視窗函數進行編碼，此時去重列原始值就多了一列編碼後的值，最後將這兩列的資料寫回dict-table。

5. 將dict-table與hive-table進行JOIN，完成hive-table中原始值替換成整型編碼值的工作。

6. hive-table會被下一步資料預先處理的流程所讀取，經過計算後匯入到StarRocks中。

資料預先處理

基本操作

配置ETL叢集

Spark作為一種外部計算資源在StarRocks中用來完成ETL工作，未來可能還有其他的外部資源會加入到StarRocks中使用。例如，Spark或GPU用於查詢，HDFS或S3用於外部儲存，MapReduce用於ETL等，因此引入Resource Management來管理StarRocks使用的這些外部資源。

-- create spark resource
CREATE EXTERNAL RESOURCE resource_name
PROPERTIES
(
 type = spark,
 spark_conf_key = spark_conf_value,
 working_dir = path,
 broker = broker_name,
 broker.property_key = property_value
);

-- drop spark resource
DROP RESOURCE resource_name;

-- show resources
SHOW RESOURCES
SHOW PROC "/resources";

-- privileges
GRANT USAGE_PRIV ON RESOURCE resource_name TO user_identityGRANT USAGE_PRIV ON RESOURCE resource_name TO ROLE role_name;
REVOKE USAGE_PRIV ON RESOURCE resource_name FROM user_identityREVOKE USAGE_PRIV ON RESOURCE resource_name FROM ROLE role_name;

• 建立資源

  resource-name為StarRocks中配置的Spark資源的名字。

  PROPERTIES是Spark資源的相關參數，參數描述如下表所示，更多參數描述請參見Spark Configuration。

  參數		描述
  type		資源類型。必填參數，目前僅支援Spark。
  Spark相關參數	spark.master	必填參數，目前支援yarn。
  	spark.submit.deployMode	Spark程式的部署模式。必填參數，支援cluster和client兩種。
  	spark.hadoop.fs.defaultFS	Master為YARN時必填。
  	spark.hadoop.yarn.resourcemanager.address	單點Resource Manager地址。
  	spark.hadoop.yarn.resourcemanager.ha.enabled	Resource Manager啟用HA。預設值為true。
  	spark.hadoop.yarn.resourcemanager.ha.rm-ids	Resource Manager邏輯ID列表。
  	spark.hadoop.yarn.resourcemanager.hostname.rm-id	對於每個rm-id，指定Resource Manager對應的主機名稱。 說明 HA Resource Manager只需配置spark.hadoop.yarn.resourcemanager.hostname.rm-id或spark.hadoop.yarn.resourcemanager.address.rm-id中的任意一個。
  	spark.hadoop.yarn.resourcemanager.address.rm-id	對於每個rm-id，指定host:port以供用戶端提交作業。 說明 HA Resource Manager只需配置spark.hadoop.yarn.resourcemanager.hostname.rm-id或spark.hadoop.yarn.resourcemanager.address.rm-id中的任意一個。
  working_dir		ETL使用的目錄。 說明 Spark作為ETL資源使用時必填。例如，hdfs://host:port/tmp/starrocks。
  broker		Broker名字。 說明 Spark作為ETL資源使用時必填。需要使用ALTER SYSTEM ADD BROKER命令提前完成配置。
  broker.property_key		Broker讀取ETL產生中間檔案時需要指定的認證資訊等。

  建立資源樣本如下所示。

  -- yarn cluster模式
  CREATE EXTERNAL RESOURCE "spark0"
  PROPERTIES
  (
      "type" = "spark",
      "spark.master" = "yarn",
      "spark.submit.deployMode" = "cluster",
      "spark.jars" = "xxx.jar,yyy.jar",
      "spark.files" = "/tmp/aaa,/tmp/bbb",
      "spark.executor.memory" = "1g",
      "spark.yarn.queue" = "queue0",
      "spark.hadoop.yarn.resourcemanager.address" = "resourcemanager_host:8032",
      "spark.hadoop.fs.defaultFS" = "hdfs://namenode_host:9000",
      "working_dir" = "hdfs://namenode_host:9000/tmp/starrocks",
      "broker" = "broker0",
      "broker.username" = "user0",
      "broker.password" = "password0"
  );
  
  -- yarn HA cluster模式
  CREATE EXTERNAL RESOURCE "spark1"
  PROPERTIES
  (
      "type" = "spark",
      "spark.master" = "yarn",
      "spark.submit.deployMode" = "cluster",
      "spark.hadoop.yarn.resourcemanager.ha.enabled" = "true",
      "spark.hadoop.yarn.resourcemanager.ha.rm-ids" = "rm1,rm2",
      "spark.hadoop.yarn.resourcemanager.hostname.rm1" = "host1",
      "spark.hadoop.yarn.resourcemanager.hostname.rm2" = "host2",
      "spark.hadoop.fs.defaultFS" = "hdfs://namenode_host:9000",
      "working_dir" = "hdfs://namenode_host:9000/tmp/starrocks",
      "broker" = "broker1"
  );
  
  -- HDFS HA cluster模式
  CREATE EXTERNAL RESOURCE "spark2"
  PROPERTIES
  (
      "type" = "spark",
      "spark.master" = "yarn",
      "spark.hadoop.yarn.resourcemanager.address" = "resourcemanager_host:8032",
      "spark.hadoop.fs.defaultFS" = "hdfs://myha",
      "spark.hadoop.dfs.nameservices" = "myha",
      "spark.hadoop.dfs.ha.namenodes.myha" = "mynamenode1,mynamenode2",
      "spark.hadoop.dfs.namenode.rpc-address.myha.mynamenode1" = "nn1_host:rpc_port",
      "spark.hadoop.dfs.namenode.rpc-address.myha.mynamenode2" = "nn2_host:rpc_port",
      "spark.hadoop.dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
      "working_dir" = "hdfs://myha/tmp/starrocks",
      "broker" = "broker2",
      "broker.dfs.nameservices" = "myha",
      "broker.dfs.ha.namenodes.myha" = "mynamenode1,mynamenode2",
      "broker.dfs.namenode.rpc-address.myha.mynamenode1" = "nn1_host:rpc_port",
      "broker.dfs.namenode.rpc-address.myha.mynamenode2" = "nn2_host:rpc_port",
      "broker.dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
  );

• 查看資源

  show resources;

  普通賬戶只能看到自己有USAGE-PRIV使用許可權的資源。root和admin賬戶可以看到所有的資源。

• 資源許可權

  資源許可權通過GRANT REVOKE來管理，目前僅支援USAGE-PRIV使用許可權。您可以將USAGE-PRIV許可權賦予某個使用者或者某個角色，角色的使用與之前一致。樣本如下。

  -- 授予spark0資源的使用許可權給使用者user0
  GRANT USAGE_PRIV ON RESOURCE "spark0" TO "user0"@"%";
  
  -- 授予spark0資源的使用許可權給角色role0
  GRANT USAGE_PRIV ON RESOURCE "spark0" TO ROLE "role0";
  
  -- 授予所有資源的使用許可權給使用者user0
  GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";
  
  -- 授予所有資源的使用許可權給角色role0
  GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";
  
  -- 撤銷使用者user0的spark0資源使用許可權
  REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM "user0"@"%";

配置Spark用戶端

除了Spark依賴（預設以spark-2x.zip命名），FE還會上傳DPP的依賴包至遠端倉庫。如果此次Spark Load提交的所有依賴檔案都已存在遠端倉庫，則不需要再上傳依賴，節省下了每次重複上傳大量檔案的時間。

配置YARN用戶端

建立匯入任務

• 建立文法

  LOAD LABEL load_label
      (data_desc, ...)
  WITH RESOURCE resource_name
  [resource_properties]
  [PROPERTIES (key1=value1, ... )]
  
  * load_label:
      db_name.label_name
  
  * data_desc:
      DATA INFILE ('file_path', ...)
      [NEGATIVE]
      INTO TABLE tbl_name
      [PARTITION (p1, p2)]
      [COLUMNS TERMINATED BY separator ]
      [(col1, ...)]
      [COLUMNS FROM PATH AS (col2, ...)]
      [SET (k1=f1(xx), k2=f2(xx))]
      [WHERE predicate]
  
      DATA FROM TABLE hive_external_tbl
      [NEGATIVE]
      INTO TABLE tbl_name
      [PARTITION (p1, p2)]
      [SET (k1=f1(xx), k2=f2(xx))]
      [WHERE predicate]
  
  * resource_properties:
   (key2=value2, ...)

  建立匯入的詳細文法可以執行HELP SPARK LOAD命令查看協助。Spark Load的建立匯入文法中參數意義如下：

  • Label

    匯入任務的標識。每個匯入任務，都有一個在單DataBase內部唯一的Label。具體規則與Broker Load一致。

  • 資料描述類參數

    目前支援的資料來源有CSV和Hive table。其他規則與Broker Load一致。

  • 匯入作業參數

    匯入作業參數主要指的是Spark Load建立匯入語句中的屬於opt_properties部分的參數。匯入作業參數是作用於整個匯入作業的。規則與Broker Load一致。

  • Spark資源參數

    Spark資源需要提前配置到StarRocks系統中並且賦予使用者USAGE-PRIV許可權後才能使用Spark Load。當您有臨時性的需求，例如增加任務使用的資源而修改Spark configs時，可以設定以下參數，設定僅對本次任務生效，並不影響StarRocks叢集中已有的配置。

    WITH RESOURCE 'spark0'
    (
       "spark.driver.memory" = "1g",
       "spark.executor.memory" = "3g"
    )

  • 資料來源為Hive表時的匯入

    如果期望在匯入流程中將Hive表作為資料來源，則需要先建立一張類型為Hive的外部表格，然後提交匯入命令時指定外部表格的表名即可。

  • 匯入流程構建全域字典

    適用於StarRocks表彙總列的資料類型為BITMAP類型。在Load命令中指定需要構建全域字典的欄位即可，格式為StarRocks欄位名稱=bitmap_dict（hive表欄位名稱）。

    重要

    目前只有在上遊資料來源為Hive表時才支援全域字典的構建。

• 樣本：

  • 上遊資料來源為HDFS檔案時建立匯入任務的情況

    LOAD LABEL db1.label1 
    (
        DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/starRocks/test/ml/file1")
        INTO TABLE tbl1
        COLUMNS TERMINATED BY ","
        (tmp_c1,tmp_c2)
        SET
        (
            id=tmp_c2,
            name=tmp_c1
        ),
        DATA INFILE("hdfs://emr-header-1.cluster-xxx:9000/user/starRocks/test/ml/file2")
        INTO TABLE tbl2
        COLUMNS TERMINATED BY ","
        (col1, col2)
        where col1 > 1
    )
    WITH RESOURCE 'spark0'
    (
        "spark.executor.memory" = "2g",
        "spark.shuffle.compress" = "true"
    )
    PROPERTIES
    (
        "timeout" = "3600"
    );

  • 上遊資料來源是Hive表時建立匯入任務的情況

    1. 建立Hive資源。

       CREATE EXTERNAL RESOURCE hive0
       properties
       (
           "type" = "hive",
           "hive.metastore.uris" = "thrift://emr-header-1.cluster-xxx:9083"
       );

    2. 建立Hive外部表格。

       CREATE EXTERNAL TABLE hive_t1
       (
           k1 INT,
           K2 SMALLINT,
           k3 varchar(50),
           uuid varchar(100)
       )
       ENGINE=hive
       properties
       (
           "resource" = "hive0",
           "database" = "tmp",
           "table" = "t1"
       );

    3. 提交load命令，要求匯入的StarRocks表中的列必須在Hive外部表格中存在。

       LOAD LABEL db1.label1
       (
           DATA FROM TABLE hive_t1
           INTO TABLE tbl1
           SET
           (
               uuid=bitmap_dict(uuid)
           )
       )
       WITH RESOURCE 'spark0'
       (
           "spark.executor.memory" = "2g",
           "spark.shuffle.compress" = "true"
       )
       PROPERTIES
       (
           "timeout" = "3600"
       );

查看匯入任務

Spark Load和Broker Load都是非同步匯入方式。您必須將建立匯入的Label記錄下來，並且在SHOW LOAD命令中使用此Label來查看匯入結果。查看匯入的命令在所有匯入方式中是通用的，具體文法可執行HELP SHOW LOAD命令查看。

show load order by createtime desc limit 1\G

 *************************** 1. row ***************************
  JobId: 76391
  Label: label1
  State: FINISHED
 Progress: ETL:100%; LOAD:100%
  Type: SPARK
 EtlInfo: unselected.rows=4; dpp.abnorm.ALL=15; dpp.norm.ALL=28133376
 TaskInfo: cluster:cluster0; 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:49:44
 LoadStartTime: 2019-07-27 11:49:44
LoadFinishTime: 2019-07-27 11:50:16
  URL: http://1.1.*.*:8089/proxy/application_1586619723848_0035/
 JobDetails: {"ScannedRows":28133395,"TaskNumber":1,"FileNumber":1,"FileSize":200000}

LOAD進度 = 當前已完成所有replica匯入的tablet個數 / 本次匯入任務的總tablet個數* 100%

 {"ScannedRows":139264,"TaskNumber":1,"FileNumber":1,"FileSize":940754064}

其餘返回結果集中參數含義可以參見Broker Load，詳情請參見Broker Load。

查看Spark Launcher提交日誌

Spark任務提交過程中產生的詳細日誌，日誌預設儲存在FE根目錄下log/spark_launcher_log路徑下，並以spark-launcher-{load-job-id}-{label}.log格式命名，日誌會在此目錄下儲存一段時間，當FE中繼資料中的匯入資訊被清理時，相應的日誌也會被清理，預設儲存時間為3天。

取消匯入任務

當Spark Load作業狀態不為CANCELLED或FINISHED時，您可以手動取消。取消時需要指定待取消匯入任務的Label。取消匯入命令文法可以執行HELP CANCEL LOAD命令查看。

相關係統配置

最佳實務

使用Spark Load最適合的情境是未經處理資料在檔案系統（HDFS）中，資料量在幾十GB到TB層級。小資料量還是建議使用Stream Load或者Broker Load。

完整Spark Load匯入樣本，請參見03_sparkLoad2StarRocks.md。