Apache Spark Operators

Prerequisite

SparkJDBCOperator

Launches applications on a Apache Spark server, it uses SparkSubmitOperator to perform data transfers to/from JDBC-based databases.

For parameter definition take a look at SparkJDBCOperator.

Using the operator

Using cmd_type parameter, is possible to transfer data from Spark to a database (spark_to_jdbc) or from a database to Spark (jdbc_to_spark), which will write the table using the Spark command saveAsTable.

tests/system/apache/spark/example_spark_dag.py[source]

jdbc_to_spark_job = SparkJDBCOperator(
    cmd_type="jdbc_to_spark",
    jdbc_table="foo",
    spark_jars="${SPARK_HOME}/jars/postgresql-42.2.12.jar",
    jdbc_driver="org.postgresql.Driver",
    metastore_table="bar",
    save_mode="overwrite",
    save_format="JSON",
    task_id="jdbc_to_spark_job",
)

spark_to_jdbc_job = SparkJDBCOperator(
    cmd_type="spark_to_jdbc",
    jdbc_table="foo",
    spark_jars="${SPARK_HOME}/jars/postgresql-42.2.12.jar",
    jdbc_driver="org.postgresql.Driver",
    metastore_table="bar",
    save_mode="append",
    task_id="spark_to_jdbc_job",
)

Reference

For further information, look at Apache Spark DataFrameWriter documentation.

PySparkOperator

Launches applications on a Apache Spark Connect server or directly in a standalone mode

For parameter definition take a look at PySparkOperator.

Using the operator

tests/system/apache/spark/example_spark_dag.py[source]

def my_pyspark_job(spark):
    df = spark.range(100).filter("id % 2 = 0")
    print(df.count())

spark_pyspark_job = PySparkOperator(
    python_callable=my_pyspark_job, conn_id="spark_connect", task_id="spark_pyspark_job"
)

Reference

For further information, look at Running the Spark Connect Python.

SparkPipelinesOperator

Execute Spark Declarative Pipelines using the spark-pipelines CLI. This operator wraps the spark-pipelines binary to execute declarative data pipelines, supporting both pipeline execution and validation through dry-runs.

For parameter definition take a look at SparkPipelinesOperator.

Using the operator

The operator can be used to run declarative pipelines:

from airflow.providers.apache.spark.operators.spark_pipelines import SparkPipelinesOperator

# Execute the pipeline
run_pipeline = SparkPipelinesOperator(
    task_id="run_pipeline",
    pipeline_spec="/path/to/pipeline.yml",
    pipeline_command="run",
    conn_id="spark_default",
    num_executors=2,
    executor_cores=4,
    executor_memory="2G",
    driver_memory="1G",
)

Pipeline Specification

The pipeline_spec parameter should point to a YAML file defining your declarative pipeline:

name: my_pipeline
storage: file:///path/to/pipeline-storage
libraries:
  - glob:
      include: transformations/**

Pipeline Commands

  • run - Execute the pipeline (default)

  • dry-run - Validate the pipeline without execution

Reference

For further information, look at Spark Declarative Pipelines Programming Guide.

SparkSqlOperator

Launches applications on a Apache Spark server, it requires that the spark-sql script is in the PATH. The operator will run the SQL query on Spark Hive metastore service, the sql parameter can be templated and be a .sql or .hql file.

For parameter definition take a look at SparkSqlOperator.

Using the operator

tests/system/apache/spark/example_spark_dag.py[source]

spark_sql_job = SparkSqlOperator(
    sql="SELECT COUNT(1) as cnt FROM temp_table", master="local", task_id="spark_sql_job"
)

Reference

For further information, look at Running the Spark SQL CLI.

SparkSubmitOperator

Launches applications on a Apache Spark server, it uses the spark-submit script that takes care of setting up the classpath with Spark and its dependencies, and can support different cluster managers and deploy modes that Spark supports.

For parameter definition take a look at SparkSubmitOperator.

Using the operator

tests/system/apache/spark/example_spark_dag.py[source]

submit_job = SparkSubmitOperator(
    application="${SPARK_HOME}/examples/src/main/python/pi.py", task_id="submit_job"
)

Reference

For further information, look at Apache Spark submitting applications.

Durable execution (crash recovery)

In cluster deploy mode (--deploy-mode cluster) the Spark driver runs independently of the Airflow worker. If the worker dies while the Spark job is running, the driver keeps running but Airflow loses track of it, and submitting a brand new job on retry wastes the compute already done, or even causes conflicts if the Spark job itself is not designed to be idempotent.

The SparkSubmitOperator solves this by persisting the driver identifier to task state store immediately after submission. On retry, it reads the identifier back and reconnects to the already-running driver instead of resubmitting.

This is the synchronous path — the worker holds a slot for the duration of polling. This is a crash-safety net for teams running sync operators for log observability, org constraints, or because a Triggerer is not available. Teams with a Triggerer available may also consider deferrable operators, which free the worker slot but may come with added complexity.

durable defaults to True and every cluster manager supports it. What differs is the prerequisite each one needs before the operator can check driver status on retry:

Cluster manager

Prerequisite for durable=True

Spark standalone

REST scheme and REST port connection extras

Kubernetes cluster mode

track_driver_via_k8s_api=True

YARN cluster mode

yarn_track_via_rm_api=True and yarn_resourcemanager_webapp_address

Note

Durable execution requires Airflow 3.3+ (task_state_store support). Below 3.3, durable has no effect: setting it explicitly only emits a warning, and the operator always submits fresh, exactly as before this feature existed. The deprecated reconnect_on_retry parameter (the original name for this same feature, superseded almost immediately) still emits a deprecation warning on every Airflow version and maps onto durable.

Clearing a task is treated the same as a retry, which matters specifically for a task whose driver already succeeded: clearing does not delete the stored driver ID, so the next attempt reads it back and returns immediately without resubmitting. See Resumable Tasks for why, and for the [state_store] clear_on_success setting that restores “clearing always resubmits.”

This is most reliable for deferred tasks (deferrable=True); clearing a task that’s actively polling synchronously can cancel the driver via on_kill before the next attempt gets a chance to reconnect – see Resumable Tasks for why.

Spark standalone

Reconnecting to a running driver calls the Spark standalone REST API (GET /v1/submissions/status/{driverId}). Make sure the Spark connection’s REST scheme and REST port extras match your cluster’s configuration:

  • REST scheme — set to https if your cluster has TLS enabled on the REST port (spark.ssl.standalone.enabled=true). Defaults to http.

  • REST port — set to the value of spark.master.rest.port on your cluster. Defaults to 6066.

See Apache Spark Submit Connection for how to configure these fields.

Kubernetes cluster mode

When running in Kubernetes cluster mode, spark-submit blocks for the duration of the job. The JVM runs processes which does nothing but polling of the pod phase and holds heap space for the entire duration. This is not ideal for long-running jobs, especially when the driver is idle for long periods (e.g. waiting for data or user input).

Set track_driver_via_k8s_api=True to have the operator track the driver pod status via the Python Kubernetes client rather than holding spark-submit open for the full job duration. The same flag is what lets the operator find the driver again after a crash, so it is also the prerequisite for durable execution on Kubernetes: the driver pod name is persisted to task state before polling begins, and a retry reconnects to that pod instead of submitting a fresh one.

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

run_spark = SparkSubmitOperator(
    task_id="run_spark",
    application="local:///opt/spark/examples/jars/spark-examples.jar",
    conn_id="spark_k8s",
    deploy_mode="cluster",
    track_driver_via_k8s_api=True,
    durable=True,
)

Requirements

  • The Spark connection master must be k8s://... and deploy_mode must be cluster.

  • Do not set spark.kubernetes.submission.waitAppCompletion=true in your conf — this conflicts with the flag and a ValueError will be raised at task start.

  • The Airflow worker must be able to reach the Kubernetes API server and have permission to read and delete pods in the driver’s namespace; otherwise pod tracking and cleanup will fail.

  • Pod completion is detected from pod.status.phase. If your driver pods have sidecar containers (e.g. Istio injection enabled for the driver namespace), the pod phase may not advance to Succeeded until all sidecars exit. In that case the poll loop will wait indefinitely — set execution_timeout as a hard bound.

Set durable=False to always submit a fresh driver on retry.

YARN cluster mode

When running Spark applications on YARN in cluster deploy mode, the default Spark submit path keeps the local spark-submit JVM alive on the Airflow worker while the YARN application runs. For long-running Spark applications this can keep worker memory tied up for the whole application lifetime.

Set yarn_track_via_rm_api=True to release the local spark-submit JVM after YARN accepts the application, then poll the YARN ResourceManager REST API until the application reaches a terminal state. The ResourceManager API polling interval is controlled by status_poll_interval with a minimum of 10 seconds.

The ResourceManager REST API is also what makes checking application status on retry possible, so this flag is the prerequisite for durable execution on YARN. Because durable defaults to True, leaving the flag unset raises a ValueError at task start rather than silently falling back to a fire-and-forget submission.

This mode requires the Spark connection extra to set yarn_resourcemanager_webapp_address before the application is submitted:

airflow connections add spark_yarn_rm \
    --conn-type spark \
    --conn-host yarn \
    --conn-extra '{
        "deploy-mode": "cluster",
        "yarn_resourcemanager_webapp_address": "http://rm.example.com:8088"
    }'

SparkSubmitOperator(
    task_id="spark_pi",
    conn_id="spark_yarn_rm",
    application="/path/to/spark-examples.jar",
    java_class="org.apache.spark.examples.SparkPi",
    deploy_mode="cluster",
    yarn_track_via_rm_api=True,
)

For Kerberized clusters, install requests-kerberos in the Airflow environment. When the Spark connection has both keytab and principal configured, Airflow automatically uses HTTPKerberosAuth() for the ResourceManager REST requests.

Use yarn_rm_auth only when the ResourceManager needs a custom requests authentication object:

import requests

SparkSubmitOperator(
    task_id="spark_pi",
    conn_id="spark_yarn_rm",
    application="/path/to/spark-examples.jar",
    java_class="org.apache.spark.examples.SparkPi",
    deploy_mode="cluster",
    yarn_track_via_rm_api=True,
    yarn_rm_auth=requests.auth.HTTPBasicAuth("user", "password"),
)

Was this entry helpful?