Running External Compute: Spark, Docker & Kubernetes Pods

Airflow orchestrates: it decides what runs, and when. It should not do the heavy computing itself.

This chapter shows what that rule means for real tasks. Heavy work runs on another system. The Airflow task starts it, waits for it, and records how it ended.

You keep the data from the last chapters: the DAG shop_orders_daily. You also meet a made-up heavy job, orders_history_rollup. It reads one year of orders.

Every number about that job is made up. You answer four questions:

1. Why can the Airflow worker not run a heavy job?
2. Which operators start work somewhere else?
3. How does a task wait without wasting a worker slot?
4. What happens to the job when the task fails?

A worker is the machine that runs your tasks. PythonOperator and BashOperator run on the worker, inside its memory.

Take a made-up worker with 8 GiB of memory. A GiB is about one billion bytes.

A slot is a place for one running task. The worker has 16 slots, the default worker_concurrency from Executors Compared: Local, Celery, Kubernetes.

Now the made-up job orders_history_rollup. It reads 365 daily order files of 0.2 GiB each. Look at the numbers:

The task cannot fit. The system kills it when the memory is full, and the task fails.

You could buy bigger workers. But then every worker must be big, all day, for one job that runs a few hours.

Spark solves this by splitting the data. A Spark cluster has a driver that plans the job and executors that do the work. A cluster manager such as YARN or Kubernetes gives it machines.

Say a made-up cluster has 20 executors with 8 GiB each. Each one holds 73 divided by 20, so 3.65 GiB of the year. That fits.

The Airflow task only starts the job and waits. Deploying Spark: spark-submit, YARN & Kubernetes teaches how the cluster runs and how to start it.

Airflow has an operator for each place where work can run. Some run on the worker. Others start the work somewhere else.

Look at where the work happens for each one. A container is a sealed package of a program and its libraries. A pod is a small box of containers that Kubernetes runs:

KubernetesPodOperator starts a pod for the work of one task, under any executor.

KubernetesExecutor is different. It starts a pod for every task. Executors Compared: Local, Celery, Kubernetes compares them.

SparkSubmitOperator runs the spark-submit program on the worker. The program is small, but it must be installed there. The task stays on the worker until the Spark job ends, as the next section shows.

Cloud operators call the API of a service, its interface for programs. Three examples, from the providers. Check your provider version for the exact options:

• EmrAddStepsOperator adds a step to an Amazon EMR cluster. EmrStepSensor waits for it.
• EmrServerlessStartJobOperator starts a job on EMR Serverless.
• DatabricksSubmitRunOperator sends a one-time run to Databricks. DatabricksRunNowOperator starts an existing job by its job_id.

After the job starts, the task must wait for the result. Downstream tasks need the output. So the task should end only when the job ends.

If the task waits by polling, it holds its slot for the whole job. Polling means asking again and again: are you done yet?

A deferrable task hands its wait to the triggerer, the Airflow part that waits for events. Concurrency, Pools, Priorities & Deferrable Operators teaches it.

Now count. The made-up job orders_history_rollup takes 2 hours, and ten such jobs run each night. Compare two ways to wait:

In the blocking row, 10 slots times 2 hours is 20 hours. Those 10 slots do nothing but wait, and 22 slots are left for real work.

In the deferrable row, 10 jobs times 10 seconds is 100 seconds.

Many provider operators accept deferrable=True. Not every operator does, so check the provider documentation for your version. Here is one for Databricks:

run = DatabricksRunNowOperator(
    task_id="orders_history_rollup",
    job_id=7,
    notebook_params={"run_date": "{{ ds }}"},
    deferrable=True,
)

Do not free the slot by not waiting. Some operators end as soon as the job is accepted. EmrAddStepsOperator does this by default: its wait_for_completion is False.

Then downstream tasks start with no output. If the job fails later, the Airflow run still shows green. So add EmrStepSensor, or set wait_for_completion=True or deferrable=True.

Sometimes the heavy work is your own Python code, not a Spark job. A pod gives that code its own image, its own memory and its own libraries.

An image is a saved package with your code and libraries. Now a made-up task, orders_month_rollup. It reads one month: 31 files of 0.2 GiB, so 6.2 GiB.

That does not fit a worker slot of 0.5 GiB. It fits a pod with 16 GiB. Look at the operator:

from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from kubernetes.client import models as k8s

rollup = KubernetesPodOperator(
    task_id="orders_month_rollup",
    name="orders-month-rollup",
    namespace="batch",
    image="my-registry/orders-rollup:1.0",
    cmds=["python", "rollup.py"],
    arguments=["--date", "{{ ds }}"],
    container_resources=k8s.V1ResourceRequirements(
        requests={"memory": "16Gi"},
        limits={"memory": "16Gi"},
    ),
    get_logs=True,
    on_finish_action="delete_pod",
)

The value delete_pod removes the pod at the end. Other values include delete_succeeded_pod and keep_pod.

With delete_succeeded_pod, a failed pod stays so you can inspect it. Provider versions before 7.2.0 used is_delete_operator_pod instead.

Since version 5.2.0 of the Kubernetes provider, it also accepts deferrable=True. Then the pod runs while the worker slot is free.

The remote job writes its output to storage, such as a table or a folder. It does not send the output back through Airflow.

An XCom is a small value that one task passes to the next. Airflow stores it in the metadata database. So pass a reference, not the data.

Look at what the task could pass for orders_history_rollup. The folder name is made up:

The limit depends on your metadata database, so keep XComs small. When something fails, you need the logs. They live in different places:

The Airflow log often shows only that the job failed. The real error is in the other system. So log the job URL or id right after you submit.

Airflow keeps a job id. It does not keep the job. If the task is killed, the remote job often keeps running.

Follow the task orders_history_rollup in a made-up run, dag_run 430. A heartbeat is a regular "I am alive" signal from a running task. The worker dies 30 minutes in, and Airflow retries the task:

What the two writes do depends on how the job writes. Take 2026-01-15 from the shop data: its revenue is 15.50. Look at the code, then at the table after both jobs:

# Append: adds the rows again
daily.write.mode("append").insertInto("orders_rollup")

# Overwrite one day: replaces that day only
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
daily.write.mode("overwrite").insertInto("orders_rollup")

This is idempotency from Idempotency & Safe Task Design: running twice gives the same result as running once. It protects the data even when cancellation fails. Two more defences help:

• on_kill: Airflow calls this method of the operator when it stops a task. A good operator cancels the remote job there. Not all operators do.
• Reattach: some operators save the job id and reconnect to it on retry. KubernetesPodOperator does this by default with reattach_on_restart. Newer Airflow releases may extend this, so check your version.

A hard kill of the worker may never call on_kill. So do not rely on cancellation alone.

Airflow submits the heavy work, waits cheaply, and records the result. Look at the answers you should know:

Try this alone. A task loads 40 GiB with pandas on a worker with 8 GiB. Where should it run, and how should the task wait?

The answer is a Spark cluster or a large pod. The task should wait in deferrable mode, if the operator offers it.