Orchestrating pipelines with Airflow in Spark & Data Eng

Introduction

Orchestrating pipelines with Airflow in Spark & Data Eng separates the script-and-forget approach from production-grade workflows that teams can monitor, debug, and scale with confidence. When you're running daily data ingests, transformations, and aggregations across multiple systems, manual scheduling falls apart quickly. Apache Airflow gives us a declarative way to define our tasks, express their dependencies, and handle failure gracefully—all while keeping visibility into what's running and what went wrong.

Over the past 45 days, we've built transformations in Spark, optimized joins, and tuned partitions. Now we need to glue everything together into a reliable system that runs at 2 AM without waking anyone up. This article walks through how orchestration works in practice, from DAG design to handling backfills when yesterday's data finally arrived.

DAGs — Python Files Define Tasks and Dependencies

A Directed Acyclic Graph (DAG) is the backbone of Airflow. It's a Python file where you declare tasks and tell Airflow which task depends on which. Airflow reads these files, understands the dependency graph, and schedules everything in the right order. Because it's Python, you have the full language at your disposal—no XML, no YAML spaghetti.

Let's say we run a daily pipeline that ingests user events from S3, transforms them in Spark, and loads aggregated metrics into a data warehouse. Here's a minimal DAG:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.spark_submit_operator import SparkSubmitOperator
from airflow.operators.python import PythonOperator

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'start_date': datetime(2024, 1, 1),
}

dag = DAG(
    'daily_user_metrics',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # 2 AM daily
    catchup=False,
)

def check_data_ready(**context):
    # Poll S3 or data lake to confirm events landed
    execution_date = context['execution_date']
    # ... implementation
    return True

check_task = PythonOperator(
    task_id='check_data_ready',
    python_callable=check_data_ready,
    dag=dag,
)

ingest_task = SparkSubmitOperator(
    task_id='ingest_events',
    application='/home/spark/jobs/ingest_events.py',
    conf={'spark.executor.memory': '4g'},
    dag=dag,
)

transform_task = SparkSubmitOperator(
    task_id='transform_metrics',
    application='/home/spark/jobs/transform_metrics.py',
    dag=dag,
)

load_task = SparkSubmitOperator(
    task_id='load_warehouse',
    application='/home/spark/jobs/load_warehouse.py',
    dag=dag,
)

check_task >> ingest_task >> transform_task >> load_task

The last line defines the dependency chain: check runs first, then ingest, then transform, then load. Airflow visualizes this as a graph in its web UI. Each task gets its own log file, so debugging is straightforward. The schedule_interval of '0 2 * * *' means "every day at 2 AM UTC." When it runs, Airflow passes an execution_date to each task, letting your scripts know which time period they're processing.

A key design principle: keep your DAG file lightweight and focused on orchestration. The actual work—the SQL, the Spark logic, the data validation—lives in separate scripts or operators. This lets you test and iterate on your transformations independently of your orchestration layer.

Operators — Spark, SQL, Python and Cloud Operators

Operators are the building blocks of a DAG. Each operator represents a unit of work. Airflow ships with dozens; we typically use a handful in data pipelines.

SparkSubmitOperator is your main tool for running Spark jobs. It submits a Spark application to a cluster (local, Standalone, YARN, or Kubernetes) and waits for it to complete. You configure it with the application path, executor memory, number of partitions, and any arguments:

spark_job = SparkSubmitOperator(
    task_id='compute_daily_agg',
    application='s3://my-bucket/jobs/daily_agg.py',
    name='daily_agg_{{ds}}',  # Templated task name
    conf={
        'spark.executor.memory': '8g',
        'spark.executor.cores': '4',
        'spark.sql.shuffle.partitions': '200',
    },
    application_args=['--date', '{{ ds }}'],  # Pass execution date
    dag=dag,
)

PythonOperator runs a Python function in the Airflow worker process. Use this for lightweight work: checking file existence, triggering alerts, updating metadata, or calling APIs. Avoid long-running or CPU-heavy tasks here; they'll block the worker.

SQLExecuteQueryOperator (or its predecessors) runs SQL directly against a database. Useful for populating lookup tables or materialized views:

from airflow.providers.postgres.operators.postgres import PostgresOperator

refresh_lookup = PostgresOperator(
    task_id='refresh_user_lookup',
    sql='REFRESH MATERIALIZED VIEW CONCURRENTLY user_lookup;',
    postgres_conn_id='my_warehouse',
    dag=dag,
)

Cloud operators integrate with AWS, GCP, or Azure. Common ones: S3ToRedshiftOperator, BigQueryCheckOperator, AzureDataFactoryRunPipelineOperator. They let you orchestrate cloud-native services without dropping into Python:

from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftOperator

load_s3_to_redshift = S3ToRedshiftOperator(
    task_id='load_events_to_redshift',
    s3_bucket='my-bucket',
    s3_key='events/2024-01-15/',
    schema='raw',
    table='events',
    copy_options=['IGNOREHEADER 1', 'DELIMITER \',\''],
    redshift_conn_id='redshift_default',
    dag=dag,
)

Airflow also supports branching and conditional execution. If a data quality check fails, you might skip downstream tasks or trigger an alert instead of proceeding blindly. This keeps pipelines safe and debuggable.

Retries and Alerts — Configure Per Task

Network hiccups, cluster contention, and transient AWS throttling happen. Rather than failing the entire DAG, configure retries at the task level. Most of our Spark jobs have 2–3 retries with exponential backoff:

from datetime import timedelta
from airflow.models import Variable

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=60),
}

# Or override per task
risky_job = SparkSubmitOperator(
    task_id='flaky_external_data',
    application='s3://bucket/flaky_job.py',
    retries=4,  # Extra retries for this one
    retry_delay=timedelta(minutes=10),
    dag=dag,
)

For alerting, Airflow integrates with email, Slack, PagerDuty, and custom webhooks. Define callbacks on the DAG or individual tasks:

def on_failure_callback(context):
    """Send Slack alert on task failure."""
    task_instance = context['task_instance']
    message = f"""
    :x: Task {task_instance.task_id} failed.
    DAG: {context['dag'].dag_id}
    Execution date: {context['execution_date']}
    Log: {task_instance.log_url}
    """
    # Call Slack API here, or use airflow.providers.slack

dag = DAG(
    'daily_user_metrics',
    default_args=default_args,
    on_failure_callback=on_failure_callback,
    schedule_interval='0 2 * * *',
    dag=dag,
)

You can also set per-task email alerts. In production, we alert on failure immediately and only email the team lead after three consecutive failures—too many alerts become noise.

Backfills — Rerun Historical Date Ranges

Reality: data arrives late. A vendor's API went down yesterday, or you discovered a bug in your transformation a week ago. You need to rerun your pipeline for dates 2024-01-08 through 2024-01-14. That's backfilling.

Airflow makes this simple. From the CLI:

airflow dags backfill \
  --dag-id daily_user_metrics \
  --start-date 2024-01-08 \
  --end-date 2024-01-14 \
  --reset-dag-run

This creates DAG runs for each date and executes them in parallel (respecting resource limits). Your Spark jobs receive the correct execution date via templating, so they naturally process the right date range. No manual juggling of date parameters.

Backfills are also useful when you first deploy a pipeline and want historical data loaded. A data lake ingestion DAG might backfill three years of archived logs on day one.

One caveat: if your job already processed 2024-01-10 and wrote outputs to a table, backfilling might double-load data. Design your jobs to be idempotent—use TRUNCATE + LOAD or upserts instead of appends. Or partition outputs by execution date and delete the old partition before reloading.

Common Mistakes

• Over-parameterizing DAGs. Airflow DAG files should be simple and consistent. If you find yourself writing loops or complex logic in the DAG file itself, you're doing orchestration work in the wrong place. Move that logic into your Spark job or Python operator.
• Ignoring task idempotency. A task might run twice if the scheduler hiccups. If your Spark job appends to a table without checking for duplicates, backfilling will cause data duplication. Always design for replay.
• Missing execution_date templating. Forget to pass {{ ds }} to your Spark job, and it'll process today's date every time, even during backfills. Use Airflow's templating liberally.
• Setting catch_up=True recklessly. A DAG with catch_up=True and a large gap between start_date and current date will try to run thousands of DAG instances at once. Cluster goes down, humans get angry. Use catchup=False by default and backfill manually when needed.
• Hardcoding connection credentials. Always use Airflow Connections or Secrets Manager. Hardcoded passwords in DAG files are a security nightmare and show up in version control.

Key Takeaways

• DAGs are Python files that define tasks and their dependencies. Airflow reads them at startup and executes tasks in the correct order.
• Use SparkSubmitOperator for Spark jobs, PythonOperator for lightweight custom logic, and cloud operators to integrate managed services.
• Configure retries and exponential backoff per task to handle transient failures gracefully.
• Set up alerting callbacks so your team knows immediately when pipelines fail.
• Backfilling is built in: rerun any date range with a single CLI command. Design jobs to be idempotent so replays don't corrupt data.
• Keep DAG files clean and focused on orchestration. Business logic belongs in your Spark jobs and Python scripts.