func getApplicationType(applicationType plugins.SparkApplication_Type) sparkOp.SparkApplicationType {
	switch applicationType {
	case plugins.SparkApplication_PYTHON:
		return sparkOp.PythonApplicationType
	case plugins.SparkApplication_JAVA:
		return sparkOp.JavaApplicationType
	case plugins.SparkApplication_SCALA:
		return sparkOp.ScalaApplicationType
	case plugins.SparkApplication_R:
		return sparkOp.RApplicationType
	}
	return sparkOp.PythonApplicationType
}
Convert SparkJob ApplicationType to Operator CRD ApplicationType
package spark
import (
	"context"
	"fmt"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"strconv"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/tasklog"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core/template"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/flytek8s"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/flytek8s/config"
	"github.com/flyteorg/flyteplugins/go/tasks/errors"
	"github.com/flyteorg/flyteplugins/go/tasks/logs"
	pluginsCore "github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/core"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/k8s"
	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/utils"
	"k8s.io/client-go/kubernetes/scheme"
	sparkOp "github.com/GoogleCloudPlatform/spark-on-k8s-operator/pkg/apis/sparkoperator.k8s.io/v1beta2"
	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/plugins"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"regexp"
	"strings"
	"time"
)
const KindSparkApplication = "SparkApplication"
const sparkDriverUI = "sparkDriverUI"
const sparkHistoryUI = "sparkHistoryUI"
var featureRegex = regexp.MustCompile(`^spark.((flyteorg)|(flyte)).(.+).enabled$`)
var sparkTaskType = "spark"
type sparkResourceHandler struct {
}
func validateSparkJob(sparkJob *plugins.SparkJob) error {
	if sparkJob == nil {
		return fmt.Errorf("empty sparkJob")
	}
	if len(sparkJob.MainApplicationFile) == 0 && len(sparkJob.MainClass) == 0 {
		return fmt.Errorf("either MainApplicationFile or MainClass must be set")
	}
	return nil
}
func (sparkResourceHandler) GetProperties() k8s.PluginProperties {
	return k8s.PluginProperties{}
}
func (sparkResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsCore.TaskExecutionContext) (client.Object, error) {
	taskTemplate, err := taskCtx.TaskReader().Read(ctx)
	if err != nil {
		return nil, errors.Errorf(errors.BadTaskSpecification, "unable to fetch task specification [%v]", err.Error())
	} else if taskTemplate == nil {
		return nil, errors.Errorf(errors.BadTaskSpecification, "nil task specification")
	}
	sparkJob := plugins.SparkJob{}
	err = utils.UnmarshalStruct(taskTemplate.GetCustom(), &sparkJob)
	if err != nil {
		return nil, errors.Wrapf(errors.BadTaskSpecification, err, "invalid TaskSpecification [%v], failed to unmarshal", taskTemplate.GetCustom())
	}
	if err = validateSparkJob(&sparkJob); err != nil {
		return nil, errors.Wrapf(errors.BadTaskSpecification, err, "invalid TaskSpecification [%v].", taskTemplate.GetCustom())
	}
	annotations := utils.UnionMaps(config.GetK8sPluginConfig().DefaultAnnotations, utils.CopyMap(taskCtx.TaskExecutionMetadata().GetAnnotations()))
	labels := utils.UnionMaps(config.GetK8sPluginConfig().DefaultLabels, utils.CopyMap(taskCtx.TaskExecutionMetadata().GetLabels()))
	container := taskTemplate.GetContainer()
	envVars := flytek8s.DecorateEnvVars(ctx, flytek8s.ToK8sEnvVar(container.GetEnv()), taskCtx.TaskExecutionMetadata().GetTaskExecutionID())
	sparkEnvVars := make(map[string]string)
	for _, envVar := range envVars {
		sparkEnvVars[envVar.Name] = envVar.Value
	}
	sparkEnvVars["FLYTE_MAX_ATTEMPTS"] = strconv.Itoa(int(taskCtx.TaskExecutionMetadata().GetMaxAttempts()))
	serviceAccountName := flytek8s.GetServiceAccountNameFromTaskExecutionMetadata(taskCtx.TaskExecutionMetadata())
	if len(serviceAccountName) == 0 {
		serviceAccountName = sparkTaskType
	}
	driverSpec := sparkOp.DriverSpec{
		SparkPodSpec: sparkOp.SparkPodSpec{
			Annotations: annotations,
			Labels:      labels,
			EnvVars:     sparkEnvVars,
			Image:       &container.Image,
		},
		ServiceAccount: &serviceAccountName,
	}
	executorSpec := sparkOp.ExecutorSpec{
		SparkPodSpec: sparkOp.SparkPodSpec{
			Annotations: annotations,
			Labels:      labels,
			Image:       &container.Image,
			EnvVars:     sparkEnvVars,
		},
	}
	modifiedArgs, err := template.Render(ctx, container.GetArgs(), template.Parameters{
		TaskExecMetadata: taskCtx.TaskExecutionMetadata(),
		Inputs:           taskCtx.InputReader(),
		OutputPath:       taskCtx.OutputWriter(),
		Task:             taskCtx.TaskReader(),
	})
	if err != nil {
		return nil, err
	}
	submissionFailureRetries := int32(14)
	sparkConfig := make(map[string]string)
	for k, v := range GetSparkConfig().DefaultSparkConfig {
		sparkConfig[k] = v
	}
	if sparkJob.GetExecutorPath() != "" {
		sparkConfig["spark.pyspark.python"] = sparkJob.GetExecutorPath()
		sparkConfig["spark.pyspark.driver.python"] = sparkJob.GetExecutorPath()
	}
	for k, v := range sparkJob.GetSparkConf() {
		if featureRegex.MatchString(k) {
			addConfig(sparkConfig, k, v)
		} else {
			sparkConfig[k] = v
		}
	}
	if len(sparkConfig["spark.kubernetes.driver.limit.cores"]) == 0 {
		if len(sparkConfig["spark.kubernetes.driver.request.cores"]) != 0 {
			sparkConfig["spark.kubernetes.driver.limit.cores"] = sparkConfig["spark.kubernetes.driver.request.cores"]
		} else if len(sparkConfig["spark.driver.cores"]) != 0 {
			sparkConfig["spark.kubernetes.driver.limit.cores"] = sparkConfig["spark.driver.cores"]
		}
	}
	if len(sparkConfig["spark.kubernetes.executor.limit.cores"]) == 0 {
		if len(sparkConfig["spark.kubernetes.executor.request.cores"]) != 0 {
			sparkConfig["spark.kubernetes.executor.limit.cores"] = sparkConfig["spark.kubernetes.executor.request.cores"]
		} else if len(sparkConfig["spark.executor.cores"]) != 0 {
			sparkConfig["spark.kubernetes.executor.limit.cores"] = sparkConfig["spark.executor.cores"]
		}
	}
	sparkConfig["spark.kubernetes.executor.podNamePrefix"] = taskCtx.TaskExecutionMetadata().GetTaskExecutionID().GetGeneratedName()
	sparkConfig["spark.kubernetes.driverEnv.FLYTE_START_TIME"] = strconv.FormatInt(time.Now().UnixNano()/1000000, 10)
	cores, err := strconv.ParseInt(sparkConfig["spark.driver.cores"], 10, 32)
	if err == nil {
		driverSpec.Cores = intPtr(int32(cores))
	}
	driverSpec.Memory = strPtr(sparkConfig["spark.driver.memory"])
	execCores, err := strconv.ParseInt(sparkConfig["spark.executor.cores"], 10, 32)
	if err == nil {
		executorSpec.Cores = intPtr(int32(execCores))
	}
	execCount, err := strconv.ParseInt(sparkConfig["spark.executor.instances"], 10, 32)
	if err == nil {
		executorSpec.Instances = intPtr(int32(execCount))
	}
	executorSpec.Memory = strPtr(sparkConfig["spark.executor.memory"])
	j := &sparkOp.SparkApplication{
		TypeMeta: metav1.TypeMeta{
			Kind:       KindSparkApplication,
			APIVersion: sparkOp.SchemeGroupVersion.String(),
		},
		Spec: sparkOp.SparkApplicationSpec{
			ServiceAccount: &serviceAccountName,
			Type:           getApplicationType(sparkJob.GetApplicationType()),
			Image:          &container.Image,
			Arguments:      modifiedArgs,
			Driver:         driverSpec,
			Executor:       executorSpec,
			SparkConf:      sparkConfig,
			HadoopConf:     sparkJob.GetHadoopConf(),
			RestartPolicy: sparkOp.RestartPolicy{
				Type:                       sparkOp.OnFailure,
				OnSubmissionFailureRetries: &submissionFailureRetries,
			},
		},
	}
	if val, ok := sparkConfig["spark.batchScheduler"]; ok {
		j.Spec.BatchScheduler = &val
	}
	if sparkJob.MainApplicationFile != "" {
		j.Spec.MainApplicationFile = &sparkJob.MainApplicationFile
	}
	if sparkJob.MainClass != "" {
		j.Spec.MainClass = &sparkJob.MainClass
	}
	if taskCtx.TaskExecutionMetadata().IsInterruptible() {
		j.Spec.Executor.Tolerations = config.GetK8sPluginConfig().InterruptibleTolerations
		j.Spec.Executor.NodeSelector = config.GetK8sPluginConfig().InterruptibleNodeSelector
	}
	return j, nil
}
func addConfig(sparkConfig map[string]string, key string, value string) {
	if strings.ToLower(strings.TrimSpace(value)) != "true" {
		sparkConfig[key] = value
		return
	}
	matches := featureRegex.FindAllStringSubmatch(key, -1)
	if len(matches) == 0 || len(matches[0]) == 0 {
		sparkConfig[key] = value
		return
	}
	featureName := matches[0][len(matches[0])-1]
	for _, feature := range GetSparkConfig().Features {
		if feature.Name == featureName {
			for k, v := range feature.SparkConfig {
				sparkConfig[k] = v
			}
			return
		}
	}
	sparkConfig[key] = value
}