E-MapReduce:Spark Load

Step 1: Create an external Spark resource

Register the Spark cluster as an external resource in StarRocks so that the frontend node can submit ETL jobs to it.

Syntax

CREATE EXTERNAL RESOURCE "resource_name"
PROPERTIES
(
    "type" = "spark",
    "spark.master" = "yarn",
    "spark.submit.deployMode" = "<cluster|client>",
    "spark.hadoop.fs.defaultFS" = "hdfs://<namenode-host>:<port>",
    "spark.hadoop.yarn.resourcemanager.address" = "<resourcemanager-host>:8032",
    "working_dir" = "hdfs://<namenode-host>:<port>/tmp/starrocks",
    "broker" = "<broker-name>",
    "broker.username" = "<username>",
    "broker.password" = "<password>"
);

Resource properties

For a full list of Spark configuration options, see the Spark Configuration documentation.

Examples

Yarn cluster mode (standard):

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 with YARN 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 high-availability mode:

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

Manage resources

-- List all resources
SHOW RESOURCES;
SHOW PROC "/resources";

-- Grant or revoke USAGE-PRIV
GRANT USAGE_PRIV ON RESOURCE "spark0" TO "user0"@"%";
GRANT USAGE_PRIV ON RESOURCE "spark0" TO ROLE "role0";
GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";
GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";
REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM "user0"@"%";
REVOKE USAGE_PRIV ON RESOURCE "spark0" FROM ROLE "role0";

With an ordinary account, SHOW RESOURCES only displays resources for which the account has USAGE-PRIV. Root and admin accounts see all resources.

Drop a resource

DROP RESOURCE "resource_name";

Step 2: Configure the Spark client

The frontend node uses spark-submit to submit Spark Load jobs, so the Spark client must be installed on the frontend node host.

Download Spark 2.4.5 or a later 2.x version and place it on the frontend node. Then configure the following parameters in the frontend node's configuration file (fe.conf):

1. Set the Spark home directory Set spark_home_default_dir to the directory where you placed the Spark client. The default value is lib/spark2x under the frontend node root directory. This parameter cannot be left blank.

2. Package Spark dependencies Package all JAR files from the Spark client's jars/ folder into a ZIP file. Set spark_resource_path to the path of that ZIP file. If left blank, the frontend node looks for lib/spark2x/jars/spark-2x.zip in its root directory—if not found, an error is returned. When a Spark Load job is submitted, the dependency package is uploaded to a remote staging location at working_dir/{cluster_id}/. The staging location is named in the --spark-repository--{resource-name} format. Example directory structure:

   ---spark-repository--spark0/
      |---archive-1.0.0/
      |   |---lib-990325d2c0d1d5e45bf675e54e44fb16-spark-dpp-1.0.0-jar-with-dependencies.jar
      |   |---lib-7670c29daf535efe3c9b923f778f61fc-spark-2x.zip
      |---archive-1.1.0/
      |   |---lib-64d5696f99c379af2bee28c1c84271d5-spark-dpp-1.1.0-jar-with-dependencies.jar
      |   |---lib-1bbb74bb6b264a270bc7fca3e964160f-spark-2x.zip

   The default dependency package name is spark-2x.zip. The frontend node also uploads the Dynamic Partition Pruning (DPP) dependency. Once uploaded, these packages are reused for subsequent jobs.

Step 3: Configure the YARN client

The frontend node runs YARN commands to check application status and stop applications, so the YARN client must also be installed on the frontend node.

Download Hadoop 2.5.2 or a later 2.x version and configure the following parameters in fe.conf:

1. Set the YARN executable path Set yarn_client_path to the path of the YARN binary file. The default is lib/yarn-client/hadoop/bin/yarn under the frontend node root directory.

2. (Optional) Set the YARN config directory When the frontend node checks application status or stops applications, it generates core-site.xml and yarn-site.xml in lib/yarn-config/ by default. To change this location, set yarn_config_dir.

Create an import job

Supported data sources: CSV files and Hive tables.

LOAD LABEL db_name.label_name
    (data_desc, ...)
WITH RESOURCE resource_name
[resource_properties]
[PROPERTIES (key1=value1, ...)]

Where data_desc is one of:

-- From HDFS files
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]

-- From a Hive table
DATA FROM TABLE hive_external_tbl
[NEGATIVE]
INTO TABLE tbl_name
[PARTITION (p1, p2)]
[SET (k1=f1(xx), k2=f2(xx))]
[WHERE predicate]

For the full syntax, run HELP SPARK LOAD.

Key parameters

Example 1: Import from HDFS CSV files

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

Example 2: Import from a Hive table with a global dictionary

Use this approach when the target StarRocks table has a bitmap aggregation column. The bitmap_dict() function maps raw Hive column values to encoded integers via the global dictionary.

1. Create a Hive external 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 import job:

   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 an import job

SHOW LOAD ORDER BY createtime DESC LIMIT 1\G

Example output:

*************************** 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}

Output fields

For the full parameter reference, see Broker Load.

For the full syntax, run HELP SHOW LOAD.

View job logs

Spark Load logs are stored in log/spark_launcher_log/ under the frontend node root directory, named in the format spark-launcher-{load-job-id}-{label}.log. Logs are retained for three days by default and are deleted when the associated import metadata is purged.

Cancel an import job

Cancel a job that is not in FINISHED or CANCELLED state:

CANCEL LOAD WHERE LABEL = "label1";

For the full syntax, run HELP CANCEL LOAD.