feat: support spark submit jar on k8s #4867

lenoxzhao · 2023-08-16T10:45:49Z

What is the purpose of the change

Support submit spark jar job on k8s.

Environment

Spark：v3.2.1
Kubernetes: v1.23

$ kubectl version
Client Version: version.Info{Major:"1", Minor:"23", GitVersion:"v1.23.5", GitCommit:"c285e781331a3785a7f436042c65c5641ce8a9e9", GitTreeState:"clean", BuildDate:"2022-03-16T15:58:47Z", GoVersion:"go1.17.8", Compiler:"gc", Platform:"linux/amd64"}
Server Version: version.Info{Major:"1", Minor:"23", GitVersion:"v1.23.17", GitCommit:"953be8927218ec8067e1af2641e540238ffd7576", GitTreeState:"clean", BuildDate:"2023-02-22T13:27:46Z", GoVersion:"go1.19.6", Compiler:"gc", Platform:"linux/amd64"}

Linkis：

Test Command

linkis-cli --mode once \
-code 'sparkjars' \
-engineType spark-3.2.1 \
-codeType jar \
-labelMap engineConnMode=once \
-sourceMap jobName=OnceJobTest \
-k8sCluster 'K8S-default' \
-jobContentMap spark.app.main.class='org.apache.spark.examples.SparkPi' \
-jobContentMap runType='jar' \
-confMap spark.master='k8s-native' \
-confMap linkis.spark.k8s.serviceAccount='spark' \
-confMap spark.submit.deployMode='cluster' \
-confMap linkis.spark.k8s.master.url='k8s://https://172.31.226.155:6443' \
-confMap linkis.spark.k8s.config.file='/home/hadoop/.kube/config' \
-confMap linkis.spark.k8s.imagePullPolicy='IfNotPresent' \
-confMap spark.app.name='spark-submit-jar-k8s-test-linkis' \
-confMap spark.app.resource='local:///opt/spark/examples/jars/spark-examples_2.12-3.2.1.jar' \
-confMap linkis.spark.k8s.namespace='default'

Brief change log

SparkConfig.java: add spark.kubernetes.driver.request.cores and spark.kubernetes.executor.request.cores related fields.
ClusterDescriptorAdapterFactory.java: add new k8s submission way("k8s-native", different from "k8s-operator").
KubernetesApplicationClusterDescriptorAdapter.java: main features are as below
- SparkLauncher add k8s related params setting.
- Manually specifying driver pod name for follow-up task monitoring.
- initJobId() method converts the state of the driver pod to the SparkAppHandle.State to determine whether the task has started.
SparkOnKubernetesSubmitOnceExecutor.scala: new once executor type for k8s.
SparkEngineConnFactory.scala: add linkis.spark.k8s.driver.request.cores and linkis.spark.k8s.executor.request.cores to sparkConf.
SparkEngineConnResourceFactory.scala: According to Spark Doc , spark.kubernetes.executor.request.cores and spark.kubernetes.driver.request.cores have priority over spark.executor.cores and spark.driver.cores. So if spark.kubernetes.executor.request.cores is not specified when check resources, manually assign the value of spark.executor.cores to spark.kubernetes.executor.request.cores to make sure the default value of spark.kubernetes.executor.request.cores won't affect the tasks.
SparkOnceExecutorFactory.scala: Create SparkOnKubernetesSubmitOnceExecutor if -k8sCluster is specified.
SparkJobProgressUtil.scala: main features are as below
- Methods add podIP params to be compatible with the spark tasks on k8s.
- getKubernetesSparkJobInfo(): method for getting status of spark tasks through Spark UI exposed on driver pod’s 4040 port.

Checklist

I have read the Contributing Guidelines on pull requests.
I have explained the need for this PR and the problem it solves
I have explained the changes or the new features added to this PR
I have added tests corresponding to this change
I have updated the documentation to reflect this change
I have verified that this change is backward compatible (If not, please discuss on the Linkis mailing list first)
If this is a code change: I have written unit tests to fully verify the new behavior.

chenghuichen · 2023-08-16T16:15:33Z

...nkis/engineplugin/spark/client/deployment/KubernetesApplicationClusterDescriptorAdapter.java

+        .setDeployMode(sparkConfig.getDeployMode())
+        .setAppName(sparkConfig.getAppName())
+        .setVerbose(true);
+    this.driverPodName = sparkConfig.getAppName() + "-" + System.currentTimeMillis() + "-driver";


Using timestamp to make driverPodName unique is not enough good, UUID would be better.
If this is for pod create time recording, there is creationTimestamp in pod spec already.

Using timestamp to make driverPodName unique is not enough good, UUID would be better. If this is for pod create time recording, there is creationTimestamp in pod spec already.

Thanks for reviewing the codes and changes have been made in the last commitment.

chenghuichen · 2023-08-16T16:29:02Z

.../org/apache/linkis/engineplugin/spark/client/deployment/ClusterDescriptorAdapterFactory.java

+    if (StringUtils.isNotBlank(master)) {
+      if (master.equalsIgnoreCase("k8s-operator")) {
+        clusterDescriptorAdapter = new KubernetesOperatorClusterDescriptorAdapter(executionContext);
+      } else if (master.equalsIgnoreCase("k8s-jar")) {


In my opinion, "k8s-native" would be more suitable than "k8s-jar", which is aligned to the concept "k8s-operator".

chenghuichen · 2023-08-16T16:29:22Z

...cala/org/apache/linkis/engineplugin/spark/executor/SparkOnKubernetesSubmitOnceExecutor.scala

+      logger.info("clusterDescriptorAdapter.getJobState is null")
+    } else {
+      logger.info("clusterDescriptorAdapter/getJobState is not null")
+    }


These logs seem to be used for debugging. May be too noisy in prod env.

peacewong · 2023-08-25T08:12:05Z

...k/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkOnceExecutorFactory.scala

    engineConn.getEngineConnSession match {
+      case context: SparkEngineConnContext


It is recommended to optimize it into an if else judgment

Already done in the last commitment.

guoshupei

LGTM

peacewong

LGTM.

lenoxzhao added 2 commits August 14, 2023 11:25

feat: support spark submit jar on k8s

62d1d3d

feat: add spark cores setting priority

f1ea3e3

chenghuichen reviewed Aug 16, 2023

View reviewed changes

feat: use UUID to generate driverPodName

4f9c2bd

peacewong reviewed Aug 25, 2023

View reviewed changes

feat: optimize code of executor creation

f12c22f

guoshupei reviewed Aug 28, 2023

View reviewed changes

peacewong approved these changes Sep 8, 2023

View reviewed changes

peacewong merged commit 5ca6726 into apache:master Sep 8, 2023
12 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support spark submit jar on k8s #4867

feat: support spark submit jar on k8s #4867

lenoxzhao commented Aug 16, 2023 •

edited

Loading

chenghuichen Aug 16, 2023

lenoxzhao Aug 17, 2023

chenghuichen Aug 16, 2023

chenghuichen Aug 16, 2023

peacewong Aug 25, 2023

lenoxzhao Aug 26, 2023

guoshupei left a comment

peacewong left a comment

		engineConn.getEngineConnSession match {
		case context: SparkEngineConnContext

feat: support spark submit jar on k8s #4867

feat: support spark submit jar on k8s #4867

Conversation

lenoxzhao commented Aug 16, 2023 • edited Loading

What is the purpose of the change

Environment

Test Command

Brief change log

Checklist

chenghuichen Aug 16, 2023

Choose a reason for hiding this comment

lenoxzhao Aug 17, 2023

Choose a reason for hiding this comment

chenghuichen Aug 16, 2023

Choose a reason for hiding this comment

chenghuichen Aug 16, 2023

Choose a reason for hiding this comment

peacewong Aug 25, 2023

Choose a reason for hiding this comment

lenoxzhao Aug 26, 2023

Choose a reason for hiding this comment

guoshupei left a comment

Choose a reason for hiding this comment

peacewong left a comment

Choose a reason for hiding this comment

lenoxzhao commented Aug 16, 2023 •

edited

Loading