Spark load interface

[wyb]

1. User interface:

• Submit Spark load job

LOAD LABEL db_name.label_name 
(
  DATA INFILE "/tmp/file" FROM TABLE hive_db.table ...,
  DATA INFILE "/tmp/file1" ...
  -- DATA FROM TABLE hive_cluster.db.table ...
)
WITH spark.cluster_name
[PROPERTIES (key1=value1, ... )]

spark.cluster_name is the name of the cluster used by Spark ETL application. You can use the following sql command to manage it.

• Spark cluster info management

-- Add import cluster for user 'user_name'
SET PROPERTY FOR 'user_name' 
    'load_cluster.spark.cluster_name.output_path' = '/user/output', 
    'load_cluster.spark.cluster_name.configs' = 'key1=value1;key2=value2';

-- Delete import cluster for user 'user_name'
SET PROPERTY FOR 'user_name' 'load_cluster.spark.cluster_name' = '';

-- Set default import cluster for user 'user_name'
SET PROPERTY FOR 'user_name' 'default_load_cluster' = 'spark.cluster_name';

2. Spark ETL application interface

public EtlSubmitResult submitEtlJob(long jobId, long txnId, JobConf jobConf) 

public EtlStatus getEtlJobStatus(String etlJobId) 

public void killEtlJob(String etlJobId)

public Map<String, Long> getEtlFiles(String outputPath)

2.1 jobConf includes:

• Spark cluster infos

• ETL output directory and file name format

• Schema of the imported table, including columns, partitions, and rollups

• Infos of the source file, including split rules, corresponding columns, and conversion rules

2.2 Save config as local json file named config.json for Spark ETL app.

{
	"tables": {
		10014: {
			"columns": {
				"k1": {
					"default_value": "\\N",
					"column_type": "DATETIME",
					"is_allow_null": true
				},
				"k2": {
					"default_value": "0",
					"column_type": "SMALLINT",
					"is_allow_null": true
				},
				"v": {
					"default_value": "0",
					"column_type": "BIGINT",
					"is_allow_null": false
				}
			},
			"indexes": {
				10014: {
					"column_refs": [{
						"name": "k1",
						"is_key": true,
						"aggregation_type": "NONE"
					}, {
						"name": "k2",
						"is_key": true,
						"aggregation_type": "NONE"
					}, {
						"name": "v",
						"is_key": false,
						"aggregation_type": "NONE"
					}],
					"schema_hash": 1294206574
				},
				10017: {
					"column_refs": [{
						"name": "k1",
						"is_key": true,
						"aggregation_type": "NONE"
					}, {
						"name": "v",
						"is_key": false,
						"aggregation_type": "SUM"
					}],
					"schema_hash": 1294206575
				}
			},
			"partition_info": {
				"partition_type": "range", // or unpartitioned
				"partition_column_refs": ["k2"],
				"distribution_column_refs": ["k1"],
				"partitions": {
					10020: {
						"start_keys": [-100],
						"end_keys": [10],
						"is_max_partition": false,
						"buckets_num": 3
					}
				}
			},
			"sources": {
				"source0": {
					"partitions": [10020],
					"file_urls": ["hdfs://hdfs_host:port/user/palo/test/file"],
					"columns": ["tmp_k1", "k2"],
					"column_separator": ",",
					"column_mappings": {
						"k1": {
							"function_name": "strftime",
							"args": ["%Y-%m-%d %H:%M:%S", "tmp_k1"]
						}
					},
					"where": "k2 > 10",
					"is_negative": false,
					"hive_table_name": "hive_db.table" // for global dict
				}
			}
		}
	},
	"output_path": "hdfs://hdfs_host:port/user/output/10003/label1/1582599203397",
	"output_file_pattern": "label1.%(table_id)d.%(index_id)d.%(bucket)d.%(schema_hash)d"
}

2.3 getEtlJobStatus returns the status of Spark ETL app, including state, numCompletedTasks, numTasks, counters.

2.4 getEtlFiles returns a Map, the key is the file generated by ETL, and the value is the size of file.

[wyb]

Discussed with @wangbo, update loadstmt for temporary use.

LOAD LABEL db_name.label_name 
(
  DATA INFILE "/tmp/file1" INTO TABLE t1 ...,
  -- DATA FROM TABLE hive_table ...
)
WITH spark.cluster_name
[PROPERTIES ("bitmap_data" = "hive_db.table", key1=value1, ... )]

[kangpinghuang]

hi, I think JobConf should add the level relation(tree) of rollup indexes which can be decided better in FE.
tables/columns/sources/indexes is a array, not a struct?

[wyb]

@kangpinghuang
Level relation for rollup data conversion is the strategy in ETL, if we update the strategy later, we can only update the ETL code, so i think it is better to deal with it in ETL.

We will get the attributes of column by column name, it is more convenient when using map.

[wangbo]

I think JobConf should add the level relation(tree) of rollup indexes which can be decided better in FE.

Level relation for rollup data conversion is the strategy in ETL, if we update the strategy later, we can only update the ETL code, so i think it is better to deal with it in ETL.

I think we can make a RollupUtil,it read input info(table schema) and calculate the rollup relation,then we can package it into a jar which would be submitted with spark jar.
So we can use it both in Fe and EtLJob.
Totally,We can calculate the tree relation using table schema when we need it.

[wangbo]

1. add spark job optimize args in jobConfig

Every Spark ETL job has different input data,so they have different resource requirements.
It's better to add a spark job submit args info in job config in order to adjust every job's usage.
User can adjust respective job config to make job running successfully.

2. Job Retry

In the initial implementation version,I wonder whether we need to support retry interface ,or user need to resubmit the whole job when job failed due to spark memory exceeds and so on.

3. What does Fe going to do with interface getEtlFiles?

Just a little question.

[wyb]

@wangbo

1. General configs can be set when adding a Spark cluster, and user can set different configs in the loadstmt which will override the generals when submitting the load job.

LOAD LABEL db_name.label_name 
(
  DATA INFILE "/tmp/file1" INTO TABLE t1 ...,
  -- DATA FROM TABLE hive_table ...
)
WITH CLUSTER spark.cluster_name
(spark_param_key=spark_param_value, ...)
[PROPERTIES ("bitmap_data" = "hive_db.table", key1=value1, ... )]

2. In the initial implementation version, user need to resubmit the whole job when job failed.

3. After ETL is completed, FE will get the ETL files path and schedule push tasks to the relevant BE tablet.

public Map<String, Long> getEtlFilePaths(String outputPath)

[kangpinghuang]

the output file pattern should be: "output_file_pattern": "label1..%(table_id)d.%(partition_id)d%(index_id)d.%(bucket)d.%(schema_hash)d"

[wyb]

update jobconfig.json file format

{
	"tables": {
		10014: {
			"columns": {
				"k1": {
					"default_value": "\\N",
					"column_type": "DATETIME",
					"is_allow_null": true
				},
				"k2": {
					"default_value": "0",
					"column_type": "SMALLINT",
					"is_allow_null": true
				},
				"v": {
					"default_value": "0",
					"column_type": "BIGINT",
					"is_allow_null": false
				}
			},
			"indexes": {
				10014: {
					"column_refs": [{
						"name": "k1",
						"is_key": true,
						"aggregation_type": "NONE"
					}, {
						"name": "k2",
						"is_key": true,
						"aggregation_type": "NONE"
					}, {
						"name": "v",
						"is_key": false,
						"aggregation_type": "NONE"
					}],
					"distribution_column_refs": ["k1"],
					"schema_hash": 1294206574
				},
				10017: {
					"column_refs": [{
						"name": "k1",
						"is_key": true,
						"aggregation_type": "NONE"
					}, {
						"name": "v",
						"is_key": false,
						"aggregation_type": "SUM"
					}],
					"distribution_column_refs": ["k1"],
					"schema_hash": 1294206575
				}
			},
			"partition_info": {
				"partition_type": "RANGE",
				"partition_column_refs": ["k2"],
				"partitions": {
					10020: {
						"start_keys": [-100],
						"end_keys": [10],
						"is_max_partition": false,
						"bucket_num": 3
					}
				}
			},
			"file_groups": [{
				"partitions": [10020],
				"file_paths": ["hdfs://hdfs_host:port/user/palo/test/file"],
				"file_field_names": ["tmp_k1", "k2"],
				"value_separator": ",",
				"line_delimiter": "\n"
				"column_mappings": {
					"k1": {
						"function_name": "strftime",
						"args": ["%Y-%m-%d %H:%M:%S", "tmp_k1"]
					}
				},
				"where": "k2 > 10",
				"is_negative": false,
				"hive_table_name": "hive_db.table"
			}]
		}
	},
	"output_path": "hdfs://hdfs_host:port/user/output/10003/label1/1582599203397",
	"output_file_pattern": "label1.%(table_id)d.%(partition_id)d.%(index_id)d.%(bucket)d.%(schema_hash)d"
}

[wyb]

1. User interface:

1.1 Spark resource management

Spark is used as an external computing resource in Doris to do ETL work. In the future, there may be other external resources that will be used in Doris, for example, MapReduce is used for ETL, Spark/GPU is used for queries, HDFS/S3 is used for external storage. We introduced resource management to manage these external resources used by Doris.

-- 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_identity
GRANT USAGE_PRIV ON RESOURCE resource_name TO ROLE role_name

REVOKE USAGE_PRIV ON RESOURCE resource_name FROM user_identity
REVOKE USAGE_PRIV ON RESOURCE resource_name FROM ROLE role_name

• CREATE EXTERNAL RESOURCE:

FOR user_name is optional. If there has, the external resource belongs to this user. If not, the external resource belongs to the system and all users are available.

PROPERTIES：

1. type: resource type. Only support spark now.
2. spark configuration: follow the standard writing of Spark configurations, refer to: https://spark.apache.org/docs/latest/configuration.html.
3. working_dir: optional, used to store ETL intermediate results in spark ETL.
4. broker: optional, used in spark ETL. The ETL intermediate results need to be read with the broker when pushed into BE.

Example:

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.yarn.queue" = "queue0",
  "spark.executor.memory" = "1g",
  "spark.hadoop.yarn.resourcemanager.address" = "127.0.0.1:9999",
  "spark.hadoop.fs.defaultFS" = "hdfs://127.0.0.1:10000",
  "working_dir" = "hdfs://127.0.0.1:10000/tmp/doris",
  "broker" = "broker0",
  "broker.username" = "user0",
  "broker.password" = "password0"
)

• SHOW RESOURCES:
  General users can only see their own resources.
  Admin and root users can show all resources.

1.2 Create spark load job

LOAD LABEL db_name.label_name 
(
  DATA INFILE ("/tmp/file1") INTO TABLE table_name, ...
)
WITH RESOURCE resource_name
[(key1 = value1, ...)]
[PROPERTIES (key2 = value2, ... )]

Example:

LOAD LABEL example_db.test_label 
(
  DATA INFILE ("hdfs:/127.0.0.1:10000/tmp/file1") INTO TABLE example_table
)
WITH RESOURCE "spark0"
(
  "spark.executor.memory" = "1g",
  "spark.files" = "/tmp/aaa,/tmp/bbb"
)
PROPERTIES ("timeout" = "3600")

The spark configuartions in load stmt can override the existing configuration in the resource for temporary use.