Building a Data Pipeline on GCP End to End

Learning Google Cloud's data services one at a time leaves a gap: none of the individual tutorials says how they fit together. A real data pipeline on GCP is a chain of small, dull, reliable steps, and its quality lies in what happens on a bad day: when a file arrives half empty, when a job is run twice, when yesterday has to be reprocessed.

This guide builds one batch pipeline end to end. It names the service for each stage, then runs the whole flow on real data to show the two behaviours that matter most.

The shape

Three tables, not one. That is the first design decision, and each has a job:

• Raw holds exactly what the source sent. It is the evidence, and the starting point when logic changes.
• Staging holds the cleaned version: consistent names and types, impossible rows removed.
• Reporting holds what people query. Nothing reaches it that has not passed the checks.

Stage by stage

Land. The source system writes each day's files under a dated prefix, such as trips/dt=2024-01-15/, in a Cloud Storage bucket (Google Cloud: Cloud Storage overview). Files are never edited; a correction is a new delivery. See Google Cloud Storage.

Load raw. A load job writes the day's files into the day's partition, replacing whatever was there (Google Cloud: load data into partitioned tables). Because load jobs are atomic, the partition holds the old day or the new day, never a mixture (Google Cloud: load data from Cloud Storage). The details are in Cloud Storage to BigQuery.

Clean. A SQL statement builds the staging partition from the raw one. It selects the columns that matter, gives them clear names and drops rows that cannot be right.

Check. Before anything is published, queries test the result. A SQL transformation tool such as Dataform runs these as assertions in the same workflow (Google Cloud: Dataform overview). See Dataform vs dbt.

Publish. Only if the checks pass does the reporting partition get replaced.

The whole flow, on real data

The script is that pipeline in miniature. The SQL is GoogleSQL, translated by the sqlglot library and run on DuckDB, which stands in for BigQuery. The data is New York yellow taxi trips for 15 and 16 January 2024 (NYC TLC: trip record data). Nothing here runs on Google Cloud.

It runs four times: a normal day, the same day again, a day whose file arrived truncated, and that day again once the full file is delivered.

# An end-to-end daily pipeline, reproduced locally: land, load raw, clean into staging, check, publish.
# The SQL is GoogleSQL, translated by sqlglot and run on DuckDB with real trip data. DuckDB stands in
# for BigQuery and a view over the source file for the landing bucket; nothing runs on Google Cloud.
import duckdb
import sqlglot

con = duckdb.connect()
con.execute("""CREATE VIEW source AS SELECT CAST(tpep_pickup_datetime AS DATE) AS dt, *
               FROM read_parquet('../../dedata/yellow_tripdata_2024-01.parquet')""")
con.execute("CREATE TABLE raw_trips AS SELECT * FROM source WHERE false")
con.execute("""CREATE TABLE stg_trips (dt DATE, pickup_at TIMESTAMP, pickup_zone INTEGER, payment_type BIGINT,
               trip_miles DOUBLE, fare DOUBLE, total DOUBLE)""")
con.execute("CREATE TABLE daily_zone_revenue (dt DATE, pickup_zone INTEGER, trips BIGINT, revenue DOUBLE)")


def bq(sql, day):
    return con.execute(sqlglot.transpile(sql, read="bigquery", write="duckdb")[0], {"run_date": day})


def deliver(day, share=1.0):                       # the file the source system drops for one day
    con.execute(f"CREATE OR REPLACE TEMP TABLE landing AS SELECT * FROM source WHERE dt = DATE '{day}' "
                f"ORDER BY tpep_pickup_datetime LIMIT (SELECT CAST(COUNT(*) * {share} AS BIGINT) FROM source "
                f"WHERE dt = DATE '{day}')")


def run(day):
    # 1. Load raw: replace the day's partition with exactly what was delivered.
    bq("DELETE FROM raw_trips WHERE dt = @run_date", day)
    con.execute("INSERT INTO raw_trips SELECT * FROM landing")
    # 2. Staging: typed, renamed, and only rows that make sense.
    bq("DELETE FROM stg_trips WHERE dt = @run_date", day)
    bq("""INSERT INTO stg_trips
          SELECT dt, tpep_pickup_datetime, PULocationID, payment_type, trip_distance, fare_amount, total_amount
          FROM raw_trips
          WHERE dt = @run_date AND fare_amount > 0 AND trip_distance > 0
            AND tpep_dropoff_datetime > tpep_pickup_datetime""", day)
    # 3. Checks: the gate between building and publishing.
    raw, staged, previous = bq("""
          SELECT (SELECT COUNT(*) FROM raw_trips WHERE dt = @run_date),
                 (SELECT COUNT(*) FROM stg_trips WHERE dt = @run_date),
                 (SELECT SUM(trips) FROM daily_zone_revenue WHERE dt = DATE_SUB(@run_date, INTERVAL 1 DAY))""",
                                day).fetchone()
    checks = {"rows arrived": raw > 0,
              "at most 5% rejected": staged >= 0.95 * raw,
              "volume within 50% of yesterday": previous is None or 0.5 * previous <= staged <= 1.5 * previous}
    failed = [name for name, passed in checks.items() if not passed]
    # 4. Publish only if every check passed.
    if not failed:
        bq("DELETE FROM daily_zone_revenue WHERE dt = @run_date", day)
        bq("""INSERT INTO daily_zone_revenue
              SELECT dt, pickup_zone, COUNT(*), ROUND(SUM(total), 2) FROM stg_trips
              WHERE dt = @run_date GROUP BY dt, pickup_zone""", day)
    published = bq("SELECT COALESCE(SUM(trips), 0) FROM daily_zone_revenue WHERE dt = @run_date", day).fetchone()[0]
    status = "published" if not failed else "HELD: " + ", ".join(failed)
    print(f"{day}  raw {raw:6,}  staging {staged:6,}  rejected {raw - staged:5,}  in report {published:6,}  {status}")


from datetime import date
deliver("2024-01-15"); run(date(2024, 1, 15))
run(date(2024, 1, 15))                              # the same day again: nothing changes
deliver("2024-01-16", share=0.4); run(date(2024, 1, 16))     # a truncated delivery
deliver("2024-01-16"); run(date(2024, 1, 16))                # the full file arrives, and the day is rerun

2024-01-15  raw 77,033  staging 74,786  rejected 2,247  in report 74,786  published
2024-01-15  raw 77,033  staging 74,786  rejected 2,247  in report 74,786  published
2024-01-16  raw 37,223  staging 35,966  rejected 1,257  in report      0  HELD: volume within 50% of yesterday
2024-01-16  raw 93,057  staging 90,220  rejected 2,837  in report 90,220  published

Run 1: a normal day

77,033 rows arrive. Staging keeps 74,786 and rejects 2,247, about 3%, for having no fare, no distance or a drop-off before the pickup. All three checks pass and the reporting table is updated.

Notice that rejected rows are counted, not hidden. Raw still holds all 77,033. If someone later decides that zero-distance trips should count after all, staging can be rebuilt from raw without asking the source for anything.

Run 2: the same day again

Every figure is identical. Each stage deletes its day and writes it again, so a second run replaces the first instead of adding to it. This is the property that lets an orchestrator retry safely and lets a person rerun a day without fear.

Run 3: a truncated file

The file for the 16th arrives with only 37,223 rows. Nothing about it is malformed. It loads, it cleans, and a pipeline without checks would publish it, and the morning's dashboard would show trips down by half.

Here the volume check fails, because the day is less than half of the day before. The pipeline stops before publishing. The reporting table shows nothing for the 16th, which is the honest state: the data is not ready.

Run 4: the full file

The complete file is delivered and the day is rerun. 93,057 rows load, the checks pass and the day is published. No cleanup was needed between runs 3 and 4, because every stage replaces its partition.

The checks worth having

The example uses three, and they generalise:

• Something arrived. A row count above zero.
• Not too much was rejected. A ceiling on the share of rows dropped in cleaning. A jump in rejections means the source changed.
• Volume is plausible. Within a band around the previous day or the same weekday last week.

Add to them over time: uniqueness of keys, no nulls in required columns, totals that reconcile with the source. Keep each one cheap and specific. A check that fails for understandable reasons gets fixed; one that fails mysteriously gets disabled.

Orchestration and identity

Something must run the stages in order, retry what fails and tell someone. For a pipeline this size, the options are:

• Scheduled queries plus a scheduled load, if the stages are few and independent.
• A SQL transformation tool's own workflow, for the clean, check and publish stages.
• Managed Airflow, when files must be waited for, several pipelines depend on each other, or history and backfills matter (Google Cloud: Managed Airflow overview). See Cloud Composer.

Whichever runs it, run it as one dedicated service account that can read the landing bucket and write the three datasets, and nothing else (Google Cloud: service accounts overview). See service accounts and workload identity.

Two more settings are part of the pipeline, not extras. Partition all three tables by date, so that each run touches one day and costs accordingly; see partitioning and clustering. And alert on a held run, because a pipeline that stops silently is only slightly better than one that publishes bad data.

From batch to streaming

Everything above is a daily batch. If the business needs data within seconds, the landing stage becomes a message topic and the load becomes a streaming pipeline. The principles stay: keep the raw events, make every step repeatable, and check before you trust. See Pub/Sub to Dataflow to BigQuery.

Most pipelines should start as batch. It is simpler to build, cheaper to run and far easier to reason about, and many requirements described as real-time turn out to mean "this morning".

A build order

1. Create the bucket and one partitioned raw table. Load one day by hand.
2. Write the staging SQL. Count what it rejects and understand why.
3. Write three checks.
4. Write the reporting SQL.
5. Run two days in a row, then rerun the first. Confirm that nothing doubles.
6. Deliver a deliberately broken file. Confirm that it is held.
7. Only then schedule it.

Steps 5 and 6 are the ones people skip, and the ones that decide whether the pipeline can be trusted.

Building pipelines like this one is the core of the Data Engineering course.

Sources

• Cloud Storage overview (Google Cloud Documentation) — Google Cloud
• Load data into partitioned tables (Google Cloud Documentation) — Google Cloud
• Load CSV data from Cloud Storage (Google Cloud Documentation) — Google Cloud
• Dataform overview (Google Cloud Documentation) — Google Cloud
• Managed Airflow overview (Google Cloud Documentation) — Google Cloud
• Service accounts overview (Google Cloud Documentation) — Google Cloud
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission