GCP Data Engineering · Chapter 04

Orchestrating Batch Loads with Cloud Composer

Chapter 3 loaded data by hand. Orchestration automates it: what runs, in what order, on what schedule, and what happens when something is late or fails. Airflow is the tool, and Cloud Composer is Airflow managed by Google.

0 of – checks answered

How Airflow thinks

Airflow does three jobs: chain tasks by their dependencies, schedule them, and integrate with other systems. All three are written in Python. Learning Cloud Composer is, practically, learning Airflow.

A workflow written as Python

A directed acyclic graph is a set of tasks chained by their dependencies, plus a schedule. "Directed" means arrows have a direction; "acyclic" means no loops. Every DAG has an ID that must be unique across the whole environment.

Defined by: DAG ID, owner, start date, schedule
Each execution: A "DAG Run"
Written in: Python, not drag-and-drop
Why code: Testable, versioned in git, automatable

The arrows: what runs after what

Airflow writes dependencies with the bit-shift operator. a >> b means b runs after a. A list means "wait for all of these": [a, b] >> c runs c only when both a and b have finished.

Chain: a >> b >> c
Fan-in: [a, b] >> c
Independent tasks: Run in parallel when their inputs are ready
Shown in UI: Tree View and Graph View

When it runs, and the counter-intuitive part

The schedule uses cron syntax: five fields for minute, hour, day of month, month and day of week. The tricky bit: a run covers a time interval and fires after that interval ends. So to run "today" at midnight you set the start date to yesterday.

Daily at 05:00: 0 5 * * *
Every minute: * * * * *
Start date: When you expect the first interval to begin
Execution date: The interval a run is responsible for, not "now"

Run the bike-sharing DAG

This is the Level 3 pipeline from the book: three tables loaded in parallel, then the fact and dimension tables built once everything has landed, each followed by a row-count check. Click a ready task to run it, or run everything that is ready at once.

Grey tasks are waiting on upstream tasks, blue ones are ready, green ones are done. Tasks with no dependency on each other can run side by side, which is why "run all ready" finishes the graph in far fewer rounds than one-at-a-time.

The five levels

The chapter builds orchestration maturity step by step. You can follow every practice or none; Airflow will not stop you. Each level adds one idea.

Create a DAG and deploy it

A dummy workflow with two Bash tasks. Learn how a DAG, task and dependency look in code, how timing works, and how to deploy by copying the file to the environment’s dags/ folder.

New idea: DAG, tasks, >>, deployment
Watch out: Start date and schedule interplay
Tip: Version DAG names (_v1) when there is no built-in versioning
Tools: Airflow UI: Clear, View Log

Time, reruns and idempotency

Data pipelines constantly touch the past. Three Airflow terms, one confusing date rule, and one property that makes reruns safe.

Start date and catchup calculator

Start date Today Schedule Catchup

The key idea: a run is labelled by the interval it covers (its execution date) and fires when that interval has ended. So the "today" run happens tomorrow, and a start date of yesterday makes a daily DAG run straight away.

Backfill, rerun or catchup?

01 A pipeline has run daily for a week. The business now wants the previous month loaded too, without disturbing the running schedule.
02 Last Tuesday’s load failed. You click Clear on that run in the Airflow UI.
03 You deploy a new DAG whose start date is two weeks ago, and Airflow immediately launches runs for every missed day.

Is this task idempotent?

A task is idempotent if running it again gives the same result. That is what makes reruns and backfills safe.

01 Each daily run appends that day’s events to the whole table with WRITE_APPEND. A run is repeated.
02 Each daily run writes with WRITE_TRUNCATE to only that day’s partition of a date-partitioned table.
03 Each run truncates and reloads the entire 1 TB table from all files.

Partitioning divides a table into segments by a key: a date or timestamp column, ingestion time, or an integer range. Daily is the usual grain. Loading to table$20210101 targets a single day’s partition, so truncating rewrites one day, not the whole table. Partitioning also cuts query cost (Chapter 10).

Best practice

Python inside a DAG can do anything, which is exactly the danger. Keep DAGs as configuration that triggers other systems.

Good practice or avoid?

01 Use the BigQuery native operators for every load and transform step.
02 Download a Cloud Storage file to the worker’s disk with the storage client, then process it in Python.
03 Set priority='BATCH' on BigQuery query tasks in pipelines.
04 Load a BigQuery table into a pandas DataFrame inside a task for a quick calculation.

Where should this value live?

01 A bucket name and dataset names shared by many DAGs, editable in the Airflow UI as one JSON value.
02 A constant that only one DAG needs, written at the top of its script.
03 A Cloud SQL instance name supplied when the Composer environment is created or updated.

	Level	Set where
DAG variable	One DAG only	In the DAG script
Airflow variable	All DAGs	Airflow UI, Admin then Variables (JSON recommended)
Environment variable	Whole environment	Composer environment settings or at deployment
Macro	Per DAG run	Filled in by Airflow, for example the execution date

Other habits from the chapter: keep table schemas in JSON files under the environment’s data/ folder, version DAG names when you change them, add row-count check tasks so an empty result fails the run, and set use_legacy_sql to false.

Hands-on: the five levels

Do these in your own project. The main cost is the environment itself, so work in a focused block of time and delete it at the end.

Create a Cloud Composer environment
Environments cost money for as long as they exist, not per DAG or task. Create one only when you are ready to work, and delete it afterwards. Pick a current image version and a region close to your bucket and BigQuery.
Open the Airflow web UI and find the example DAG
Look at Tree View, Graph View, the run indicators (circles are runs, squares are tasks) and the Code button.

Level 1: deploy a two-task Hello / World DAG

Write the DAG, then copy it to the environment’s dags/ folder.


						gcloud composer environments storage dags import \
  --environment YOUR_ENV --location YOUR_REGION \
  --source level_1_dag.py

Practise Clear and View Log
Clear a run circle (reruns the DAG) and a task square (reruns that task), then read a task log.
Level 2: Cloud SQL → GCS → BigQuery → transform
Chain three native-operator tasks. Use standard SQL and batch priority. Recreate the Cloud SQL source only if you need it, and delete it afterwards.
Level 3: variables, macros and row-count checks
Create one Airflow variable holding JSON, use the execution-date macro, load three tables in parallel, then build the fact and dimension tables and add a check task after each.
Level 4: make the trips load idempotent
Partition the table by date and truncate only the run date’s partition. Rerun a day and prove there are no duplicates.
Level 5: sensor and signal file
Have the first DAG write a success file; have the second start on the same schedule but wait on it.
Delete the environment (and Cloud SQL) when you finish
This is the step that saves you money.

Key terms

Orchestration: Automating tasks, their dependencies and their schedule.
Airflow: Open-source workflow tool where workflows are Python code.
Cloud Composer: Google’s managed Airflow.
DAG / DAG Run: A workflow definition, and one execution of it.
Operator / task: A connector to a system, and one configured use of it.
Cron: Five-field schedule syntax: minute, hour, day, month, weekday.
Execution date: The interval a run is responsible for, not the wall-clock time.
Backfill / rerun / catchup: Load the past; repeat something that ran; auto-run missed intervals.
Idempotent: Safe to run repeatedly with the same result.
Partition: A segment of a table split by a date, timestamp or integer key.
Sensor: A task that waits for a condition before continuing.
Macro: A template value such as the execution date, filled in at run time.

Check yourself

Pick an answer to see the reasoning straight away.

What is a DAG in Airflow?

Where must a DAG file go for Composer to pick it up and run it?

Why is it a bad idea to process large data with pandas inside a DAG task?

A run for one day fails and you rerun it. Which load design leaves the table correct?

Why use a sensor between dependent DAGs instead of a later fixed schedule?

You want a daily-midnight DAG to run for today’s date as soon as you deploy it. What start date follows Airflow’s rule?

Study notes in my own words, following the structure of Adi Widjaja's Data Engineering with Google Cloud Platform (Packt, 2022). GCP changes quickly, so check each service's current documentation before relying on names, prices or console steps.