Maintaining and automating data workloads · Topic 2 of 5

Cloud Composer DAGs and Repeatable Scheduling

Write Cloud Composer DAGs with operators, sensors, retries and templated dates, make tasks idempotent for backfills, and pick a scheduler for each job.

5 min

In a nutshell

A DAG is tasks plus dependencies; operators do work, sensors wait.
Use templated logical dates such as {{ ds }} so reruns process the right day.
Idempotent tasks make retries and backfills safe.

What it is

Cloud Composer is managed Apache Airflow. You describe a workflow as a directed acyclic graph (DAG) in Python: tasks, the order they run in, and when the whole graph starts. Repeatability means each run produces the same result for the same date, however many times it runs, so retries and backfills are safe.

How it works

DAG building blocks

Concept	Meaning
DAG	The workflow: tasks plus dependencies, with no cycles
Operator	A task type, such as `BigQueryInsertJobOperator` or a Dataflow or Dataproc operator
Sensor	A task that waits for a condition, such as a file in Cloud Storage
Dependencies	Set with `>>`, for example `extract >> load >> check`
`schedule`	A cron expression or preset such as `@daily`
`start_date` and `catchup`	When scheduling begins and whether missed past runs are created
`retries` and `retry_delay`	Automatic retries for transient failures
Templated dates such as `{{ ds }}`	The run's logical date, so each run processes its own day

A DAG for a daily load

from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceSensor
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryCheckOperator, BigQueryInsertJobOperator)

with DAG(
    dag_id="daily_sales",
    start_date=datetime(2026, 1, 1),
    schedule="0 3 * * *",
    catchup=False,
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
) as dag:
    wait_for_file = GCSObjectExistenceSensor(
        task_id="wait_for_file",
        bucket="landing-bucket",
        object="sales/{{ ds }}/_SUCCESS",
        mode="reschedule",
    )
    load = BigQueryInsertJobOperator(
        task_id="load_partition",
        configuration={"query": {
            "query": "CALL sales.load_day('{{ ds }}')",
            "useLegacySql": False,
        }},
    )
    check = BigQueryCheckOperator(
        task_id="check_rows",
        sql="SELECT COUNT(*) > 0 FROM sales.daily WHERE sale_date = '{{ ds }}'",
        use_legacy_sql=False,
    )
    wait_for_file >> load >> check

The sensor runs in reschedule mode, so it frees its worker slot between checks. The check task fails the run if the day's partition is empty, which catches silent failures.

Making runs repeatable

Practice	Why
Idempotent writes: overwrite the day's partition or `MERGE` on a key	A retry or rerun replaces data instead of duplicating it
Use the logical date, not "now"	A rerun for 3 October processes 3 October, even if run later
Keep tasks small and single-purpose	Failed steps retry alone
Let services do the heavy work	Composer orchestrates; BigQuery, Dataflow and Dataproc process
Store DAGs in Git and deploy with CI/CD	Every environment runs the same tested code
Pass only small metadata between tasks (XCom)	Large data belongs in Cloud Storage or BigQuery

Backfilling runs a DAG for past logical dates, for example after a bug fix. It is only safe when tasks are idempotent and use the templated date.

Exam tip

When a stem says "reruns create duplicate rows", the fix is idempotency: write with partition overwrite or MERGE. Adding more retries makes the problem worse.

Scheduling and orchestrating jobs in a repeatable way

Need	Choose
Many dependent tasks, sensors, backfills	Cloud Composer
One SQL statement on a schedule	BigQuery scheduled query
SQL transformation graph in BigQuery	Dataform workflow configuration
A few API calls, serverless	Workflows, started by Cloud Scheduler or Eventarc
A fixed sequence of Spark jobs on a temporary cluster	Dataproc workflow template
Start work when a file arrives	Event trigger (Eventarc) or a sensor in a DAG

A worked example

A team's cron scripts on a VM run five steps each night. When one fails, someone reruns the rest by hand, and reruns double-load data. The engineer replaces the scripts with a Composer DAG.

A sensor waits for the day's success file, and each load writes to the logical date's partition with overwrite. A row-count check closes the DAG, and retries handle transient errors. The DAG lives in Git and deploys through CI, and missed days are backfilled without duplicates.

How the exam asks about this

"A workflow must wait for a file, run three dependent jobs and retry failures automatically." A Cloud Composer DAG with a sensor, operators and retries.
"Rerunning last week's DAG runs created duplicate rows." Make tasks idempotent with partition overwrite or MERGE keyed on the logical date.
"Only one SQL statement must run every hour." A BigQuery scheduled query, not a Composer environment.

What to remember for the exam

A DAG is tasks plus dependencies; operators do work, sensors wait.
Use templated logical dates such as {{ ds }} so reruns process the right day.
Idempotent tasks make retries and backfills safe.
Composer orchestrates; the data services do the processing.
Pick the lightest scheduler that meets the need.

Watch out

Using datetime.now() inside a task breaks repeatability. A rerun next week would process next week's data instead of the day it was meant for.

Common mistakes

Processing large data inside Airflow workers. Workers are for orchestration; push work to BigQuery, Dataflow or Dataproc.

Leaving catchup on with an old start date. Airflow schedules every missed run at once when the DAG is enabled.

Appending on every run. Retries then duplicate data; overwrite partitions or merge.

Check yourself

What does catchup control in an Airflow DAG?
Whether Airflow creates runs for past schedule intervals between the start date and now when the DAG is enabled.
Why should a DAG task use the logical date instead of the current time?
So that retries, reruns and backfills process the intended day and give the same result every time.
What is the benefit of a sensor in reschedule mode?
It releases its worker slot between checks instead of occupying it while waiting.

Topic 2 of 5 in Maintaining and automating data workloads

Test yourself on Maintaining and automating data workloads

The full Google Cloud Professional Data Engineer question bank has 733 questions, with timed practice exams and an explanation for every answer.

Subscribe Try Sample Exam

This study guide is independent and is not affiliated with, endorsed by, or sponsored by Google Cloud. Vendor names, certification names and exam codes are trademarks of their respective owners and are used for identification only.