Skip to content

Application examples

These examples describe the current source checkout. Run build and submission commands from the repository root, with namespace spark and service account spark configured as described in Getting Started. The example Dockerfiles default to the Apache Spark 3.5.9 image. To build them with Spark 4, set the image and matching Java dependency versions as described in Spark compatibility. For scheduled jobs, see the Airflow guide or the CronJob recipe. The ecosystem recipes also cover batch schedulers, OpenLineage and S3/MinIO storage.

Python

In this example, we use a small PySpark application that takes a parameter num_points to calculate the value of Pi: 

from __future__ import annotations

import random
import sys

from my_modules.pi import is_point_in_the_unit_circle
from pyspark.sql import SparkSession

if __name__ == "__main__":
    # Check if the number of arguments is correct
    if len(sys.argv) != 2:
        print("Usage: python script.py <num_points>")
        sys.exit(1)

    # Get the number of points from command-line arguments
    num_points = int(sys.argv[1])

    # Create a Spark session
    spark = SparkSession.builder.appName("Pi-Estimation").getOrCreate()

    # Generate random points within the unit square
    points = spark.sparkContext.parallelize(range(1, num_points + 1)).map(
        lambda _: (random.random(), random.random())
    )

    # Count points within the unit circle
    inside_circle = points.filter(is_point_in_the_unit_circle)

    # Estimate Pi
    pi_estimate = 4 * inside_circle.count() / num_points

    # Display the result
    print(f"Pi is approximately {pi_estimate}")

    # Stop the Spark session
    spark.stop()
and a Dockerfile to package the application with the spark image: 
ARG SPARK_IMAGE=apache/spark:3.5.9-scala2.12-java17-python3-ubuntu
FROM ${SPARK_IMAGE}

COPY my_modules /opt/spark/work-dir/my_modules
COPY job.py /opt/spark/work-dir/
ENV PYTHONPATH="${PYTHONPATH}:/opt/spark/work-dir"

Build the image from the repository root:

docker build -t pyspark-job examples/python

For a local cluster, load the image into one of the following cluster runtimes:

# Minikube
minikube image load pyspark-job
# Or kind
kind load docker-image pyspark-job

The supplied submission scripts use image_pull_policy="Never" (CLI: --image-pull-policy Never) for these preloaded images. For a remote cluster, replace the example registry below, push the tagged image, and update the script's image to that same name and its pull policy to IfNotPresent:

docker tag pyspark-job registry.example.com/team/pyspark-job:example
docker push registry.example.com/team/pyspark-job:example
Then, you can submit the job using the python client: 
from __future__ import annotations

import logging

from spark_on_k8s.client import ExecutorInstances, PodResources, SparkOnK8S

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    spark_client = SparkOnK8S()
    spark_client.submit_app(
        image="pyspark-job",
        app_path="local:///opt/spark/work-dir/job.py",
        app_arguments=["100000"],
        app_name="pyspark-job-example",
        namespace="spark",
        service_account="spark",
        app_waiter="log",
        # If you test this locally (minikube or kind) without pushing the image to a registry,
        # you need to set the image_pull_policy to Never.
        image_pull_policy="Never",
        ui_reverse_proxy=True,
        driver_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
        executor_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
        # Run with 5 executors
        executor_instances=ExecutorInstances(initial=5),
    )
or using the CLI: 
spark-on-k8s app submit \
  --image pyspark-job \
  --path local:///opt/spark/work-dir/job.py \
  --namespace spark \
  --name pyspark-job-example \
  --service-account spark \
  --image-pull-policy Never \
  --driver-cpu 1 \
  --driver-memory 512 \
  --driver-memory-overhead 128 \
  --executor-cpu 1 \
  --executor-memory 512 \
  --executor-memory-overhead 128 \
  --executor-initial-instances 5 \
  --ui-reverse-proxy \
  --logs \
  100000

Java

This example is similar to the previous one, but it's implemented in java (with maven).

pom.xml: 

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.oss-tech.examples</groupId>
    <artifactId>java-job</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <spark.version>3.5.9</spark.version>
        <scala.binary.version>2.12</scala.binary.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>
</project>
the job class (src/main/java/com/oss_tech/examples/TestJob.java): 
package com.oss_tech.examples;

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;


public class TestJob {
  public static void main(String[] args) {
    // Check if the correct number of arguments is provided
    if (args.length != 1) {
      System.err.println("Usage: PiEstimation <num_points>");
      System.exit(1);
    }

    // Initialize Spark session
    final SparkSession spark = SparkSession.builder()
        .appName("Spark Java example")
        .getOrCreate();

    // Get the number of points from command-line arguments
    final int numPoints = Integer.parseInt(args[0]);

    Dataset<Row> points = spark.range(1, numPoints + 1)
        .selectExpr("rand() as x", "rand() as y");

    // Count points within the unit circle
    long insideCircle = points.filter("x * x + y * y <= 1").count();

    // Estimate Pi
    double piEstimate = 4.0 * insideCircle / numPoints;

    // Display the result
    System.out.println("Pi is approximately " + piEstimate);
  }
}

Build the image from the repository root:

docker build -t java-spark-job examples/java

For a local cluster, load the image into one of the following cluster runtimes:

# Minikube
minikube image load java-spark-job
# Or kind
kind load docker-image java-spark-job

The supplied submission scripts use image_pull_policy="Never" (CLI: --image-pull-policy Never) for these preloaded images. For a remote cluster, replace the example registry below, push the tagged image, and update the script's image to that same name and its pull policy to IfNotPresent:

docker tag java-spark-job registry.example.com/team/java-spark-job:example
docker push registry.example.com/team/java-spark-job:example
Then, submit the job using the python client: 
from __future__ import annotations

import logging

from spark_on_k8s.client import ExecutorInstances, SparkOnK8S

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    spark_client = SparkOnK8S()
    spark_client.submit_app(
        image="java-spark-job",
        app_path="local:///java-job.jar",
        app_arguments=["100000"],
        app_name="spark-java-job-example",
        namespace="spark",
        service_account="spark",
        app_waiter="log",
        class_name="com.oss_tech.examples.TestJob",
        # If you test this locally (minikube or kind) without pushing the image to a registry,
        # you need to set the image_pull_policy to Never.
        image_pull_policy="Never",
        ui_reverse_proxy=True,
        # Run with dynamic allocation enabled, with minimum of 0 executors and maximum of 5 executors
        executor_instances=ExecutorInstances(min=0, max=5),
    )
or using the CLI: 
spark-on-k8s app submit \
  --image java-spark-job \
  --path local:///java-job.jar \
  --namespace spark \
  --name spark-java-job-example \
  --service-account spark \
  --image-pull-policy Never \
  --executor-min-instances 0 \
  --executor-max-instances 5 \
  --executor-initial-instances 5 \
  --ui-reverse-proxy \
  --logs \
  --class com.oss_tech.examples.TestJob \
  100000

Driver configuration

This package manages Spark applications in Kubernetes by directly creating the driver pod and submitting the job in client mode. This approach bypasses some of Spark's built-in driver configuration mechanisms, causing certain settings to be ignored (e.g., spark.kubernetes.driver.podTemplateFile, spark.kubernetes.driver.label.[LabelName], ...).

To address this, the package provides internal implementations for some key driver configurations, ensuring greater control and flexibility when deploying Spark applications in Kubernetes.

Driver service account

Instead of providing spark.kubernetes.authenticate.driver.serviceAccountName=<some-service-account>, you can use: 

spark_client.submit_app(
    ...,
    service_account="<some-service-account>",
)

Driver resources

Instead of providing spark.driver.memory, spark.driver.cores, spark.driver.memoryOverhead, you can use: 

from spark_on_k8s.client import PodResources

spark_client.submit_app(
    ...,
    driver_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
)

Driver environment variables

The source version supports spark.kubernetes.driverEnv.NAME in spark_conf. The client copies these properties into the driver container's environment before creating the Pod. This also applies to the CLI and both Airflow operators, which use the same submission code. This requires 0.13.0 or newer.

For example, a launcher that loads a YAML property mapping into spark_conf can use:

spark.kubernetes.driverEnv.DATABASE_NAME: dbname
spark.executorEnv.DATABASE_NAME: executor_dbname

The driver receives DATABASE_NAME=dbname; Spark independently sets DATABASE_NAME=executor_dbname on executors. A driver-only property does not add an executor environment variable. With job files, put these properties under spec.spark_conf; with the CLI, pass them through --conf.

Names must be nonempty printable ASCII without =. Values may be strings, booleans, integers or finite floats; booleans become true/false, and an empty string remains empty. Null, structured values and NUL characters are rejected before Kubernetes writes. SPARK_DRIVER_BIND_ADDRESS is managed from the Pod IP and cannot be supplied through spark.kubernetes.driverEnv.

An explicit driver property replaces a matching driver template environment entry, including its valueFrom reference. Explicit environment entries also take precedence over matching envFrom Secret/ConfigMap keys, including secret_values and driver_env_vars_from_secrets. Sidecar and init-container environments are unchanged. Kubernetes retains its normal $(NAME) expansion behavior for environment values.

Use these properties for non-secret settings: they remain ordinary Spark configuration in the driver arguments and may appear in Spark diagnostics. Offline plan output redacts their values in both environment entries and configuration arguments, as it does for spark.executorEnv.* values; raw ApplicationPlan.resources retains the actual values for inspection. For credentials, use Kubernetes Secret references in the driver template or the Secret options below.

Use secret_values to create an application-owned Kubernetes Secret. Its keys become environment variables in both the driver and executors; executor delivery uses Spark's spark.kubernetes.executor.secretKeyRef configuration. To import existing Secrets into the driver only, use driver_env_vars_from_secrets. 

spark_client.submit_app(
    ...,
    secret_values={"EnvironmentVariableName": "some-value"},
)

Driver volume mounts

Pass Kubernetes volume and mount objects to volumes and driver_volume_mounts. The Secret secret1, ConfigMap configmap1 and persistent volume claim pvc in this example must already exist in the application's namespace. Replace the host and NFS paths with resources available in your cluster:

from kubernetes import client as k8s

spark_client.submit_app(
    ...,
    volumes=[
        k8s.V1Volume(
            name="volume1",
            host_path=k8s.V1HostPathVolumeSource(path="/mnt/volume1"),
        ),
        k8s.V1Volume(
            name="volume2",
            empty_dir=k8s.V1EmptyDirVolumeSource(medium="Memory", size_limit="1Gi"),
        ),
        k8s.V1Volume(
            name="volume3",
            secret=k8s.V1SecretVolumeSource(
                secret_name="secret1", items=[k8s.V1KeyToPath(key="key1", path="path1")]
            ),
        ),
        k8s.V1Volume(
            name="volume4",
            config_map=k8s.V1ConfigMapVolumeSource(
                name="configmap1", items=[k8s.V1KeyToPath(key="key1", path="path1")]
            ),
        ),
        k8s.V1Volume(
            name="volume5",
            projected=k8s.V1ProjectedVolumeSource(
                sources=[
                    k8s.V1VolumeProjection(
                        secret=k8s.V1SecretProjection(
                            name="secret1",
                            items=[k8s.V1KeyToPath(key="key1", path="secret/key1")],
                        )
                    ),
                    k8s.V1VolumeProjection(
                        config_map=k8s.V1ConfigMapProjection(
                            name="configmap1",
                            items=[k8s.V1KeyToPath(key="key1", path="config/key1")],
                        )
                    ),
                ]
            ),
        ),
        k8s.V1Volume(
            name="volume6",
            nfs=k8s.V1NFSVolumeSource(server="nfs-server", path="/mnt/volume6", read_only=True),
        ),
        k8s.V1Volume(
            name="volume7",
            persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(
                claim_name="pvc", read_only=True
            ),
        ),
    ],
    driver_volume_mounts=[
        k8s.V1VolumeMount(name="volume1", mount_path="/mnt/volume1"),
        k8s.V1VolumeMount(name="volume2", mount_path="/mnt/volume2", sub_path="sub-path"),
        k8s.V1VolumeMount(name="volume3", mount_path="/mnt/volume3"),
        k8s.V1VolumeMount(name="volume4", mount_path="/mnt/volume4"),
        k8s.V1VolumeMount(name="volume5", mount_path="/mnt/volume5"),
        k8s.V1VolumeMount(name="volume6", mount_path="/mnt/volume6"),
        k8s.V1VolumeMount(name="volume7", mount_path="/mnt/volume7", read_only=True),
    ],
)

Driver node selector

Instead of providing spark.kubernetes.driver.node.selector.[labelKey]=<labelValue>, you can use: 

spark_client.submit_app(
    ...,
    driver_node_selector={"labelKey": "labelValue"},
)

Driver tolerations

Instead of configuring tolerations for the driver pod via a pod template file, you can use: 

from kubernetes import client as k8s

spark_client.submit_app(
    ...,
    driver_tolerations=[
        k8s.V1Toleration(
            key="key1",
            operator="Equal",
            value="value1",
            effect="NoSchedule",
        ),
        k8s.V1Toleration(
            key="key2",
            operator="Exists",
            effect="NoExecute",
        ),
    ],
)

Driver init containers

Instead of adding init containers to the driver pod via a pod template file, you can use: 

from kubernetes import client as k8s

spark_client.submit_app(
    ...,
    driver_init_containers=[
        k8s.V1Container(
            name="init-container1",
            image="init-container1-image",
            command=["init-command1"],
            args=["init-arg1"],
        ),
        k8s.V1Container(
            name="init-container2",
            image="init-container2-image",
            command=["init-command2"],
            args=["init-arg2"],
        ),
    ],
)

Driver host aliases

Instead of providing host aliases (entries added to pod's /etc/hosts file) for the driver pod via a pod template file, you can use: 

from kubernetes import client as k8s

spark_client.submit_app(
    ...,
    driver_host_aliases=[
        k8s.V1HostAlias(
            hostnames=["foo.local", "bar.local"],
            ip="127.0.0.1",
        )
    ],
)

Executor configuration

Executor pod template

The package provides an enhanced way to configure executor pod templates compared to Spark's built-in spark.kubernetes.executor.podTemplateFile configuration.

Instead of manually creating ConfigMaps and managing volume mounts, you can use the executor_pod_template parameter that accepts either:

  • Template content as a string (YAML)
  • Path to a local template file

The package automatically:

  1. Creates a ConfigMap with the template content
  2. Mounts it to the driver pod (so the driver can validate the template)
  3. Configures Spark to use the template for executor pods
  4. Sets the driver pod as the ConfigMap owner, so Kubernetes cleans up the ConfigMap when that pod is deleted

These submissions use the locally preloaded pyspark-job image and the spark namespace/service account prepared earlier. For a registry image, change the image name and use IfNotPresent as described in the Python example above.

Using template content as string: 

executor_template_yaml = """
apiVersion: v1
kind: Pod
spec:
  containers:
  - name: executor
    resources:
      requests:
        memory: "4Gi"
        cpu: "2"
      limits:
        memory: "4Gi"
        cpu: "2"
    env:
    - name: CUSTOM_ENV_VAR
      value: "custom-value"
  nodeSelector:
    node-type: "compute-optimized"
  tolerations:
  - key: "spark-executor"
    operator: "Equal"
    value: "true"
    effect: "NoSchedule"
"""

spark_client.submit_app(
    image="pyspark-job",
    app_path="local:///opt/spark/work-dir/job.py",
    namespace="spark",
    service_account="spark",
    image_pull_policy="Never",
    executor_pod_template=executor_template_yaml,
)

Using local template file: 

# Save your template to a file, e.g., executor-template.yaml
spark_client.submit_app(
    image="pyspark-job",
    app_path="local:///opt/spark/work-dir/job.py",
    namespace="spark",
    service_account="spark",
    image_pull_policy="Never",
    executor_pod_template="./executor-template.yaml",
)

Via CLI: 

# Using template file
spark-on-k8s app submit --image pyspark-job --path local:///opt/spark/work-dir/job.py \
    --namespace spark --service-account spark --image-pull-policy Never \
    --executor-template ./executor-template.yaml

# You can also set it via environment variable
export SPARK_ON_K8S_EXECUTOR_TEMPLATE="./executor-template.yaml"
spark-on-k8s app submit --image pyspark-job --path local:///opt/spark/work-dir/job.py \
    --namespace spark --service-account spark --image-pull-policy Never

Note: executor_pod_template and executor_pod_template_path are mutually exclusive. Use executor_pod_template_path (or --executor-pod-template-path in the CLI) for a template path already available to the Spark driver. Do not configure both options, including through environment defaults.