Deploying the Spark Operator, Building Custom Images, Running Structured Streaming through Helm, Integrating Airflow, and Enforcing Multi-Tenant Resource Boundaries

Running an isolated Spark job on Kubernetes is relatively straightforward. Running dozens of long-lived streaming applications, interactive Spark Connect sessions, machine-learning workloads, and Airflow-triggered batch jobs on the same Kubernetes cluster is not.

At that point, Spark is no longer merely an application runtime. It becomes a platform with its own control plane, workload APIs, scheduling policies, images, observability model, security boundaries, and release process.

This article describes an architecture for operating Apache Spark as a multi-tenant Kubernetes platform. It covers:

• Installing the Kubeflow Spark Operator through a wrapper Helm chart
• Operating the controller and admission webhook in a highly available configuration
• Watching multiple workload namespaces from one operator deployment
• Integrating the Volcano batch scheduler
• Creating a reusable Spark and PySpark base image
• Packaging long-running Structured Streaming applications with Helm
• Consuming Kafka and writing to an Iceberg-backed object store
• Managing checkpoints and application state
• Submitting bounded Spark jobs through Apache Airflow
• Separating critical, standard, interactive, and machine-learning workloads
• Enforcing namespace quotas, limits, priorities, RBAC, and network boundaries
• Collecting operator, driver, executor, and application-level metrics
• Avoiding common operational and security mistakes

The examples intentionally use generic registry names, service endpoints, namespace names, and secret references.

1. The Platform Architecture

The platform consists of two distinct layers:

1. The Spark control plane
2. The Spark workload plane

The control plane runs the Spark Operator controller, admission webhook, CRDs, metrics endpoints, and supporting RBAC.

The workload plane contains the driver and executor pods created for each Spark application.

A simplified architecture looks like this:

                                      ┌──────────────────────────────┐
                                      │ CI/CD or GitOps Controller   │
                                      │ Helm, Argo CD, Flux, etc.    │
                                      └──────────────┬───────────────┘
                                                     │
                                                     │ applies
                                                     ▼
┌────────────────────┐                ┌──────────────────────────────┐
│ Apache Airflow     │                │ SparkApplication CR          │
│                    │───────────────▶│ ScheduledSparkApplication CR │
│ Bounded batch jobs │                │ SparkConnect CR              │
└────────────────────┘                └──────────────┬───────────────┘
                                                     │ watched by
                                                     ▼
                                      ┌──────────────────────────────┐
                                      │ Kubeflow Spark Operator      │
                                      │                              │
                                      │ Controller replicas          │
                                      │ Admission webhook            │
                                      │ Metrics endpoint             │
                                      └──────────────┬───────────────┘
                                                     │ creates
                         ┌───────────────────────────┴───────────────────────────┐
                         │                                                       │
                         ▼                                                       ▼
              ┌────────────────────┐                                  ┌────────────────────┐
              │ Spark Driver Pod   │                                  │ Volcano PodGroup   │
              │                    │                                  │ and queue metadata │
              │ Creates executors  │                                  └────────────────────┘
              └──────────┬─────────┘
                         │
          ┌──────────────┼──────────────┐
          ▼              ▼              ▼
    ┌──────────┐   ┌──────────┐   ┌──────────┐
    │Executor 1│   │Executor 2│   │Executor N│
    └──────────┘   └──────────┘   └──────────┘
          │              │              │
          └──────────────┼──────────────┘
                         ▼
      Kafka → Structured Streaming → Iceberg/Object Storage
                         │
                         ▼
                  Durable Checkpoint

Spark already supports Kubernetes as a native scheduler backend. A Spark driver creates executor pods through the Kubernetes API, communicates with them, and removes them when they are no longer required. The driver therefore needs a Kubernetes service account with permission to create, list, watch, and delete the relevant resources in its own namespace.

The Spark Operator adds a declarative API on top of this mechanism. Instead of running a procedural spark-submit command from an administrator workstation, we create a SparkApplication resource and allow the operator to reconcile the desired state.

That changes the operational model significantly:

spark-submit command
        ↓
imperative, client-owned lifecycle

becomes:

SparkApplication manifest
        ↓
declarative, Kubernetes-owned lifecycle

The operator creates and monitors the driver, tracks application status, applies restart policies, exposes metrics, and updates the custom resource with the observed state.

2. Pinning the Platform Versions

The implementation described here uses:

Apache Spark:           3.5.6
Java:                   17
Python:                 3.11
Spark Operator chart:   2.3.0
SparkApplication API:   sparkoperator.k8s.io/v1beta2
SparkConnect API:       sparkoperator.k8s.io/v1alpha1

Version 2.3.0 of the operator introduced the SparkConnect CRD and an optional Helm hook for upgrading CRDs during installation and upgrades.

These versions should be treated as a compatibility unit.

A Spark image is not simply "Spark plus Python." Its effective compatibility surface includes:

Spark version
Scala binary version
Java version
Python version
Hadoop client version
Kafka connector version
Iceberg runtime version
Hadoop AWS module
AWS SDK version
JMX exporter version
Kubernetes client behavior

An image should therefore be promoted as one immutable artifact:

registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4

Using a mutable tag such as latest makes debugging and rollback unnecessarily difficult:

image:
  repository: registry.example.com/data-platform/stream-processor
  tag: "1.8.3"

Part I: Deploying the Spark Operator

3. Why Use a Wrapper Helm Chart?

Rather than installing the upstream Spark Operator chart directly, we wrap it in an internal platform chart:

spark-platform/
├── Chart.yaml
├── Chart.lock
├── charts/
│   └── spark-operator-2.3.0.tgz
├── templates/
│   └── podmonitors.yaml
├── values.yaml
└── values-production.yaml

The upstream chart remains a dependency:

apiVersion: v2
name: spark-platform
description: Production Spark platform for Kubernetes
type: application
version: 0.1.0
appVersion: "2.3.0"
dependencies:
  - name: spark-operator
    repository: https://kubeflow.github.io/spark-operator
    version: 2.3.0

The wrapper chart serves several purposes:

• Pins the exact operator chart version
• Centralizes registry mirroring
• Applies high-availability settings
• Defines watched workload namespaces
• Enables CRD upgrades
• Configures the batch scheduler
• Creates organization-wide monitoring resources
• Establishes standard labels and security settings
• Separates generic defaults from production-specific values

The wrapper must not copy the upstream chart templates. Keeping it as a dependency makes upstream upgrades easier to review and reduces template drift.

Run:

# When dependency declarations change:
helm dependency update ./spark-platform

# In CI or during reproducible builds:
helm dependency build ./spark-platform

Commit both Chart.lock and the packaged dependency if deployments operate in an environment without reliable outbound internet access.

4. High-Availability Operator Configuration

A production operator configuration can look like this:

spark-operator:
  commonLabels:
    app.kubernetes.io/part-of: spark-platform

  image:
    registry: registry.example.com

  hook:
    upgradeCrd: true
    image:
      registry: registry.example.com

  controller:
    replicas: 3

    batchScheduler:
      enable: true
      kubeSchedulerNames:
        - default-scheduler
        - volcano
      default: volcano

    affinity:
      podAntiAffinity:
        requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchLabels:
                app.kubernetes.io/component: controller
            topologyKey: kubernetes.io/hostname

    # The chart injects the controller selector automatically.
    topologySpreadConstraints:
      - maxSkew: 1
        topologyKey: kubernetes.io/hostname
        whenUnsatisfiable: DoNotSchedule

    podDisruptionBudget:
      enable: true
      minAvailable: 2

    resources:
      requests:
        cpu: "500m"
        memory: 512Mi
      limits:
        cpu: "2"
        memory: 2Gi

  webhook:
    replicas: 3

    affinity:
      podAntiAffinity:
        requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchLabels:
                app.kubernetes.io/component: webhook
            topologyKey: kubernetes.io/hostname

    # The chart injects the webhook selector automatically.
    topologySpreadConstraints:
      - maxSkew: 1
        topologyKey: kubernetes.io/hostname
        whenUnsatisfiable: DoNotSchedule

    podDisruptionBudget:
      enable: true
      minAvailable: 2

    resources:
      requests:
        cpu: "500m"
        memory: 512Mi
      limits:
        cpu: "2"
        memory: 2Gi

  prometheus:
    metrics:
      enable: true
      port: 8080
      portName: metrics
      endpoint: /metrics
      jobStartLatencyBuckets: "30,60,90,120,180,240,300"

    podMonitor:
      create: true
      jobLabel: spark-operator
      podMetricsEndpoint:
        scheme: http
        interval: 15s

4.1 Why Three Replicas?

Running three replicas does not mean three controllers independently submit every job. Leader election ensures that one controller acts as the active reconciler while the others are available for failover.

The controller is part of the platform control plane. If it is unavailable:

• New applications will not be reconciled
• Restart policies will not be processed
• Status fields can become stale
• Submitted custom resources may remain idle
• Spark UI services and ingress resources may not be generated

The webhook is also operationally important when pod mutation is used. If the webhook is configured with a failure policy that blocks requests, webhook unavailability can prevent Spark pods from being created.

For both components, anti-affinity, topology spreading, and PodDisruptionBudgets protect against:

• A single-node failure
• A rolling node upgrade
• Cluster autoscaler node removal
• Voluntary maintenance
• Simultaneous replica eviction

4.2 A Subtle Scheduling Trap

Required pod anti-affinity combined with:

whenUnsatisfiable: DoNotSchedule

requires at least three eligible Kubernetes nodes for three replicas.

In a smaller cluster, this configuration can make the third controller or webhook replica permanently Pending.

For smaller environments, use preferred anti-affinity:

preferredDuringSchedulingIgnoredDuringExecution:
  - weight: 100
    podAffinityTerm:
      topologyKey: kubernetes.io/hostname
      labelSelector:
        matchLabels:
          app.kubernetes.io/name: spark-operator

High availability settings must reflect the physical topology that actually exists.

5. Managing CRDs during Helm Upgrades

Helm installs CRDs from a chart's crds/ directory but does not generally upgrade them as part of a normal release upgrade. This can leave the controller and CRD schemas at incompatible versions.

Operator 2.3.0 introduced:

spark-operator:
  hook:
    upgradeCrd: true

This creates a pre-install and pre-upgrade job that applies updated CRDs server-side.

The upgrade path should still be reviewed carefully:

helm template spark-platform ./spark-platform \
  --namespace spark-operator \
  -f values.yaml \
  -f values-production.yaml > rendered.yaml
kubectl diff -f rendered.yaml

Before upgrading:

kubectl get crd sparkapplications.sparkoperator.k8s.io -o yaml \
  > backup-sparkapplications-crd.yaml

kubectl get crd scheduledsparkapplications.sparkoperator.k8s.io -o yaml \
  > backup-scheduledsparkapplications-crd.yaml

kubectl get crd sparkconnects.sparkoperator.k8s.io -o yaml \
  > backup-sparkconnects-crd.yaml

Also back up the actual custom resources, not only their schemas:

kubectl get sparkapplications.sparkoperator.k8s.io -A -o yaml \
  > backup-sparkapplications.yaml

kubectl get scheduledsparkapplications.sparkoperator.k8s.io -A -o yaml \
  > backup-scheduledsparkapplications.yaml

kubectl get sparkconnects.sparkoperator.k8s.io -A -o yaml \
  > backup-sparkconnects.yaml

CRDs are cluster-scoped resources. A chart release in one namespace can therefore affect Spark applications in every namespace.

Add --include-crds when rendering for CRD review:

helm template spark-platform ./spark-platform \
  --namespace spark-operator \
  --include-crds \
  -f values.yaml \
  -f values-production.yaml \
  > rendered.yaml

This is another reason the operator chart should be managed by a platform team rather than independently installed by each workload team.

6. Watching Multiple Spark Namespaces

The operator itself runs in a dedicated namespace:

spark-operator

Spark workloads run elsewhere:

spark-streaming-critical
spark-streaming-standard
spark-batch
spark-ml
spark-interactive

Production values can define the namespaces watched by the operator:

spark-operator:
  controller:
    uiIngress:
      enable: true
      urlFormat: "https://spark-ui.example.com/{{$appNamespace}}/{{$appName}}"

  spark:
    jobNamespaces:
      - spark-streaming-critical
      - spark-streaming-standard
      - spark-batch
      - spark-ml
      - spark-interactive

    serviceAccount:
      create: true
      name: spark
      automountServiceAccountToken: true

There are two separate identities involved:

1. The operator service account, used by the controller
2. The Spark driver service account, used by driver pods

The driver service account must exist in every workload namespace. A service account is namespaced; creating spark in spark-operator does not create spark in spark-batch.

A minimal namespaced RBAC model is preferable to granting the driver a broad cluster role:

apiVersion: v1
kind: ServiceAccount
metadata:
  name: spark
  namespace: spark-batch
automountServiceAccountToken: true
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: spark-driver
  namespace: spark-batch
rules:
  - apiGroups: [""]
    resources:
      - pods
      - pods/log
      - services
      - configmaps
      - persistentvolumeclaims
    verbs:
      - create
      - get
      - list
      - watch
      - patch
      - update
      - delete
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: spark-driver
  namespace: spark-batch
subjects:
  - kind: ServiceAccount
    name: spark
    namespace: spark-batch
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: spark-driver

Spark drivers normally create executor pods in the same namespace, so a namespaced Role is sufficient in most deployments.

Part II: Resource Isolation and Workload-Aware Scheduling

7. Separating Spark Workloads by Namespace

A production Spark platform should not place every driver and executor in one unrestricted namespace.

Different Spark workloads have different operational characteristics:

This separation allows the platform to apply different:

• Resource quotas
• Limit ranges
• Service accounts
• RBAC policies
• Priority classes
• Network policies
• Secret access rules
• Monitoring and alerting thresholds

A namespace is not merely an organizational folder. It is an enforceable resource and security boundary.

The Spark Operator must explicitly watch each workload namespace:

spark-operator:
  spark:
    jobNamespaces:
      - spark-streaming-critical
      - spark-streaming-standard
      - spark-batch
      - spark-machine-learning
      - spark-interactive

The Spark driver service account must also exist in every watched namespace. Kubernetes service accounts are namespaced resources; creating spark in the operator namespace does not create it in the application namespaces.

8. Enforcing Namespace Resource Budgets

Kubernetes ResourceQuota establishes the hard resource ceiling for a namespace.

For example:

apiVersion: v1
kind: ResourceQuota
metadata:
  name: spark-standard-budget
  namespace: spark-streaming-standard
spec:
  hard:
    requests.cpu: "50"
    limits.cpu: "50"
    requests.memory: 200Gi
    limits.memory: 200Gi

    pods: "100"
    services: "40"
    configmaps: "100"
    secrets: "100"

    count/sparkapplications.sparkoperator.k8s.io: "30"

The quota applies to the complete Spark application:

Application CPU request =
    driver CPU request
    + executor count × executor CPU request

Memory calculations must include more than the configured JVM heap:

Executor pod memory ≈
    spark.executor.memory
    + spark.executor.memoryOverhead
    + spark.executor.pyspark.memory
    + off-heap memory

For example:

driver:
  coreRequest: "2"
  coreLimit: "2"
  memory: 2g
  memoryOverhead: 1g
executor:
  instances: 4
  coreRequest: "2"
  coreLimit: "2"
  memory: 4g
  memoryOverhead: 2g

The approximate quota footprint is:

CPU:
    driver:       2
    executors:    4 × 2 = 8
    total:        10 CPU
Memory:
    driver:       2 GiB + 1 GiB = 3 GiB
    executors:    4 × (4 GiB + 2 GiB) = 24 GiB
    total:        approximately 27 GiB

If the application is expected to reach maxExecutors, the namespace quota and physical cluster capacity must be large enough to admit that executor fleet. Otherwise, Spark can remain healthy while scaling stops below the configured maximum.

dynamicAllocation:
  enabled: true
  minExecutors: 2
  initialExecutors: 4
  maxExecutors: 20

Dynamic allocation does not bypass Kubernetes quotas. Spark may request additional executors, but Kubernetes can reject them once the namespace reaches its hard ceiling.

A LimitRange should also constrain malformed or unexpectedly large containers:

apiVersion: v1
kind: LimitRange
metadata:
  name: spark-container-policy
  namespace: spark-streaming-standard
spec:
  limits:
    - type: Container
      min:
        cpu: "100m"
        memory: 256Mi
      max:
        cpu: "8"
        memory: 32Gi
      defaultRequest:
        cpu: "500m"
        memory: 1Gi
      default:
        cpu: "1"
        memory: 2Gi

Spark manifests should still declare explicit resources. LimitRange defaults are a safety mechanism, not a substitute for capacity planning.

9. What the Default Kubernetes Scheduler Lacks for Spark

The default Kubernetes scheduler is highly effective at placing individual pods onto nodes. Its fundamental scheduling unit, however, is a pod — not an entire distributed Spark application.

A Spark application usually requires:

one driver
+
a minimum number of executors

The default scheduler can produce this state:

Driver:       Running
Executor 1:   Running
Executor 2:   Pending
Executor 3:   Pending
Executor 4:   Pending

From Kubernetes's perspective, some pods were scheduled successfully.

From Spark's perspective, the application may be unable to achieve useful throughput.

This is particularly harmful for:

• Streaming applications with latency targets
• Large batch jobs
• Stateful processing
• Machine-learning preprocessing
• Interactive Spark sessions
• Jobs that require several executors before meaningful progress begins

Kubernetes provides several useful controls, but they solve different problems:

What ResourceQuota Answers:

What is the maximum this namespace may consume?

It does not guarantee that a Spark application receives enough resources to start usefully.

What PriorityClass Answers:

Which pending pod should be considered first?

It does not inherently treat a driver and its executors as one scheduling unit.

What Pod affinity Answers:

Where should related pods be placed?

It does not create job-level admission or queue fairness.

What Kubernetes HPA Answers:

Should the replica count of this Kubernetes controller change?

It does not understand Spark executor demand and should not be used to scale Spark executors.

The missing concept is workload-level admission:

Do enough resources exist to start the distributed application at its minimum useful size?

10. Using Volcano as Spark's Workload-Aware Scheduler

We use Volcano to add distributed-workload awareness to Spark scheduling.

Volcano introduces two concepts that are especially relevant to Spark:

PodGroup

A PodGroup represents the pods belonging to one distributed workload.

For Spark, it can represent:

driver
+
minimum executor fleet

Volcano can wait until the PodGroup's minimum resource requirements can be satisfied instead of partially admitting an application that cannot make useful progress.

Queue

A Volcano Queue groups distributed workloads and controls how scheduler capacity is shared between workload classes or tenants.

Queue configuration is discussed in the companion Volcano article. In the Spark platform, the important point is that each application can select Volcano and declare its minimum useful resource size.

A SparkApplication enables Volcano with:

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: example-stream
  namespace: spark-streaming-standard
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster

  sparkVersion: "3.5.6"
  image: registry.example.com/data-platform/example-stream:1.0.0
  mainApplicationFile: local:///opt/spark/app/runner.py
  mode: cluster
  sparkVersion: "3.5.6"
  batchScheduler: volcano
  batchSchedulerOptions:
    priorityClassName: spark-standard
    resources:
      cpu: "6"
      memory: 16Gi
  driver:
    serviceAccount: spark
    cores: 2
    coreRequest: "2"
    coreLimit: "2"
    memory: 2g
    memoryOverhead: 1g
  executor:
    instances: 2
    cores: 2
    coreRequest: "2"
    coreLimit: "2"
    memory: 5g
    memoryOverhead: 1g

The declared minimum is approximately:

CPU:
    driver:       2
    executors:    2 × 2 = 4
    total:        6 CPU
Memory:
    driver:       2 GiB + 1 GiB = 3 GiB
    executors:    2 × (5 GiB + 1 GiB) = 12 GiB
    total:        approximately 15 GiB

We round the scheduling requirement to:

resources:
  cpu: "6"
  memory: 16Gi

This tells the batch scheduler:

Do not admit this application until sufficient capacity exists for the driver and its minimum useful executor fleet.

A queue can also be specified:

batchSchedulerOptions:
  queue: spark-streaming-standard

but only when a Volcano Queue resource with that name actually exists.

Our initial scheduling scaffolding created namespaces, ResourceQuotas, and PriorityClasses. Per-tenant Volcano Queue resources are a separate scheduling-layer concern and must not be implied merely by adding a queue field to the SparkApplication.

The complete design of queues, PodGroups, gang scheduling, borrowing, reclamation, preemption, scheduler high availability, and non-Spark integrations is covered in the companion article which I will share later and link in here.

Part III: Building the Spark Runtime Image

11. Why Build a Custom Spark Base Image?

The upstream Apache Spark image provides Spark, Java, and the Kubernetes entrypoint. A platform image extends it with the dependencies shared by many applications:

• A controlled Python version
• Python package configuration
• Kafka and Iceberg runtime JARs
• S3A dependencies
• Logging configuration
• Metrics exporter
• Certificate authorities
• Runtime security defaults

The application image then extends the platform image and adds only application code and application-specific Python packages.

Apache Spark upstream image
            ↓
Platform Spark base image
            ↓
Application-specific image

This reduces duplicated build logic and ensures that drivers and executors use the same runtime.

12. Use the Python Runtime Included in the Spark Image

The official Apache Spark Python image already contains Python and the PySpark runtime required by drivers and executors. We therefore do not need to install another Python version, replace /usr/bin/python3, or create a separate virtual environment in the platform image.

Keeping the standard Python runtime also reduces the number of components that must be patched, tested, and kept compatible with Spark.

Our platform image only adds the shared JAR files and logging configuration required by the applications:

ARG SPARK_VERSION=3.5.6 
FROM spark:${SPARK_VERSION}-scala2.12-java17-python3-ubuntu 
USER root 
ENV TZ=UTC \ 
    PYSPARK_PYTHON=/usr/bin/python3 \ 
    PYSPARK_DRIVER_PYTHON=/usr/bin/python3 

WORKDIR /opt/spark/work-dir 

COPY --chown=spark:spark storage/jars/ /opt/spark/jars/ 
COPY --chown=spark:spark log4j2.properties /opt/spark/conf/log4j2.properties 

USER spark

PYSPARK_PYTHON controls the Python executable used by PySpark workers, while PYSPARK_DRIVER_PYTHON controls the executable used by the driver.

Because drivers and executors use the same image, both sides receive the same Python runtime, Spark libraries, connector JARs, and logging configuration.

Application-specific Python dependencies are installed later in the application image:

ARG SPARK_BASE_IMAGE=registry.example.com/data-platform/spark-runtime:3.5.6

FROM ${SPARK_BASE_IMAGE}

USER root

COPY requirements.txt /tmp/requirements.txt

RUN python3 -m pip install \
      --no-cache-dir \
      --requirement /tmp/requirements.txt && \
    rm -f /tmp/requirements.txt

COPY --chown=spark:spark app/ /opt/spark/app/

USER spark

This creates a straightforward image hierarchy:

Official Spark Python image
          ↓
Shared platform Spark image
          ↓
Application-specific Spark image

The platform image owns Spark-wide dependencies and configuration. The application image owns application code and Python packages.

13. Use UTC inside Containers

Setting a local timezone in the runtime image creates unnecessary ambiguity:

• Kafka event timestamps may use UTC
• Airflow may display another timezone
• Spark SQL session timezone may differ
• Object-store partitions may be date-based
• The host node may use UTC
• JVM and Python timezone data can differ

Use:

ENV TZ=UTC

and configure the business timezone explicitly where required:

sparkConf:
  spark.sql.session.timeZone: UTC

Convert to business-local time at the data-model boundary, not implicitly through the container operating system.

Environments that use an internal Python package proxy can configure pip during the application-image build without placing repository credentials

14. JAR Compatibility

The platform image may include:

iceberg-spark-runtime-3.5_2.12-<version>.jar
spark-sql-kafka-0-10_2.12-3.5.6.jar
spark-token-provider-kafka-0-10_2.12-3.5.6.jar
kafka-clients-<version>.jar
hadoop-aws-<matching-hadoop-version>.jar
aws-java-sdk-bundle-<compatible-version>.jar
jmx_prometheus_javaagent-<version>.jar

The Scala suffix matters:

_2.12

An artifact compiled for Scala 2.13 is not interchangeable with one compiled for Scala 2.12.

The Hadoop AWS and AWS SDK versions must also be compatible with the Hadoop libraries bundled into the Spark distribution. Arbitrarily upgrading only hadoop-aws often produces runtime failures such as:

NoSuchMethodError
ClassNotFoundException
NoClassDefFoundError

Validate the final image:

docker run --rm \
  registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4 \
  sh -lc '
    java -version
    python --version
    pyspark --version
    ls -1 /opt/spark/jars | grep -E "iceberg|kafka|hadoop-aws|prometheus"
  '

15. Logging Configuration

A platform-level Log4j configuration can reduce noise:

status = WARN
name = SparkLog4j2
appender.console.type = Console
appender.console.name = console
appender.console.target = SYSTEM_ERR
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d{ISO8601} %p %c: %m%n
rootLogger.level = ${sys:spark.log.level:-WARN}
rootLogger.appenderRefs = console
rootLogger.appenderRef.console.ref = console
logger.spark.name = org.apache.spark
logger.spark.level = WARN
logger.jetty.name = org.sparkproject.jetty
logger.jetty.level = WARN
logger.hadoop.name = org.apache.hadoop
logger.hadoop.level = WARN

A global WARN level is appropriate for ordinary production operation, but it can hide valuable diagnostics during incident response.

Support an override:

sparkConf:
  spark.driver.extraJavaOptions: >-
    -Dlog4j.configurationFile=/opt/spark/conf/log4j2.properties
    -Dlog4j2.level={{ .Values.logging.level }}
  spark.executor.extraJavaOptions: >-
    -Dlog4j.configurationFile=/opt/spark/conf/log4j2.properties
    -Dlog4j2.level={{ .Values.logging.level }}

Do not rebuild the image merely to change the log level during an incident.

16. Building the Application Image

Each stream processor extends the platform image:

ARG SPARK_BASE_IMAGE=registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4
FROM ${SPARK_BASE_IMAGE}
USER root
COPY requirements.txt /tmp/requirements.txt
RUN python3 -m pip install \
      --no-cache-dir \
      --requirement /tmp/requirements.txt && \
    rm -f /tmp/requirements.txt
COPY app/ /opt/spark/app/
RUN chown -R 185:0 /opt/spark/app && \
    chmod -R g=u /opt/spark/app
USER 185

The entrypoint remains the Spark image entrypoint. The SparkApplication references the Python application with:

mainApplicationFile: local:///opt/spark/app/runner.py

The local:// URI indicates that the file already exists inside the driver and executor image. Spark documents this model for dependencies baked into custom images.

Part IV: Designing the Structured Streaming Application

17. Configuration Boundaries

Application configuration should be divided into three categories:

Non-secret application configuration

Examples:

SPARK_APP_NAME
KAFKA_TOPIC
CHECKPOINT_LOCATION
ICEBERG_CATALOG
ICEBERG_DATABASE
ICEBERG_TABLE
TRIGGER_INTERVAL
MAX_OFFSETS_PER_TRIGGER

These can live in Helm values or a ConfigMap.

Secret configuration

Examples:

KAFKA_SASL_USERNAME
KAFKA_SASL_PASSWORD
OBJECT_STORE_ACCESS_KEY
OBJECT_STORE_SECRET_KEY
TRUSTSTORE_PASSWORD
KEYSTORE_PASSWORD

These must come from an existing Kubernetes Secret or a secret-management controller like vault.

Binary secret material

Examples:

Kafka truststore
Kafka keystore
Private CA bundle
Client certificate
Private key

These should be mounted as files from a Kubernetes Secret, not encoded into Helm values.

18. A Simplified Streaming Application

The following example reads from Kafka and writes to an Iceberg table:

import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StringType, StructField, StructType
def required_env(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Required environment variable is missing: {name}")
    return value

schema = StructType(
    [
        StructField("event_id", StringType(), nullable=False),
        StructField("event_type", StringType(), nullable=False),
        StructField("event_time", StringType(), nullable=False),
        StructField("payload", StringType(), nullable=True),
    ]
)
app_name = required_env("SPARK_APP_NAME")
kafka_servers = required_env("KAFKA_BOOTSTRAP_SERVERS")
kafka_topic = required_env("KAFKA_TOPIC")
checkpoint = required_env("SPARK_CHECKPOINT_LOCATION")
catalog = required_env("ICEBERG_CATALOG")
database = required_env("ICEBERG_DATABASE")
table = required_env("ICEBERG_TABLE")
spark = (
    SparkSession.builder
    .appName(app_name)
    .getOrCreate()
)
source = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", kafka_servers)
    .option("subscribe", kafka_topic)
    .option("startingOffsets", "latest")
    .option(
        "kafka.security.protocol",
        required_env("KAFKA_SECURITY_PROTOCOL"),
    )
    .option(
        "kafka.ssl.truststore.location",
        required_env("KAFKA_SSL_TRUSTSTORE_LOCATION"),
    )
    .option(
        "kafka.ssl.truststore.password",
        required_env("KAFKA_SSL_TRUSTSTORE_PASSWORD"),
    )
    .option(
        "kafka.sasl.mechanism",
        required_env("KAFKA_SASL_MECHANISM"),
    )
    .option(
        "kafka.sasl.jaas.config",
        (
            "org.apache.kafka.common.security.scram."
            "ScramLoginModule required "
            f'username="{required_env("KAFKA_SASL_USERNAME")}" '
            f'password="{required_env("KAFKA_SASL_PASSWORD")}";'
        ),
    )
    .load()
)
events = (
    source
    .selectExpr(
        "CAST(key AS STRING) AS kafka_key",
        "CAST(value AS STRING) AS kafka_value",
        "topic",
        "partition",
        "offset",
        "timestamp AS kafka_timestamp",
    )
    .withColumn("event", from_json(col("kafka_value"), schema))
    .select(
        "kafka_key",
        "topic",
        "partition",
        "offset",
        "kafka_timestamp",
        "event.*",
    )
)
query = (
    events.writeStream
    .format("iceberg")
    .outputMode("append")
    .option(
        "checkpointLocation",
        checkpoint,
    )
    .trigger(processingTime=os.getenv("TRIGGER_INTERVAL", "30 seconds"))
    .toTable(f"{catalog}.{database}.{table}")
)
query.awaitTermination()

The production version should add:

• Schema validation and dead-letter handling
• Event-time parsing
• Watermarks for stateful operations
• Duplicate detection
• Corrupt record metrics
• Controlled offset ingestion
• Batch-level audit records
• A strategy for poison messages
• Query progress logging
• Explicit shutdown behavior

Structured Streaming tracks source offsets and query progress through checkpoint data and write-ahead logs. End-to-end exactly-once behavior also depends on replayable sources and sinks that correctly handle retries or duplicate writes.

Exactly-once is therefore a property of the whole pipeline, not a magic setting on the Spark session.

19. Checkpoint Design

Every independently deployed query needs a unique checkpoint location:

s3a://spark-checkpoints/<environment>/<application>/<query-name>/

For example:

s3a://spark-checkpoints/production/customer-events/main/

Do not reuse the same checkpoint across:

• Two simultaneously running applications
• Different source topics
• Different target tables
• Different query plans
• Development and production
• Blue and green deployments running concurrently

Several Spark SQL and Structured Streaming settings cannot be safely changed when restarting from an existing checkpoint. Spark explicitly documents configuration restrictions for restarted streaming queries.

Treat the checkpoint as application state.

Changing the Helm release name does not create a new logical stream if the checkpoint remains the same.

Conversely, Changing the checkpoint path creates a new Structured Streaming query state. With no existing offset log, Spark calculates its initial Kafka position from startingOffsets. Depending on startingOffsets, it may replay old data or begin at the newest Kafka offsets.

20.1 Deployment Strategies

For stateless append-only streams, a rolling replacement may be possible.

For stateful streams, a safer deployment process is:

1. Stop the old application
2. Confirm the query has terminated
3. Deploy the new application
4. Reuse the checkpoint only if the new query is checkpoint-compatible
5. Observe the first several micro-batches

Do not run blue and green copies against the same checkpoint.

If parallel validation is required, use:

separate Kafka consumer/query identity
separate checkpoint
separate validation output table

Part V: Packaging the Streaming Service with Helm

20. Chart Structure

A reusable streaming chart might be:

stream-processor/
├── Chart.yaml
├── values.yaml
├── values/
│   ├── domain-a/
│   │   └── values.yaml
│   ├── domain-b/
│   │   └── values.yaml
│   └── domain-c/
│       └── values.yaml
└── templates/
    ├── _helpers.tpl
    ├── serviceaccount.yaml
    ├── sparkapplication.yaml
    ├── podmonitor.yaml
    └── networkpolicy.yaml

Different deployments share one chart but provide different values:

helm upgrade --install domain-a-stream ./stream-processor \
  --namespace spark-streaming-standard \
  --create-namespace \
  -f values.yaml \
  -f values/domain-a/values.yaml

The chart describes a Spark application, not a normal web service.

Generic templates copied from a standard Deployment chart should usually be removed:

deployment.yaml    not applicable
hpa.yaml           generally not applicable
service.yaml       usually unnecessary
ingress.yaml       usually unnecessary

Kubernetes HPA does not understand Spark executor demand. Use Spark dynamic allocation for executor scaling.

A service or ingress may still be useful for a driver UI, metrics, or Spark Connect, but it should be created intentionally rather than inherited from a generic chart scaffold.

21. Base Values

image:
  repository: registry.example.com/data-platform/stream-processor
  tag: "1.0.0"
  pullPolicy: IfNotPresent
  imagePullSecrets:
    - private-registry  
sparkVersion: "3.5.6"
mainApplicationFile: local:///opt/spark/app/runner.py
serviceAccount:
  create: true
  name: spark
scheduler:
  name: volcano
  # Leave empty when the application should use Volcano's default queue.
  # Set this only when a matching Volcano Queue resource exists.
  queue: ""
  priorityClassName: spark-standard
  # Minimum resource set required before the distributed application
  # should be admitted by the batch scheduler.
  minimumResources:
    cpu: "3"
    memory: 10Gi
restartPolicy:
  type: OnFailure
  onFailureRetries: 10
  onFailureRetryInterval: 30
  onSubmissionFailureRetries: 10
  onSubmissionFailureRetryInterval: 30
driver:
  cores: 1
  coreRequest: "1"
  coreLimit: "2"
  memory: 2g
  memoryOverhead: 1g
executor:
  cores: 1
  coreRequest: "1"
  coreLimit: "2"
  instances: 2
  memory: 3g
  memoryOverhead: 1g
dynamicAllocation:
  enabled: false
  initialExecutors: 2
  minExecutors: 1
  maxExecutors: 10
  shuffleTrackingEnabled: true
  shuffleTrackingTimeout: 300000
application:
  env:
    SPARK_APP_NAME: example-stream
    SPARK_CHECKPOINT_LOCATION: s3a://spark-checkpoints/example-stream/main
    KAFKA_BOOTSTRAP_SERVERS: kafka-bootstrap.data-platform.svc:9093
    KAFKA_TOPIC: example-events
    KAFKA_SECURITY_PROTOCOL: SASL_SSL
    KAFKA_SASL_MECHANISM: SCRAM-SHA-512
    OBJECT_STORE_ENDPOINT: http://object-store.data-platform.svc:9000
    HIVE_METASTORE_URI: thrift://hive-metastore.catalog.svc:9083
    ICEBERG_CATALOG: production
    ICEBERG_DATABASE: events
    ICEBERG_TABLE: example_events
existingSecret: example-stream-secrets
kafkaTlsSecret: example-stream-kafka-tls
monitoring:
  enabled: true
  jmxExporterJar: /opt/spark/jars/jmx_prometheus_javaagent-1.1.0.jar
  port: 8090
  portName: prometheus
sparkConf:
  spark.sql.adaptive.enabled: "true"
  spark.sql.adaptive.coalescePartitions.enabled: "true"
  spark.sql.session.timeZone: UTC
  spark.sql.extensions: org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
  spark.sql.catalog.production: org.apache.iceberg.spark.SparkCatalog
  spark.sql.catalog.production.type: hive
  spark.sql.catalogImplementation: hive
  spark.hadoop.fs.s3a.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
  spark.hadoop.fs.s3a.path.style.access: "true"
  spark.hadoop.fs.s3a.connection.ssl.enabled: "false"
  spark.task.maxFailures: "8"
  spark.stage.maxConsecutiveAttempts: "8"
  spark.kubernetes.executor.deleteOnTermination: "true"

Do not include a secrets: dictionary containing literal values.

Helm release history is not a secret store. Depending on the storage driver, rendered release values can be persisted in cluster Secrets and retrieved by users with sufficient access.

22. SparkApplication Template

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: {{ include "stream-processor.fullname" . }}
  namespace: {{ .Release.Namespace }}
  labels:
    {{- include "stream-processor.labels" . | nindent 4 }}
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
sparkVersion: {{ .Values.sparkVersion | quote }}
  image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}"
  imagePullPolicy: {{ .Values.image.pullPolicy }}
  {{- with .Values.image.imagePullSecrets }}
  imagePullSecrets:
    {{- toYaml . | nindent 4 }}
  {{- end }}
  mainApplicationFile: {{ .Values.mainApplicationFile | quote }}
  restartPolicy:
    type: {{ .Values.restartPolicy.type }}
    onFailureRetries: {{ .Values.restartPolicy.onFailureRetries }}
    onFailureRetryInterval: {{ .Values.restartPolicy.onFailureRetryInterval }}
    onSubmissionFailureRetries: {{ .Values.restartPolicy.onSubmissionFailureRetries }}
    onSubmissionFailureRetryInterval: {{ .Values.restartPolicy.onSubmissionFailureRetryInterval }}
  batchScheduler: {{ .Values.scheduler.name }}
  batchSchedulerOptions:
    {{- with .Values.scheduler.queue }}
    queue: {{ . | quote }} 
    {{- end }}
    priorityClassName: {{ .Values.scheduler.priorityClassName | quote }}
    resources:
      memory: {{ .Values.scheduler.minimumResources.memory | quote }}
      cpu: {{ .Values.scheduler.minimumResources.cpu | quote }}
  {{- if .Values.dynamicAllocation.enabled }}
  dynamicAllocation:
    enabled: true
    initialExecutors: {{ .Values.dynamicAllocation.initialExecutors }}
    minExecutors: {{ .Values.dynamicAllocation.minExecutors }}
    maxExecutors: {{ .Values.dynamicAllocation.maxExecutors }}
    shuffleTrackingEnabled: {{ .Values.dynamicAllocation.shuffleTrackingEnabled }}
    shuffleTrackingTimeout: {{ .Values.dynamicAllocation.shuffleTrackingTimeout }}
  {{- end }}
  sparkConf:
    {{- range $key, $value := .Values.sparkConf }}
    {{ $key | quote }}: {{ $value | quote }}
    {{- end }}
    "spark.hadoop.fs.s3a.endpoint": {{ .Values.application.env.OBJECT_STORE_ENDPOINT | quote }}
    "spark.hadoop.hive.metastore.uris": {{ .Values.application.env.HIVE_METASTORE_URI | quote }}
  monitoring:
    exposeDriverMetrics: true
    exposeExecutorMetrics: true
    prometheus:
      jmxExporterJar: {{ .Values.monitoring.jmxExporterJar | quote }}
      port: {{ .Values.monitoring.port }}
      portName: {{ .Values.monitoring.portName }}
  volumes:
    - name: kafka-tls
      secret:
        secretName: {{ .Values.kafkaTlsSecret }}
  driver:
    serviceAccount: {{ .Values.serviceAccount.name }}
    cores: {{ .Values.driver.cores }}
    coreRequest: {{ .Values.driver.coreRequest | quote }}
    coreLimit: {{ .Values.driver.coreLimit | quote }}
    memory: {{ .Values.driver.memory | quote }}
    memoryOverhead: {{ .Values.driver.memoryOverhead | quote }}
    labels:
      app.kubernetes.io/name: {{ include "stream-processor.name" . }}
      app.kubernetes.io/instance: {{ .Release.Name }}
      spark-platform/workload-class: streaming
    env:
      {{- range $name, $value := .Values.application.env }}
      - name: {{ $name }}
        value: {{ $value | quote }}
      {{- end }}
    envFrom:
      - secretRef:
          name: {{ .Values.existingSecret }}
    volumeMounts:
      - name: kafka-tls
        mountPath: /opt/spark/work-dir/ssl
        readOnly: true
    podSecurityContext:
      runAsUser: 185
      runAsGroup: 185
      runAsNonRoot: true
      fsGroup: 185
      seccompProfile:
        type: RuntimeDefault
    securityContext:
      allowPrivilegeEscalation: false
      capabilities:
        drop:
          - ALL
      readOnlyRootFilesystem: false
  executor:
    {{- if not .Values.dynamicAllocation.enabled }}
    instances: {{ .Values.executor.instances }}
    {{- end }}
    cores: {{ .Values.executor.cores }}
    coreRequest: {{ .Values.executor.coreRequest | quote }}
    coreLimit: {{ .Values.executor.coreLimit | quote }}
    memory: {{ .Values.executor.memory | quote }}
    memoryOverhead: {{ .Values.executor.memoryOverhead | quote }}
    labels:
      app.kubernetes.io/name: {{ include "stream-processor.name" . }}
      app.kubernetes.io/instance: {{ .Release.Name }}
      spark-platform/workload-class: streaming
    env:
      {{- range $name, $value := .Values.application.env }}
      - name: {{ $name }}
        value: {{ $value | quote }}
      {{- end }}
    envFrom:
      - secretRef:
          name: {{ .Values.existingSecret }}
    volumeMounts:
      - name: kafka-tls
        mountPath: /opt/spark/work-dir/ssl
        readOnly: true
    podSecurityContext:
      runAsUser: 185
      runAsGroup: 185
      runAsNonRoot: true
      fsGroup: 185
      seccompProfile:
        type: RuntimeDefault
    securityContext:
      allowPrivilegeEscalation: false
      capabilities:
        drop:
          - ALL
      readOnlyRootFilesystem: false

Operator 2.3.0 supports native Kubernetes-style env and envFrom fields. Its older envVars and envSecretKeyRefs fields are deprecated.

23.1 Do Not Hard-Code Batch Scheduler Memory

The example above contains:

memory: "8Gi"

only to show the field.

A production chart should calculate or explicitly configure the lower-bound memory requirement.

Helm templates are poorly suited to parsing Spark memory strings such as:

512m
2g
4Gi

A better model is:

scheduler:
  minimumResources:
    cpu: "3"
    memory: 10Gi

and then:

batchSchedulerOptions:
  resources:
    cpu: {{ .Values.scheduler.minimumResources.cpu | quote }}
    memory: {{ .Values.scheduler.minimumResources.memory | quote }}

Make the reservation explicit and reviewable.

23. Secret Management

The chart expects a secret to exist:

apiVersion: v1
kind: Secret
metadata:
  name: example-stream-secrets
  namespace: spark-streaming-standard
type: Opaque
stringData:
  KAFKA_SASL_USERNAME: "<managed externally>"
  KAFKA_SASL_PASSWORD: "<managed externally>"
  OBJECT_STORE_ACCESS_KEY: "<managed externally>"
  OBJECT_STORE_SECRET_KEY: "<managed externally>"

In production, generate this through:

• External Secrets Operator
• Secrets Store CSI Driver
• Vault Agent or Vault Secrets Operator
• Sealed Secrets
• Another audited secret-management system

For example:

apiVersion: external-secrets.io/v1
kind: ExternalSecret
metadata:
  name: example-stream-secrets
  namespace: spark-streaming-standard
spec:
  refreshPolicy: Periodic
  refreshInterval: 1h

  secretStoreRef:
    kind: ClusterSecretStore
    name: platform-vault

  target:
    name: example-stream-secrets
    creationPolicy: Owner

  data:
    - secretKey: KAFKA_SASL_USERNAME
      remoteRef:
        key: production/streams/example
        property: kafka_username

    - secretKey: KAFKA_SASL_PASSWORD
      remoteRef:
        key: production/streams/example
        property: kafka_password

Environment variables are visible in pod specifications when their values are directly embedded. Secret references avoid placing the literal value in the SparkApplication manifest, although processes inside the container can still read the resulting environment.

For certificates and keystores, file mounts are preferable.

Part VI:‌ Airflow

24. Airflow Should Orchestrate Jobs, Not Own Streaming Services

This distinction is important.

A bounded Spark job has a natural lifecycle:

submit → run → succeed/fail → Airflow task completes

A streaming service has a different lifecycle:

deploy → run indefinitely → restart or roll out independently

Keeping an Airflow task running for the lifetime of a streaming application:

• Consumes an Airflow task slot indefinitely
• Couples stream uptime to Airflow scheduler behavior
• Complicates retries and task clearing
• Makes planned stream restarts look like task failures
• Blurs application deployment and data orchestration

Use Helm or GitOps to own long-running stream processors.

Use Airflow for:

• Finite batch jobs
• Backfills
• Compaction
• Iceberg maintenance
• Snapshot expiration
• Historical replay
• Data-quality validation
• Starting or coordinating controlled maintenance

25. SparkKubernetesOperator for Bounded Jobs

Airflow's SparkKubernetesOperator creates a SparkApplication, monitors it, retrieves driver logs, and can delete the application after termination. Current provider versions support either a YAML/JSON application file or an in-memory template_spec.

A DAG can submit a templated manifest:

from __future__ import annotations
from datetime import datetime
from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.spark_kubernetes import (
    SparkKubernetesOperator,
)

with DAG(
    dag_id="daily_events_compaction",
    start_date=datetime(2026, 1, 1),
    schedule="0 3 * * *",
    catchup=False,
    max_active_runs=1,
    tags=["spark", "iceberg", "maintenance"],
) as dag:
    compact_events = SparkKubernetesOperator(
        task_id="compact_events",
        namespace="spark-batch",
        application_file=(
            "/opt/airflow/dags/spark/"
            "events-compaction.sparkapplication.yaml"
        ),
        kubernetes_conn_id="kubernetes_default",
        get_logs=True,
        log_events_on_failure=True,
        reattach_on_restart=True,
        delete_on_termination=True,
        startup_timeout_seconds=900,
    )

A manifest can use Airflow templating:

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: events-compaction-{{ ds_nodash }}
  namespace: spark-batch
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
  image: registry.example.com/data-platform/maintenance-jobs:2.4.1
  imagePullPolicy: IfNotPresent
  sparkVersion: {{ .Values.sparkVersion | quote }}
  mainApplicationFile: local:///opt/spark/app/compact.py
  arguments:
    - "--process-date"
    - "{{ ds }}"
  restartPolicy:
    type: Never
  batchScheduler: volcano
  batchSchedulerOptions:
    queue: spark-batch
    priorityClassName: spark-batch
    resources:
      cpu: "6"
      memory: 24Gi
  driver:
    serviceAccount: spark
    cores: 1
    memory: 2g
    memoryOverhead: 1g
  executor:
    instances: 5
    cores: 1
    memory: 3g
    memoryOverhead: 1g

reattach_on_restart=True is useful when the Airflow scheduler restarts while the Spark application is still running. Airflow labels the driver and executor pods so it can rediscover the existing workload instead of submitting a duplicate.

26.1 Airflow RBAC

Airflow needs permission to create and observe SparkApplication resources in the target namespace:

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: airflow-spark-submitter
  namespace: spark-batch
rules:
  - apiGroups:
      - sparkoperator.k8s.io
    resources:
      - sparkapplications
      - sparkapplications/status
    verbs:
      - create
      - get
      - list
      - watch
      - patch
      - update
      - delete
  - apiGroups:
      - ""
    resources:
      - pods
      - pods/log
      - events
    verbs:
      - get
      - list
      - watch

Do not give the Airflow service account unrestricted permissions across all Spark namespaces unless cross-namespace orchestration is explicitly required.

Part VII: Spark Connect

26. Separating Interactive Spark from Production Streaming

Spark Connect provides a remote client/server architecture for interactive Spark usage.

Operator 2.3.0 added:

apiVersion: sparkoperator.k8s.io/v1alpha1
kind: SparkConnect

The CRD supports a server pod, executor configuration, Spark configuration, Hadoop configuration, pod templates, and dynamic allocation.

A sanitized example:

apiVersion: sparkoperator.k8s.io/v1alpha1
kind: SparkConnect
metadata:
  name: shared-spark-connect
  namespace: spark-interactive
spec:
  sparkVersion: "3.5.6"
  image: registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4
  dynamicAllocation:
    enabled: true
    initialExecutors: 2
    minExecutors: 1
    maxExecutors: 10
    shuffleTrackingEnabled: true
    shuffleTrackingTimeout: 300000
  sparkConf:
    spark.connect.grpc.binding.address: "0.0.0.0"
    spark.connect.grpc.binding.port: "15002"
    spark.ui.bindAddress: "0.0.0.0"
    spark.ui.enabled: "true"
    spark.ui.port: "4040"
    spark.sql.execution.arrow.pyspark.enabled: "true"
    spark.sql.execution.pythonUDF.arrow.enabled: "true"
    spark.sql.extensions: org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
    spark.sql.catalog.production: org.apache.iceberg.spark.SparkCatalog
    spark.sql.catalog.production.type: hive
  server:
    cores: 2
    memory: 8Gi
    template:
      spec:
        serviceAccountName: spark
        securityContext:
          runAsUser: 185
          runAsGroup: 185
          runAsNonRoot: true
          seccompProfile:
            type: RuntimeDefault
        containers:
          - name: spark-kubernetes-driver
            image: registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4
            resources:
              requests:
                cpu: "2"
                memory: 8Gi
              limits:
                cpu: "2"
                memory: 8Gi
            securityContext:
              allowPrivilegeEscalation: false
              capabilities:
                drop:
                  - ALL
  executor:
    instances: 2
    cores: 2
    memory: 8Gi
    template:
      spec:
        securityContext:
          runAsUser: 185
          runAsGroup: 185
          runAsNonRoot: true
          seccompProfile:
            type: RuntimeDefault
        containers:
          - name: spark-kubernetes-executor
            image: registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4
            resources:
              requests:
                cpu: "2"
                memory: 8Gi
              limits:
                cpu: "2"
                memory: 8Gi
            securityContext:
              allowPrivilegeEscalation: false
              capabilities:
                drop:
                  - ALL

Do not include:

status:
  ...

in a declarative manifest.

The status field is maintained by the Kubernetes API and controller. Similarly, an Ingress manifest should not contain a copied status.loadBalancer section.

27.1 Exposing Spark Connect

Spark Connect uses gRPC:

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: shared-spark-connect
  namespace: spark-interactive
  annotations:
    nginx.ingress.kubernetes.io/backend-protocol: GRPC
    nginx.ingress.kubernetes.io/proxy-body-size: "0"
    nginx.ingress.kubernetes.io/proxy-read-timeout: "3600"
    nginx.ingress.kubernetes.io/proxy-send-timeout: "3600"
spec:
  ingressClassName: nginx
  rules:
    - host: spark-connect.example.com
      http:
        paths:
          - path: /
            pathType: Prefix
            backend:
              service:
                name: shared-spark-connect-server
                port:
                  number: 15002

Do not expose an unauthenticated Spark Connect server to an untrusted network.

Spark's Kubernetes documentation explicitly warns that Spark security features are not enabled automatically merely because an application runs on Kubernetes.

Place Spark Connect behind:

• An authenticated gateway
• TLS
• Network policy
• A restricted service account
• Namespace quota
• Request and session controls
• Audit logging

Interactive users should not share the same unrestricted identity as production pipelines.

Part VIII: Observability

27. Four Observability Layers

A usable Spark platform needs four different types of telemetry.

Operator telemetry

Examples:

Reconciliation errors
Submission failures
Applications by state
Job start latency
Webhook failures
Controller queue depth

These come from the operator metrics endpoint.

Kubernetes telemetry

Examples:

Pending driver and executor pods
CPU throttling
Container OOM kills
Quota rejection
Evictions
Node pressure
Scheduling latency
Pod restart count

These come from kube-state-metrics, cAdvisor, events, and scheduler metrics.

Spark runtime telemetry

Examples:

Executor count
Active tasks
Failed tasks
Shuffle read/write
JVM heap
GC time
Input rows
Processed rows
Batch duration
State-store size

The SparkApplication monitoring API supports exposing driver and executor metrics and configuring the Prometheus JMX exporter.

Application telemetry

Examples:

Kafka lag
Invalid records
Deduplicated records
Late records
Target commit duration
Last successful micro-batch
Checkpoint age
Business event counts

These must usually be emitted by application code.

28. PodMonitor per Workload Namespace

Because Prometheus Operator resources are namespaced, a PodMonitor can be created for each Spark namespace:

apiVersion: monitoring.coreos.com/v1
kind: PodMonitor
metadata:
  name: spark-metrics
  namespace: spark-streaming-standard
  labels:
    release: monitoring
spec:
  selector:
    matchExpressions:
      - key: spark-role
        operator: Exists
  podMetricsEndpoints:
    - port: prometheus
      path: /metrics
      interval: 30s

Repeat the resource for:

spark-streaming-critical
spark-batch
spark-ml
spark-interactive

Alternatively, deploy a centrally managed PodMonitor with a namespace selector if the Prometheus installation and RBAC model allow it.

29.1 Do Not Rely Only on the Live Spark UI

The driver UI disappears when the driver pod is removed.

Enable Spark event logs:

sparkConf:
  spark.eventLog.enabled: "true"
  spark.eventLog.dir: s3a://spark-event-logs/
  spark.eventLog.rolling.enabled: "true"
  spark.eventLog.rolling.maxFileSize: 128m

Then operate a Spark History Server against the same event-log storage.

Event logs are useful for:

• Post-failure investigation
• Stage and task analysis
• Shuffle diagnosis
• SQL-plan inspection
• Executor-loss analysis
• Historical performance comparison

They are not a replacement for Prometheus metrics or centralized logs.

29. Kubeflow's Grafana Dashboard

You could also use kubeflow's grafana dashboard to monitor the state of spark-operator

The community Spark Operator dashboard provides a useful starting point for controller and application-state monitoring. Because the JMX exporter exposes driver and executor metrics, you can also build workload-specific dashboards for throughput, batch duration, shuffle behavior, executor health, and JVM pressure.

I do also highly recommend reading this page for more information on benchmark of spark operator on k8s.

30. Create your own dashboards for spark jobs

Since we are grabbing all the metrics through JMX Exporter, you could create a dashboard like this to have more observability over the performance of your spark jobs.

here is an example:

31. Recommended Alerts

At minimum, alert on:

SparkApplication failed
Repeated submission failure
Driver CrashLoopBackOff
Executor OOMKilled
Driver OOMKilled
Application has zero executors
Executors pending beyond threshold
Structured Streaming batch duration above trigger interval
No completed micro-batch within threshold
Kafka consumer lag increasing continuously
Checkpoint commit failure
Object-store write failure
Operator leader unavailable
Webhook unavailable
Namespace quota above 90%
Volcano queue backlog above threshold

For a streaming application with a 30-second trigger, a useful indicator is:

processedRowsPerSecond < inputRowsPerSecond

for a sustained period.

A single slow micro-batch is not necessarily an incident. A growing backlog is.

Part IX: Security Boundaries

32. Pod Security

The driver and executor security context should include:

runAsNonRoot: true
runAsUser: 185
runAsGroup: 185
allowPrivilegeEscalation: false
capabilities:
  drop:
    - ALL
seccompProfile:
  type: RuntimeDefault

Avoid:

privileged containers
hostNetwork
hostPID
hostPath
host-mounted Docker sockets
arbitrary host ports

Spark supports mounting several volume types, including hostPath; Spark's own documentation warns that unrestricted host-path mounts create a security risk in shared environments.

Enforce these restrictions through Pod Security Admission or another admission-policy engine rather than relying only on application authors.

Part X: Failure Modes

33. Driver Running but No Executors

Check:

kubectl get pods -n <namespace>
kubectl describe pod <executor-pod> -n <namespace>
kubectl get resourcequota -n <namespace>
kubectl describe resourcequota -n <namespace>
kubectl get events -n <namespace> --sort-by=.lastTimestamp

Common causes:

• Namespace CPU quota exceeded
• Namespace memory quota exceeded
• Pod-count quota exceeded
• Volcano PodGroup not admitted
• Queue capacity unavailable
• Required node affinity cannot be satisfied
• Taints without matching tolerations
• Image pull failure
• Secret or ConfigMap missing
• Service account lacks pod-creation permissions
• Admission webhook failure

34. Python Worker Exits or OOM Kills

The configured executor heap is not the full container memory.

Inspect:

kubectl describe pod <executor> -n <namespace>
kubectl logs <executor> -n <namespace> --previous

Tune:

executor:
  memory: 4g
  memoryOverhead: 2g

or explicitly configure:

sparkConf:
  spark.executor.pyspark.memory: "1024"

Do not immediately increase the JVM heap. If Python or Arrow is consuming native memory, increasing the heap while leaving overhead unchanged can make the problem worse.

35. Application Restarts from the Wrong Kafka Position

Possible causes:

• A new checkpoint location was used
• The previous checkpoint was deleted
• Two applications wrote to the same checkpoint
• The checkpoint is unreadable
• Object-store credentials changed
• The application cannot access checkpoint metadata
• The query changed incompatibly

startingOffsets is mainly relevant when no checkpoint state exists. Once a valid checkpoint exists, the query resumes from its recorded progress.

36. Operator Upgrade Breaks New Applications

Check:

kubectl get crd sparkapplications.sparkoperator.k8s.io \
  -o jsonpath='{.spec.versions[*].name}'

kubectl get deployments -n spark-operator

kubectl logs deployment/<controller-deployment-name> -n spark-operator

kubectl get validatingwebhookconfigurations

kubectl get mutatingwebhookconfigurations

Possible causes:

• Controller upgraded but CRD did not
• CRD upgraded but webhook did not
• Webhook certificate expired
• Hook job lacked cluster-scoped RBAC
• New CRD validation rejects old manifests
• An application uses a removed or renamed field

Always render and validate manifests against the target CRD before the upgrade reaches production.

Part XI: CI/CD and Validation

37. Image Pipeline

The Spark runtime image pipeline should:

1. Build the image
2. Run unit tests
3. Verify Java, Python, and Spark versions
4. Validate required JARs
5. Run a local PySpark import test
6. Generate an SBOM
7. Scan OS and language packages
8. Sign the image
9. Push an immutable tag
10. Promote by digest

Example smoke test:

docker run --rm -i \
  registry.example.com/data-platform/spark-runtime:3.5.6-py3.11-r4 \
  python - <<'PY'
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("image-smoke-test")
    .getOrCreate()
)

assert spark.range(10).count() == 10
spark.stop()
PY

38. Helm Pipeline

Validate both the operator chart and application charts:

# 1. Validate static values and non-template manifests.
yamllint values.yaml values-production.yaml examples/

# 2. Validate Helm chart logic.
helm lint ./spark-platform
helm lint ./stream-processor

# 3. Rebuild locked dependencies.
helm dependency build ./spark-platform

# 4. Render everything.
helm template spark-platform ./spark-platform \
  --namespace spark-operator \
  --include-crds \
  -f values.yaml \
  -f values-production.yaml \
  > rendered-platform.yaml

helm template example-stream ./stream-processor \
  --namespace spark-streaming-standard \
  -f values.yaml \
  -f values/domain-a/values.yaml \
  > rendered-application.yaml

# 5. Ensure the rendered output is valid YAML.
yq eval-all '.' rendered-platform.yaml >/dev/null
yq eval-all '.' rendered-application.yaml >/dev/null

# 6. Validate ordinary Kubernetes resources.
kubeconform \
  -strict \
  -summary \
  -ignore-missing-schemas \
  rendered-platform.yaml \
  rendered-application.yaml

# 7. Validate against the actual cluster and installed CRDs.
kubectl apply \
  --dry-run=server \
  -f rendered-platform.yaml

kubectl apply \
  --dry-run=server \
  -f rendered-application.yaml

Then run:

kubeconform \
  -strict \
  -summary \
  -schema-location default \
  -schema-location \
  'https://raw.githubusercontent.com/yannh/kubernetes-json-schema/master/{{.NormalizedKubernetesVersion}}-standalone-strict/{{.ResourceKind}}.json' \
  rendered-platform.yaml

Custom resources require the Spark Operator CRD schema to be supplied to the validator.

Also run:

grep -RInE \
  '(password|secret|access[_-]?key|private[_-]?key).*[=:].+[^}]' \
  .

A real secret scanner such as Gitleaks or TruffleHog should be part of the repository pipeline.

Part XII: Final Operational Model

A maintainable production architecture ends with clear ownership:

The key architectural decisions are:

1. Treat the Spark Operator as a cluster control-plane component.
2. Treat streaming pipelines as services, not as permanently running Airflow tasks.
3. Package stream processors as SparkApplication resources through Helm.
4. Keep all credentials outside Helm values and Git.
5. Separate workload classes with namespaces, quotas, priorities, and scheduler queues.
6. Size quotas using the complete driver and executor pod footprint, including Python and off-heap memory.
7. Treat checkpoints as durable application state.
8. Use Spark dynamic allocation rather than Kubernetes HPA for executors.
9. Keep interactive Spark Connect workloads separate from production pipelines.
10. Pin the complete runtime compatibility matrix rather than upgrading individual JARs independently.
11. Observe the operator, Kubernetes, Spark, and business pipeline as four separate telemetry layers.
12. Validate CRDs and application manifests before every operator upgrade.

Spark on Kubernetes becomes reliable only when it is operated as a platform rather than a collection of isolated spark-submit commands.