E-MapReduce:Spark Load

Terms

• Spark ETL: Spark ETL is used to extract, transform, and load (ETL) data to be imported. You can use Spark ETL to perform operations such as creating bitmap global dictionaries, data partitioning, data sorting, and data aggregation.
• Broker: A broker is an independent stateless process that encapsulates file system interfaces. This allows StarRocks clusters to read files from remote storage systems.
• Global dictionary: A global dictionary stores the data structures of raw data and mapped encoded data. The raw data can be of all types, whereas the encoded data must be integers. Global dictionaries are usually used for high-accuracy deduplication in precalculation.

Workflow

After an import job of the Spark type is submitted from the MySQL client, the frontend node records metadata and returns a message to indicate that the job is successfully submitted.

The following figure shows the workflow of Spark Load.

Global dictionary

Scenario

Bitmap columns in StarRocks are realized by using roaring bitmaps. The data imported by using roaring bitmaps must be integers. To precalculate bitmap columns before you import data, convert the data type into Integer. During data import, a global dictionary stores the data structures of raw data and mapped encoded data based on a Hive table.

Workflow

1. Read data from a data source and store the data in a temporary Hive table named hive-table.
2. Deduplicate data in the hive-table table and store the deduplicated raw data in a Hive table named distinct-value-table.
3. Create a global dictionary table named dict-table to store raw data in a column and encoded data in another column.
4. Perform a LEFT JOIN operation on the distinct-value-table and dict-table tables. Use a window function to encode the raw data of the LEFT JOIN results, and write the columns of raw data and encoded data back to the dict-table table.
5. Perform a JOIN operation on the dict-table and hive-table tables. Replace the raw data in the hive-table table with the encoded integers.
6. Data in the hive-table table is read and calculated in the subsequent operations, and then imported to a StarRocks cluster.

Data preprocessing

Basic operations

Configure clusters for ETL jobs

Spark provides external resources for you to run ETL jobs in StarRocks clusters. Other types of external resources will be supported in StarRocks clusters in the future. For example, Spark or GPU can be used for data queries, HDFS or Amazon Simple Storage Service (S3) can be used to store external data, and MapReduce can be used for ETL. You can use Resource Management to manage these external resources.

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

• Create a resource

  resource-name is the property that specifies the name of the Spark resource configured for the StarRocks clusters.

  PROPERTIES includes a number of properties of the Spark resource. The following table describes the properties. For more information, see Spark Configuration of the Spark documentation.

  Property		Description
  type		The type of the resource. Set the value to spark. This property is required.
  Spark-related properties	spark.master	Set the value to yarn. This property is required.
  	spark.submit.deployMode	The mode in which the Spark program is deployed. Valid values: cluster and client. This property is required.
  	spark.hadoop.fs.defaultFS	This property is required only if the spark.master property is set to yarn.
  	spark.hadoop.yarn.resourcemanager.address	The address of a ResourceManager.
  	spark.hadoop.yarn.resourcemanager.ha.enabled	Specifies whether to enable high availability on the ResourceManager. Default value: true.
  	spark.hadoop.yarn.resourcemanager.ha.rm-ids	The logical IDs of the ResourceManager.
  	spark.hadoop.yarn.resourcemanager.hostname.rm-id	The hostname corresponds to each logical ID. Note You need to configure either the spark.hadoop.yarn.resourcemanager.hostname.rm-id property or the spark.hadoop.yarn.resourcemanager.address.rm-id property for a high-availability ResourceManager.
  	spark.hadoop.yarn.resourcemanager.address.rm-id	The address corresponds to each logical ID to which the client submits a job. The address is in the host:port format. Note You need to configure either the spark.hadoop.yarn.resourcemanager.hostname.rm-id property or the spark.hadoop.yarn.resourcemanager.address.rm-id property for a high-availability ResourceManager.
  working_dir		The directory in which the resource for ETL resides. Note This property is required if the Spark resource is used for ETL. Example: hdfs://host:port/tmp/starrocks.
  broker		The name of the broker. Note This property is required if the Spark resource is used for ETL. You must run the ALTER SYSTEM ADD BROKER command to configure the broker in advance.
  broker.property_key		The information required for verification when the broker reads the intermediate file of ETL.

  Sample code:

  -- yarn cluster mode
  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 cluster mode that enables high availability
  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 cluster mode that enables high availability
  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"
  );

• View resources

  You can view resources only on which you have the USAGE-PRIV permission by using an ordinary account. You can view all resources by using the root or admin account.

• Manage permissions on resources
  You can run the GRANT REVOKE command to manage permissions on resources. Only the USAGE-PRIV permission is supported. You can grant the USAGE-PRIV permission to a user or role. The role can be assigned as usual. Sample code:

  -- Grant the USAGE-PRIV permission on the resource spark0 to the user user0:
  GRANT USAGE_PRIV ON RESOURCE "spark0" TO "user0"@"%";
  
  -- Grant the USAGE-PRIV permission on the resource spark0 to the role role0:
  GRANT USAGE_PRIV ON RESOURCE "spark0" TO ROLE "role0";
  
  -- Grant the USAGE-PRIV permission on all resources to the user user0:
  GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";
  
  -- Grant the USAGE-PRIV permission on all resources to role role0:
  GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";
  
  -- Revoke the USAGE-PRIV permission on the resource spark0 from the user user0:
  REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM "user0"@"%";

Configure the Spark client

By default, the Spark dependency package is named spark-2x.zip. The frontend node also uploads the Dynamic Partition Pruning (DPP) dependency package to the remote warehouse. If Spark Load submits all dependency packages to the remote warehouse, the frontend node does not need to upload the two dependency packages, which saves time.

Configure the YARN client

Create an import job

• Syntax:

  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, ...)

  For more information about the syntax, run the HELP SPARK LOAD command. The following list describes the parameters in the syntax:

  • Label

    The label of the import job. The label is unique in a database. Specifications of the parameter are the same as those of the same parameter in the Broker Load topic.

  • Parameters on data description

    Only CSV files and Hive tables are supported data sources. Other specifications of the parameters are the same as those of the same parameters in the Broker Load topic.

  • Parameters on job properties

    The properties of the import job. The parameters are in the opt_properties section. Specifications of the parameters are the same as those of the parameters in the broker_properties section in the Broker Load topic.

  • Parameters on Spark resources
    Parameters on Spark resources must be configured in the StarRocks cluster in advance. Spark Load can be used only after the user is granted the USAGE-PRIV permission. In the following sample code, two parameters on Spark resources are configured. You can configure the parameters if you have temporary requirements, such as adding resources for the job. This configuration takes effect only for this job and does not affect the configurations of the StarRocks cluster. Sample code:

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

  • Import data from a Hive table

    To import data from a Hive table, create an external Hive table and specify the table name when you submit the import command.

  • Create a global dictionary during data import
    You can create a global dictionary if the aggregated column of the StarRocks table is the bitmap column. To create a global dictionary, specify the field for which you want to create a global dictionary in the LOAD command in the following format: Field name=bitmap_dict (Field name of the Hive table).

    Important You can create a global dictionary only if you import data from a Hive table.
• Sample code:
  • Create an import job to import data from an HDFS file:

    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, ...)

  • Create an import job to import data from a Hive table:
    1. Create a Hive resource.

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

    2. Create an external Hive table.

       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. Submit the LOAD command to ensure that the column in the StarRocks table to which the data is to be imported also resides in the external Hive table.

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

View the import job

Import jobs submitted by using Spark Load or Broker Load are asynchronous import jobs. You must use the job labels in the SHOW LOAD command to view the import jobs. The command is commonly used to view import jobs submitted in all methods. For more information, run the HELP SHOW LOAD command.

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.1.1:8089/proxy/application_1586619723848_0035/
 JobDetails: {"ScannedRows":28133395,"TaskNumber":1,"FileNumber":1,"FileSize":200000}

LOAD progress = Number of imported tablets in all replicas/Total number of tablets in the import job × 100%

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

For more information about the parameters in the returned result, see Broker Load.

View the logs

When the Spark Load job is submitted, logs are generated. By default, the logs are stored in the log/spark_launcher_log directory under the root directory of the frontend node and named in the spark-launcher-{load-job-id}-{label}.log format. The log is temporarily retained before it is deleted along with the import information in the metadata of the frontend node. By default, the log is retained for three days.

Cancel the import job

If the import job is not in the CANCELLED or FINISHED state, you can cancel the import job as required. When you cancel an import job, specify the label of the job. For more information about the syntax, run the HELP CANCEL LOAD command.

Related system configurations

Best practice

Spark Load is most suitable for scenarios in which tens of GB or TB of data is to be imported from HDFS files. If you import a small amount of data, we recommend that you use Stream Load or Broker Load.

For information about complete sample code, see 03_sparkLoad2StarRocks.md at GitHub.