tencent cloud

                 +
                 | 0. User create spark load job
            +----v----+
            |   FE    |---------------------------------+
            +----+----+                                 |
                 | 3. FE send push tasks                |
                 | 5. FE publish version                |
    +------------+------------+                         |
    |            |            |                         |
+---v---+    +---v---+    +---v---+                     |
|  BE   |    |  BE   |    |  BE   |                     |1. FE submit Spark ETL job
+---^---+    +---^---+    +---^---+                     |
    |4. BE push with broker   |                         |
+---+---+    +---+---+    +---+---+                     |
|Broker |    |Broker |    |Broker |                     |
+---^---+    +---^---+    +---^---+                     |
    |            |            |                         |
+---+------------+------------+---+ 2.ETL +-------------v---------------+
|               HDFS              +------->       Spark cluster         |
|                                 <-------+                             |
+---------------------------------+       +-----------------------------+

spark-submit --queue default --master yarn --deploy-mode cluster --class org.apache.spark.examples.SparkPi /usr/local/service/spark/examples/jars/spark-examples_*.jar 10

mkdir -p /usr/local/service/doris/log/spark_launcher_log; 
chmod 777 /usr/local/service/doris/log/spark_launcher_log

spark_home_default_dir=/usr/local/service/spark
spark_resource_path=/usr/local/service/spark/jars/spark_jars.zip
yarn_client_path=/usr/local/service/hadoop/bin/yarn
yarn_config_dir=/usr/local/service/hadoop/etc/hadoop
spark_dpp_version=1.2-SNAPSHOT

CREATE EXTERNAL RESOURCE spark_resource_xxx
PROPERTIES
(
  "type" = "spark",
  "spark.master" = "yarn",
  "spark.submit.deployMode" = "cluster",
  "spark.yarn.queue" = "<xxx_queue>",
  "spark.hadoop.yarn.resourcemanager.ha.enabled" = "true",
  "spark.hadoop.yarn.resourcemanager.ha.rm-ids" = "rm1,rm2",
  "spark.hadoop.yarn.resourcemanager.address.rm1" = "<rm1_host>:<rm1_port>",
  "spark.hadoop.yarn.resourcemanager.address.rm2" = "<rm2_host>:<rm2_port>",
  "spark.hadoop.fs.defaultFS" = "hdfs://<hdfs_defaultFS>",
  "spark.hadoop.dfs.nameservices" = "<hdfs_defaultFS>",
  "spark.hadoop.dfs.ha.namenodes.<hdfs_defaultFS>" = "nn1,nn2",
  "spark.hadoop.dfs.namenode.rpc-address.<hdfs_defaultFS>.nn1" = "<nn1_host>:<nn1_port>",
  "spark.hadoop.dfs.namenode.rpc-address.<hdfs_defaultFS>.nn2" = "<nn2_host>:<nn2_port>",
  "spark.hadoop.dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
  "working_dir" = "hdfs://<hdfs_defaultFS>/doris/spark_load",
  "broker" = "<doris_broker_name>",
  "broker.username" = "hadoop",
  "broker.password" = "",
  "broker.dfs.nameservices" = "<hdfs_defaultFS>",
  "broker.dfs.ha.namenodes.<hdfs_defaultFS>" = "nn1, nn2",
  "broker.dfs.namenode.rpc-address.HDFS4001273.nn1" = "<nn1_host>:<nn1_port>",
  "broker.dfs.namenode.rpc-address.HDFS4001273.nn2" = "<nn2_host>:<nn2_port>",
  "broker.dfs.client.failover.proxy.provider" = "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
);

-- Grant user0 the usage rights to spark0 resource
GRANT USAGE_PRIV ON RESOURCE "spark_resource_xxx" TO "user0"@"%";

-- Grant role0 the usage rights to spark0 resource
GRANT USAGE_PRIV ON RESOURCE "spark_resource_xxx" TO ROLE "role0";

-- Grant all resources usage rights to user0
GRANT USAGE_PRIV ON RESOURCE * TO "user0"@"%";

-- Grant all resources usage rights to role role0
GRANT USAGE_PRIV ON RESOURCE * TO ROLE "role0";

-- Revoke user0’s spark0 resource usage rights
REVOKE USAGE_PRIV ON RESOURCE "spark_resource_xxx" FROM "user0"@"%";

LOAD LABEL test_label_01    --label
(
 DATA INFILE ("hdfs://HDFS4001234/warehouse/ods.db/user_events/ds=2023-04-15/*"  --file path) 
 INTO TABLE user_events   --doris table
 FORMAT AS "parquet"      --data format
 ( event_time, user_id, op_code)  --columns in file 
 COLUMNS FROM PATH AS ( `ds` )     --partition column
 SET 
 (                                    --column mapping
  ds = ds,
  event_time = event_time,
  user_id = user_id,
  op_code = op_code
 ) 
) 
WITH RESOURCE 'spark_resource_xxx' 
(                                         --spark job params
 "spark.executor.memory" = "4g", 
 "spark.default.parallelism" = "400", 
 "spark.executor.cores" = '5', 
 "spark.executor.instances" = '10'
) 
PROPERTIES 
(                                             --doris load task params
 "timeout" = "259200" 
);

- State
    The current stage of the import task. After the task is submitted, the state is PENDING. After submitting Spark ETL, the state changes to ETL. After ETL is completed, FE schedules BE to perform the push operation and the state changes to LOADING. After the push is completed and the version takes effect, the state changes to FINISHED.
    There are two final stages of the import task: CANCELLED and FINISHED. The import is completed when the load job is in these two stages. Among them, CANCELLED means import fails, and FINISHED means import is successful.
    
- Progress
    Description of the progress of the import task. There are two types of progress: ETL and LOAD, corresponding to the two stages of the import process ETL and LOADING.
    The progress range of LOAD is: 0~100%.
     `LOAD progress = Number of Tablets that have completed importing all Replicas / Total number of Tablets for this import task * 100%` 
    **If all import tables completes import, the LOAD progress is 99%** The import enters the final effective stage. After the entire import is completed, the LOAD progress will change to 100%.
    Import progress is not linear. Therefore, if the progress does not change for a while, it does not mean that the import is not being executed.