Apache Airflow: Orchestrating ETL Data Pipelines in Python

TL;DR. Apache Airflow is not a fancy cron job. It’s an orchestrator that models a pipeline as a graph of tasks (a DAG) and handles dependencies, retries, backfills and monitoring from a web UI. This guide covers the core concepts, the Airflow 3 architecture, a complete Python ETL pipeline for a real smart agriculture case (sensors → InfluxDB → PostgreSQL → Grafana), error handling with retries and trigger rules, a Docker setup, and when Airflow is the wrong tool.

It almost always starts with a cron job. A Python script runs at 2 a.m., downloads the weather station CSV, cleans it and loads it into the database. It works for months. Then the vendor renames a column, the script fails, and nobody notices. Three weeks later the agronomist asks why the dashboard has a gap right during the late-frost window. That’s when you discover there are seventeen cron jobs on three servers, and nobody is quite sure which one depends on which.

Apache Airflow exists to solve exactly that. Created at Airbnb in 2014 and now a top-level Apache Software Foundation project, it has become the de facto standard for orchestrating data pipelines. In this installment of our Digital Stack 2026 series it closes the loop: the data Kafka moves and InfluxDB stores has to be processed, aggregated and synced with other systems. Something needs to coordinate all of it.

What Airflow is (and isn’t)

Airflow is an orchestrator: it decides what runs, when, in which order, and what happens when something breaks. It is not a data processing engine. The heavy lifting — reading a million-row CSV, computing an aggregation — is done by Python, the database, or an external tool; Airflow launches that work, checks the outcome and decides the next step.

Equally important is what Airflow is not: a streaming tool. If you need to react to an event within milliseconds, the answer is Kafka. Airflow works in discrete runs — hourly, nightly, when a file lands — and shines at batch. The two are complementary, not competing.

Airflow 3: what changed

If your mental model comes from pre-2025 tutorials, a few things are different. Airflow 3.0 (April 2025) was the biggest release in the project’s history, and by 2026 we’re on the 3.3 line. What matters when you’re getting started:

• A new React UI, faster, with redesigned Grid and Graph views.
• DAG versioning: every run is tied to the code version that produced it, and the UI shows the change history.
• Task SDK: DAGs import from airflow.sdk, a stable interface decoupled from Airflow internals.
• Assets (formerly Datasets): a DAG can start when another DAG updates some data, not just on a clock.
• SLAs are gone, Deadline Alerts are in (since Airflow 3.1) to warn you when a run is running late.
• Airflow 3.3 added durable task state across retries and an experimental SDK for writing tasks in Java or Go.

Core concepts: DAGs, tasks and operators

DAG: the pipeline graph

DAG stands for Directed Acyclic Graph. Directed because every edge has a direction: task B runs after task A. Acyclic because you can’t loop back: no task may depend, directly or indirectly, on itself. In Airflow a DAG is a Python file describing tasks and dependencies plus a few properties: when it runs (schedule), from when (start_date), how many times to retry on failure.

Tasks and operators

A task is a single step. You define it with an operator, a ready-made template for a kind of work:

• PythonOperator (or the @task decorator): runs a Python function. You’ll use it 80% of the time.
• BashOperator: runs a shell command — handy for existing scripts or CLI tools.
• SQLExecuteQueryOperator: runs a query against a database configured as a connection (PostgreSQL, MySQL, Snowflake…).
• Sensors: special tasks that wait for a condition, like a file appearing in a folder or an S3 bucket.

Hundreds more live in the official provider packages for AWS, Google Cloud, Azure, dbt, Slack and many other services.

Dependencies

You declare dependencies with >>: extract >> validate >> load means “extract, then validate, then load”. With the TaskFlow API, when a task uses another task’s return value, the dependency is created automatically.

Airflow 3 architecture

Knowing the components helps you debug when something doesn’t start.

• DAG processor: parses the Python files in dags/ into a serialized representation.
• Scheduler: the brain. It checks which DAGs are due and which tasks have their dependencies met.
• Executor: decides where tasks run — in-process (LocalExecutor), on a worker fleet (CeleryExecutor) or in dedicated pods (KubernetesExecutor).
• Workers: the processes that actually execute task code. In Airflow 3 they talk through an API and no longer hit the database directly.
• Metadata database: usually PostgreSQL. Holds the state of every run, log references and connections.
• API server: serves the web UI and the REST API.

Tutorial: a four-stage smart agriculture pipeline

The use case comes from one of the capstone projects in my Advanced Data Modeling course at Politecnico di Torino. A group of wineries collects data from in-field sensors. Every night a CSV with the day’s readings lands in a staging folder. The pipeline must:

1. Extract and validate the CSV (expected columns, plausible values, no duplicates).
2. Load the readings into InfluxDB, tagged by farm, plot, sensor.
3. Compute daily KPIs per plot: min/max temperature, Growing Degree Days, water-stress indicators.
4. Write the KPIs to PostgreSQL next to historical yields and field interventions, and raise an alert if there’s frost risk.

Here’s the full DAG using the TaskFlow API:

from datetime import datetime, timedelta
import pandas as pd
from airflow.sdk import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook

STAGING = "/opt/airflow/data/staging"

@dag(
    schedule="@daily",
    start_date=datetime(2026, 3, 1),
    catchup=False,
    default_args={"retries": 3, "retry_delay": timedelta(minutes=10)},
    tags=["agro", "iot"],
)
def smart_agriculture_pipeline():

    @task
    def extract_and_validate(ds=None):
        df = pd.read_csv(f"{STAGING}/sensors_{ds}.csv", parse_dates=["timestamp"])
        expected = {"timestamp", "farm", "plot", "sensor", "air_temp", "soil_temp", "soil_moisture"}
        missing = expected - set(df.columns)
        if missing:
            raise ValueError(f"Missing columns: {missing}")
        df = df.drop_duplicates().query("-30 < air_temp < 50")
        path = f"{STAGING}/clean_{ds}.parquet"
        df.to_parquet(path)
        return path  # pass the path, not the data

    @task
    def load_to_influx(path):
        from influxdb_client import InfluxDBClient, WriteOptions
        df = pd.read_parquet(path).set_index("timestamp")
        with InfluxDBClient.from_env_properties() as client:
            with client.write_api(write_options=WriteOptions(batch_size=5000)) as w:
                w.write(bucket="agro", record=df, data_frame_measurement_name="sensors",
                        data_frame_tag_columns=["farm", "plot", "sensor"])
        return path

    @task
    def compute_daily_kpi(path):
        df = pd.read_parquet(path)
        kpi = df.groupby(["farm", "plot"]).agg(
            t_min=("air_temp", "min"), t_max=("air_temp", "max"),
            soil_min=("soil_temp", "min"), moisture_avg=("soil_moisture", "mean"))
        kpi["gdd"] = ((kpi.t_min + kpi.t_max) / 2 - 10).clip(lower=0)
        return kpi.reset_index().to_dict("records")

    @task
    def write_to_postgres(rows, ds=None):
        hook = PostgresHook(postgres_conn_id="agro_pg")
        hook.insert_rows("daily_kpi", [(ds, r["farm"], r["plot"], r["t_min"], r["t_max"],
                          r["gdd"], r["moisture_avg"]) for r in rows],
                         replace=True, replace_index=["day", "farm", "plot"])
        return rows

    @task.short_circuit
    def frost_risk(rows):
        return any(r["soil_min"] < 2 for r in rows)

    @task
    def send_frost_alert(rows):
        plots = [r["plot"] for r in rows if r["soil_min"] < 2]
        print(f"FROST ALERT for plots: {plots}")  # e.g. email, Slack, Teams...

    path = extract_and_validate()
    kpi = compute_daily_kpi(load_to_influx(path))
    saved = write_to_postgres(kpi)
    frost_risk(saved) >> send_frost_alert(saved)

smart_agriculture_pipeline()

Three details that make the difference

• Pass references, not data. Values move between tasks through XCom, stored in the metadata database. That’s why the first task returns the Parquet file path, not the DataFrame. Pushing a million rows through XCom is one of the most common mistakes.
• The logical date ds. Every run knows which day it’s for. Rerun March 15 and the DAG reads March 15’s file, not today’s.
• Idempotent writes. insert_rows with replace=True overwrites the same day’s rows instead of duplicating them. Run the DAG ten times and the result is the same.

Scheduling, catchup and backfill

The schedule parameter accepts presets (@hourly, @daily, @weekly), cron expressions ("0 2 * * *") or assets: “start when the import DAG has updated table X”.

The key concept is the data interval. A daily DAG running overnight between March 15 and 16 processes March 15’s data: its logical date is the 15th. Set catchup=True with a start_date three months ago and Airflow creates every missing run. A backfill is the same thing on demand: “recompute all of March with the new GDD formula”. This is where idempotency stops being a nice-to-have and becomes a requirement.

Handling failures: retries, trigger rules and alerts

In a real pipeline something always fails: the network, a slow external API, a late file. Airflow gives you three lines of defense.

Automatic retries

With retries=3 and a 10-minute retry_delay, transient errors fix themselves. Turn on retry_exponential_backoff to lengthen the wait progressively — useful with rate-limited APIs.

Trigger rules

By default a task runs only if every upstream task succeeded (all_success). Sometimes you want something else:

Callbacks and Deadline Alerts

on_failure_callback runs a function when a task fails for good — typically a Slack or Teams message linking to the logs. Airflow 3’s Deadline Alerts warn you when a run hasn’t finished within a set time: if the agronomist’s report must be ready by 7:00, you want to know at 6:30 that it’s late, not at 8:00.

Monitoring from the UI

The most visible difference from cron is visibility. The web UI shows every run of every DAG as a color-coded grid: green succeeded, red failed, orange retrying. Click a red square and you land on that task’s log for that date. You can rerun a single task, mark it successful or launch a backfill without touching a terminal.

Running it with Docker

The fastest way to try Airflow locally is the official docker-compose.yaml from the docs:

mkdir airflow-lab && cd airflow-lab
curl -LfO "https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml"
mkdir -p ./dags ./logs ./plugins ./config
echo -e "AIRFLOW_UID=$(id -u)" > .env
docker compose up airflow-init
docker compose up -d
# UI at http://localhost:8080 (user/password: airflow)

Mind the resources: Docker needs at least 4 GB of RAM, ideally 8. On student laptops that’s a real constraint, so the course notes suggest lighter options when needed: Prefect (with a free cloud tier) or, for very simple cases, APScheduler inside a Python app. For a single machine there’s also airflow standalone, which runs everything in one process on SQLite — fine for learning, never for production.

A second case: public transport punctuality

Another course project shows a typical pattern. The city’s GTFS timetable lives in PostgreSQL; simulated vehicle positions and delays flow through Kafka into InfluxDB. Every night a DAG computes On-Time Performance per line: extract_from_influxdb >> compute_otp >> write_to_pg >> refresh_view. The interesting bit is step two: neither database can join actual delays (InfluxDB) with scheduled times (PostgreSQL) on its own. Python does it inside the operator, and Airflow guarantees it only happens once both sources are ready. That’s the multi-system orchestration an InfluxDB-internal task can’t offer.

Airflow, cron or n8n? Picking the right tool

If your problem is “when an email with an attachment arrives, save it to Drive and ping sales”, Airflow is overkill — a tool like the ones compared in our n8n vs Make.com guide will do. If you have several databases to keep aligned, history to recompute and a real need to know what happened on every run, Airflow pays back its learning curve many times over.

Best practices at a glance

• Small, atomic tasks: one task, one responsibility. Easier to rerun and debug.
• Always idempotent: rerunning a task for the same date must give the same result.
• No heavy data in XCom: pass paths, keys, IDs.
• No heavy logic at the top level of the file: the DAG processor re-parses it constantly, so queries and API calls belong inside tasks.
• Credentials in Airflow Connections or a secrets manager, never in code.
• Use the logical date, never datetime.now(), to decide which data to process.

The picture comes together

With Airflow, the series stack is complete: forms and spreadsheets to collect, Kafka to move data in real time, InfluxDB for time series, Neo4j for relationships, Grafana to display, Airflow to coordinate. In the next article in the series we step back and look at the whole design: why leading companies run four or five different databases, and how they keep them consistent.

Got nightly scripts nobody dares touch, or data that should refresh daily but actually refreshes “whenever someone remembers”? Request a data and AI analysis: we’ll map your current pipelines and figure out together whether you need Airflow or just some order.