Files
Yi ChenandGitHub 984c47c22b Add support for Spark Connect (#2569)
* Add v1alpha1 version of CRD SparkConnect

Signed-off-by: Yi Chen <[email protected]>

* Generate SparkConnect CRD manifests

Signed-off-by: Yi Chen <[email protected]>

* Implement SparkConnect controller

Signed-off-by: Yi Chen <[email protected]>

* Update Helm chart CRDs and RBAC resources

Signed-off-by: Yi Chen <[email protected]>

* Add SparkConnect example

Signed-off-by: Yi Chen <[email protected]>

* Add pre stop lifecycle handler

Signed-off-by: Yi Chen <[email protected]>

* Rename LabelSparkConnName to LabelSparkConnectName

Signed-off-by: Yi Chen <[email protected]>

* Rename SparkConnectStateUnready to SparkConnectStateNotReady

Signed-off-by: Yi Chen <[email protected]>

* Add constant SparkConfigMapVolumeMountName

Signed-off-by: Yi Chen <[email protected]>

* Use server pod status message as SparkConnect condition message

Signed-off-by: Yi Chen <[email protected]>

* Define SparkConnect condition reasons

Signed-off-by: Yi Chen <[email protected]>

* Add prefix to Hadoop properties if it does not start with 'spark.hadoop.'

Signed-off-by: Yi Chen <[email protected]>

---------

Signed-off-by: Yi Chen <[email protected]>
2025-07-02 03:21:18 +00:00

68 lines
2.1 KiB
Go

/*
Copyright 2024 The Kubeflow authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package util
import (
corev1 "k8s.io/api/core/v1"
"github.com/kubeflow/spark-operator/v2/pkg/common"
)
// IsLaunchedBySparkOperator returns whether the given pod is launched by the Spark Operator.
func IsLaunchedBySparkOperator(pod *corev1.Pod) bool {
return pod.Labels[common.LabelLaunchedBySparkOperator] == "true"
}
// IsDriverPod returns whether the given pod is a Spark driver Pod.
func IsDriverPod(pod *corev1.Pod) bool {
return pod.Labels[common.LabelSparkRole] == common.SparkRoleDriver
}
// IsExecutorPod returns whether the given pod is a Spark executor Pod.
func IsExecutorPod(pod *corev1.Pod) bool {
return pod.Labels[common.LabelSparkRole] == common.SparkRoleExecutor
}
// GetSparkExecutorID returns the Spark executor ID by checking out pod labels.
func GetSparkExecutorID(pod *corev1.Pod) string {
return pod.Labels[common.LabelSparkExecutorID]
}
// GetAppName returns the spark application name by checking out pod labels.
func GetAppName(pod *corev1.Pod) string {
return pod.Labels[common.LabelSparkAppName]
}
// GetConnName returns the spark connection name by checking out pod labels.
func GetConnName(pod *corev1.Pod) string {
return pod.Labels[common.LabelSparkConnectName]
}
// GetSparkApplicationID returns the spark application ID by checking out pod labels.
func GetSparkApplicationID(pod *corev1.Pod) string {
return pod.Labels[common.LabelSparkApplicationSelector]
}
func IsPodReady(pod *corev1.Pod) bool {
for _, condition := range pod.Status.Conditions {
if condition.Type == corev1.PodReady && condition.Status == corev1.ConditionTrue {
return true
}
}
return false
}