Inside Spark Operator: How Kubernetes Manages Spark Jobs End‑to‑End

Before diving into Spark Operator, it is useful to understand the concept of a Kubernetes operator. 2018 saw a rapid proliferation of operators, which extend Kubernetes by managing custom resources through dedicated controllers that watch and reconcile desired state.

Standard Operator Components

CRD definition that describes the abstracted functionality.

CRD controller that parses the definition and manages its lifecycle.

client-go SDK that provides the programming interface for integration.

Spark Operator Code Structure

All core logic resides under the pkg directory. apis defines versioned API objects. client contains the generated client‑go SDK. crd defines the two custom resources sparkapplication and scheduledsparkapplication. controller implements the operator’s lifecycle management. config handles conversion of Spark configuration.

The two CRDs differ: sparkapplication represents a one‑time Spark job whose Pods end in Succeed or Failed, while scheduledsparkapplication adds a cron‑like schedule for periodic offline tasks.

Job Submission Flow

When a SparkApplication CR is created, the operator translates its spec into a spark-submit command, optionally injects Prometheus monitoring, and runs the command inside the operator pod. The following function shows the core steps:

func (c *Controller) submitSparkApplication(app *v1beta1.SparkApplication) *v1beta1.SparkApplication {
    // expose Prometheus metrics
    appToSubmit := app.DeepCopy()
    if appToSubmit.Spec.Monitoring != nil && appToSubmit.Spec.Monitoring.Prometheus != nil {
        if err := configPrometheusMonitoring(appToSubmit, c.kubeClient); err != nil {
            glog.Error(err)
        }
    }
    // build spark-submit arguments from CRD
    submissionCmdArgs, err := buildSubmissionCommandArgs(appToSubmit)
    if err != nil {
        app.Status = v1beta1.SparkApplicationStatus{AppState: v1beta1.ApplicationState{State: v1beta1.FailedSubmissionState, ErrorMessage: err.Error()}, SubmissionAttempts: app.Status.SubmissionAttempts+1, LastSubmissionAttemptTime: metav1.Now()}
        return app
    }
    // run spark-submit inside the operator container
    submitted, err := runSparkSubmit(newSubmission(submissionCmdArgs, appToSubmit))
    if err != nil {
        app.Status = v1beta1.SparkApplicationStatus{AppState: v1beta1.ApplicationState{State: v1beta1.FailedSubmissionState, ErrorMessage: err.Error()}, SubmissionAttempts: app.Status.SubmissionAttempts+1, LastSubmissionAttemptTime: metav1.Now()}
        c.recordSparkApplicationEvent(app)
        glog.Errorf("failed to run spark-submit for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
        return app
    }
    if !submitted {
        // No error but submission was skipped (e.g., pod already exists)
        return app
    }
    glog.Infof("SparkApplication %s/%s has been submitted", app.Namespace, app.Name)
    app.Status = v1beta1.SparkApplicationStatus{AppState: v1beta1.ApplicationState{State: v1beta1.SubmittedState}, SubmissionAttempts: app.Status.SubmissionAttempts+1, ExecutionAttempts: app.Status.ExecutionAttempts+1, LastSubmissionAttemptTime: metav1.Now()}
    // expose Spark UI via Service and optional Ingress
    service, err := createSparkUIService(app, c.kubeClient)
    if err != nil {
        glog.Errorf("failed to create UI service for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
    } else {
        app.Status.DriverInfo.WebUIServiceName = service.serviceName
        app.Status.DriverInfo.WebUIPort = service.nodePort
        if c.ingressURLFormat != "" {
            ingress, err := createSparkUIIngress(app, *service, c.ingressURLFormat, c.kubeClient)
            if err != nil {
                glog.Errorf("failed to create UI Ingress for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
            } else {
                app.Status.DriverInfo.WebUIIngressAddress = ingress.ingressURL
                app.Status.DriverInfo.WebUIIngressName = ingress.ingressName
            }
        }
    }
    return app
}

The runSparkSubmit helper executes the generated command and handles common errors such as missing SPARK_HOME or an already‑existing driver pod:

func runSparkSubmit(submission *submission) (bool, error) {
    sparkHome, present := os.LookupEnv(sparkHomeEnvVar)
    if !present {
        glog.Error("SPARK_HOME is not specified")
    }
    var command = filepath.Join(sparkHome, "/bin/spark-submit")
    cmd := execCommand(command, submission.args...)
    glog.V(2).Infof("spark-submit arguments: %v", cmd.Args)
    if _, err := cmd.Output(); err != nil {
        var errorMsg string
        if exitErr, ok := err.(*exec.ExitError); ok {
            errorMsg = string(exitErr.Stderr)
        }
        if strings.Contains(errorMsg, podAlreadyExistsErrorCode) {
            glog.Warningf("trying to resubmit an already submitted SparkApplication %s/%s", submission.namespace, submission.name)
            return false, nil
        }
        if errorMsg != "" {
            return false, fmt.Errorf("failed to run spark-submit for SparkApplication %s/%s: %s", submission.namespace, submission.name, errorMsg)
        }
        return false, fmt.Errorf("failed to run spark-submit for SparkApplication %s/%s: %v", submission.namespace, submission.name, err)
    }
    return true, nil
}

Spark Operator State Machine

The operator defines a lifecycle state machine for SparkApplication. A typical successful job progresses through:

New -> Submitted -> Running -> Succeeding -> Completed

If a failure occurs, the operator retries up to the configured limit; exceeding the limit marks the job as failed. The operator also watches pod status, automatically recreating driver pods when they are evicted, providing built‑in fault‑tolerance.

Debugging Tips

Inspect the generated spark-submit arguments in the operator pod logs (default glog level 2).

Check events attached to the SparkApplication resource, which record state transitions.

Use the exposed Spark UI Service or Ingress to view the Spark UI.

Conclusion

Using Spark Operator is the most robust way to run Spark on Kubernetes. Compared with raw spark-submit, it adds automatic recovery, monitoring via Prometheus, UI exposure, and a declarative CRD‑based workflow that simplifies lifecycle management for both one‑off and scheduled Spark jobs.