Cloud Composer and Apache Airflow

On Wednesday, October 28, Aisha opens the Airflow web interface for Birchline's one managed Airflow environment. The data engineering team created it in early 2025 to run the warehouse system's nightly exports, which involve a file server, an FTP drop from a supplier and three BigQuery steps. Mateo has seen the word Composer in the exam guide and wants to know what it is, when Birchline would use it, and why Aisha calls it 'the tool for when the simple tools run out'.

Airflow and its managed service

Apache Airflow is an open-source platform for writing, scheduling and monitoring workflows as code. A workflow, which Airflow calls a DAG, is a Python file that defines tasks and the dependencies between them. Airflow is widely used, has hundreds of ready-made connectors called operators, and can coordinate almost any system that has an API.

Running Airflow yourself means running a scheduler, workers, a web server, a database and storage for the DAG files, and keeping all of them patched and scaled. Google's managed version does that for you. For years it was called Cloud Composer, and the exam guide still uses that name. In April 2026 Google announced that Cloud Composer is becoming Managed Service for Apache Airflow; the product, the console pages and the gcloud composer commands carry on, so expect both names in documentation for a while. This course uses Composer and managed Airflow interchangeably.

Environments and versions

An Airflow installation in Google Cloud is called an environment. Each environment has a Cloud Storage bucket with a dags folder, a set of Airflow components, and an Airflow version. The service has gone through three generations. The first used a Google Kubernetes Engine cluster in your own project and is now past its end of life. The second runs on an Autopilot GKE cluster in your project and supports Airflow 2. The third, which Birchline uses, does not put a cluster in your project at all, scales workers automatically, simplifies networking, and supports both Airflow 2 and Airflow 3. Each Airflow build is supported for about a year after release, so environments need upgrading from time to time.

An environment is billed while it exists, whether or not any DAG is running, because its scheduler, web server and database run all the time. Birchline's small environment costs a few hundred dollars a month before any task runs. That fixed cost is the main reason Aisha does not use Airflow for pipelines that a scheduled query or a short workflow could handle, a point lesson 11 returns to.

A DAG file

Aisha opens the DAG that loads the supplier's nightly stock file. It is a Python file, but it does not process data itself; it describes tasks and leaves the work to the services they call.

# dags/supplier_stock.py  (runs in Airflow, not in the browser)
import pendulum
from airflow import DAG
from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor
from airflow.providers.google.cloud.transfers.gcs_to_bigquery import GCSToBigQueryOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG(
    dag_id="supplier_stock",
    schedule="0 4 * * *",                          # 04:00 every day
    start_date=pendulum.datetime(2026, 1, 1, tz="America/Toronto"),
    catchup=False,
    default_args={"retries": 3, "retry_delay": pendulum.duration(minutes=10)},
) as dag:
    wait = GCSObjectExistenceSensor(
        task_id="wait_for_file",
        bucket="bl-data-landing",
        object="suppliers/stock_{{ ds_nodash }}.csv",
        mode="reschedule", timeout=3 * 60 * 60)
    load = GCSToBigQueryOperator(
        task_id="load_raw",
        bucket="bl-data-landing",
        source_objects=["suppliers/stock_{{ ds_nodash }}.csv"],
        destination_project_dataset_table="bl-data-prod.raw.supplier_stock_{{ ds_nodash }}",
        write_disposition="WRITE_TRUNCATE", skip_leading_rows=1)
    merge = BigQueryInsertJobOperator(
        task_id="merge_curated",
        configuration={"query": {"query": "CALL curated.merge_supplier_stock(DATE '{{ ds }}')",
                                 "useLegacySql": False}})
    wait >> load >> merge

Each part of the file has a job. The DAG's schedule says when runs start, written in cron syntax. The start date and catchup setting decide whether Airflow creates runs for past dates it has missed; with catchup off, it only schedules from now on. The default arguments give every task three retries, ten minutes apart. The sensor waits for the supplier's file, checking periodically and giving up after three hours. The two operators run a BigQuery load and a stored procedure. And the last line, wait >> load >> merge, sets the dependencies.

The double curly brackets are templates. When a run starts, Airflow replaces {{ ds }} with the run's logical date, such as 2026-10-27, and {{ ds_nodash }} with the same date without dashes. This is Airflow's version of the run-date parameter from lesson 2, and it is what makes reruns and backfills process the right day. Airflow 3 renames some of these ideas, for example using 'logical date' consistently, but the principle is the same.

Operators, sensors and retries

An operator is a template for one kind of task: run a BigQuery job, start a Dataflow template, copy files, call an HTTP endpoint, run a Dataform workflow. Google maintains a large provider package of operators for its services. A sensor is a special operator that waits for something to be true, such as a file existing or a partition appearing. Sensors that occupy a worker slot while they wait can be wasteful, so modern DAGs use reschedule mode, which frees the slot between checks, or deferrable sensors, which hand the waiting to a lightweight process called the triggerer.

Retries handle temporary failures such as a network timeout or a service briefly refusing requests. Aisha sets them deliberately: three retries with growing delays for tasks that call external systems, and none for tasks that fail only when the data is wrong, because retrying a bad file just fails three more times. Mateo works out how long the retry schedule can delay a run.

# Total extra wait when a task fails and is retried with exponential backoff
first_delay = 10      # minutes before the first retry
factor = 2            # each delay doubles
max_delay = 60        # but never more than an hour
retries = 4

delay, total = first_delay, 0
for attempt in range(1, retries + 1):
    total += delay
    print(f"retry {attempt}: waits {delay} min, {total} min after the first failure")
    delay = min(delay * factor, max_delay)

retry 1: waits 10 min, 10 min after the first failure
retry 2: waits 20 min, 30 min after the first failure
retry 3: waits 40 min, 70 min after the first failure
retry 4: waits 60 min, 130 min after the first failure

The answer, 130 minutes, is why retries and deadlines must be planned together. If the nightly pipeline must finish by 06:00 and the task normally runs at 04:00, four retries with this schedule could leave just a few minutes, or none at all, for the tasks after it. Aisha's rule is that retries cover short glitches; anything longer should alert a person.

When Airflow is the right choice

Managed Airflow earns its cost when a pipeline coordinates many systems, branches on conditions, needs sensors that wait for things outside Google Cloud, must be backfilled over long periods, or is one of dozens of pipelines that benefit from a single scheduler, interface and history. Birchline's warehouse exports qualify: they touch an on-premises file server, a supplier's FTP drop and BigQuery, and the operations team likes seeing every run in one place.

It is the wrong choice for a single nightly SQL statement, for a handful of BigQuery transformations that Dataform can order by itself, or for an event-driven step that should run seconds after a file arrives. It is also wrong when nobody on the team can write and review Python, because every DAG is code. Aisha keeps a short list on the team wiki: if a new pipeline does not need at least two of sensors, branching, many external systems and long backfills, it does not go into Airflow.

Testing and deploying DAGs

Because DAGs are code, Birchline treats them like code. They live in a Git repository, changes go through a pull request that another engineer reviews, and an automated check imports every DAG file before it is merged. That import check catches the most common failure: a DAG that does not parse, perhaps because of a typing mistake or a missing library, never appears in the Airflow UI at all, and its runs simply stop. Aisha has seen that happen twice, both times on a Friday afternoon.

Deployment is a copy. When a change is merged, the build pipeline copies the DAG file into the dags folder of the environment bucket, and the scheduler picks it up within a few minutes. Python packages that DAGs need are installed on the environment itself, and a change to them restarts the components, so package updates are scheduled for quiet periods. Birchline keeps a second, smaller environment for testing upgrades to new Airflow versions before the production environment moves.

Two settings keep a busy environment fair. Pools limit how many tasks of a certain kind run at once, for example no more than four tasks that call the supplier's slow API. Priority weights decide which waiting tasks start first when workers are scarce, so the nightly sales tasks win over a weekend backfill. Together with the cap on concurrent runs from lesson 1, they stop one large job from starving everything else.

Mateo asks how the team knows a DAG change worked. Aisha shows him the habit she insists on: after every deployment, open the DAG in the Airflow UI, confirm the new version appears in its code view, and trigger one run against a test date before the next scheduled run. A deployment that nobody has watched run once is not finished.