Cloud Storage to BigQuery: Building an Ingestion Layer

An ingestion layer is an agreement. Whoever delivers files agrees where they go and what they look like. Whoever queries the tables is promised that each day's data is there once, complete, and reloadable if something went wrong. Most of the work in moving data from Cloud Storage to BigQuery is honouring that agreement, and almost none of it is the load command itself.

This guide builds the layer in five decisions, then checks two of them against a real file.

Decision 1: a layout BigQuery can read

Give every dataset one prefix and every day one folder, written as a key and a value:

gs://example-landing/trips/dt=2024-01-14/part-000.parquet
gs://example-landing/trips/dt=2024-01-15/part-000.parquet
gs://example-landing/trips/dt=2024-01-15/part-001.parquet
gs://example-landing/trips/dt=2024-01-16/part-000.parquet

This is the Hive partitioning layout, and BigQuery understands it: the key in the path can become a column, and one day can be selected by its prefix (Google Cloud: load externally partitioned data). The same documentation is explicit about what does not work. A path such as trips/2024-01-15/ that omits the key name is not a supported layout, and keys must appear in the same order in every path.

Two further rules cost nothing now and a great deal later:

• Same location. Keep the bucket in the same location as the dataset.
• Never edit a delivered file. A correction is a new delivery of the whole day, not a change in place.

Decision 2: one day, one partition

Create the raw table partitioned by the day of the event, and make queries state which days they want (Google Cloud: create partitioned tables):

CREATE TABLE raw.trips (
  pickup_at     TIMESTAMP,
  dropoff_at    TIMESTAMP,
  pickup_zone   INT64,
  fare_amount   NUMERIC
)
PARTITION BY TIMESTAMP_TRUNC(pickup_at, DAY)
OPTIONS (require_partition_filter = TRUE);

Declare the schema here, in version control, not by auto-detection at load time. The raw table holds what the source sent, typed and unmodified. Cleaning belongs to the next layer.

Decision 3: loads that replace, not append

A load job appends by default. That default is what turns an innocent rerun into doubled revenue. Load each day into its own partition and replace it (Google Cloud: load data into partitioned tables):

bq load \
  --replace=true \
  --source_format=PARQUET \
  'raw.trips$20240115' \
  "gs://example-landing/trips/dt=2024-01-15/*.parquet"

The $20240115 suffix is a partition decorator. With --replace=true the job overwrites that one partition and leaves every other day alone. Because load jobs are atomic, the partition holds either the old day or the new one, never a mixture (Google Cloud: load CSV data from Cloud Storage).

To backfill many days in one job, load the whole prefix and let BigQuery read the key from the paths. The key arrives as an extra column named dt:

bq load \
  --source_format=PARQUET \
  --hive_partitioning_mode=AUTO \
  --hive_partitioning_source_uri_prefix="gs://example-landing/trips" \
  raw.trips_backfill \
  "gs://example-landing/trips/*"

Decision 4: a guard on what is inside the file

The decorator has a useful strictness. For a table partitioned on a column, the partition named in the decorator must match the data being written, and the job fails otherwise. A file dropped in the wrong folder is refused, not silently filed under the wrong day.

That strictness also means a real file can surprise you, as the test below shows.

The design against a real file

The script models the layer locally on DuckDB, using the New York yellow taxi trip file for January 2024 (NYC TLC: trip record data). It is not BigQuery. Each load is plain SQL written to copy the documented behaviour: append, replace one partition, and reject mismatched rows.

# A local model of an ingestion layer, on DuckDB with real trip data. It is not BigQuery: the
# "table" is a DuckDB table, and each "load" is plain SQL that copies the documented behaviour.
import duckdb

SOURCE = "../../dedata/yellow_tripdata_2024-01.parquet"       # NYC yellow taxi trips, January 2024
con = duckdb.connect()
con.execute(f"CREATE VIEW landing AS SELECT CAST(tpep_pickup_datetime AS DATE) AS dt, * "
            f"FROM read_parquet('{SOURCE}')")

# 1. What is really inside the file named for January?
print("rows in the January file, by the month of the pickup time")
for month, n in con.execute("SELECT strftime(dt, '%Y-%m'), COUNT(*) FROM landing GROUP BY 1 ORDER BY 1").fetchall():
    print(f"  {month}  {n:>9,}")

# 2. One day loaded twice: by appending, and by replacing that day's partition.
con.execute("CREATE TABLE trips AS SELECT * FROM landing WHERE false")
DAY = "2024-01-15"


def rows(day):
    return con.execute("SELECT COUNT(*) FROM trips WHERE dt = CAST(? AS DATE)", [day]).fetchone()[0]


def append(day):
    con.execute("INSERT INTO trips SELECT * FROM landing WHERE dt = CAST(? AS DATE)", [day])


def replace_partition(day):
    con.execute("BEGIN")
    con.execute("DELETE FROM trips WHERE dt = CAST(? AS DATE)", [day])
    con.execute("INSERT INTO trips SELECT * FROM landing WHERE dt = CAST(? AS DATE)", [day])
    con.execute("COMMIT")


append(DAY)
first = rows(DAY)
append(DAY)
print(f"\nappend, run twice:             {first:>7,} -> {rows(DAY):>7,} rows for {DAY}")
replace_partition(DAY)
first = rows(DAY)
replace_partition(DAY)
print(f"replace partition, run twice:  {first:>7,} -> {rows(DAY):>7,} rows for {DAY}")


# 3. A guard: refuse a file whose rows do not all belong to the partition it is loaded into.
def guard(partition_month):
    outside = con.execute("SELECT COUNT(*) FROM landing WHERE strftime(dt, '%Y-%m') <> ?",
                          [partition_month]).fetchone()[0]
    return "accepted" if outside == 0 else f"rejected, {outside} rows fall outside {partition_month}"


print(f"\nwhole file into the 2024-01 partition: {guard('2024-01')}")

rows in the January file, by the month of the pickup time
  2002-12          2
  2009-01          3
  2023-12         10
  2024-01  2,964,606
  2024-02          3

append, run twice:              77,033 -> 154,066 rows for 2024-01-15
replace partition, run twice:   77,033 ->  77,033 rows for 2024-01-15

whole file into the 2024-01 partition: rejected, 18 rows fall outside 2024-01

Three findings, all from real data:

• The January file is not purely January. Of 2,964,624 rows, 18 have a pickup time in another month, two of them in 2002. Source systems produce such rows through clock faults and late corrections. A layer that assumes the file name is the truth will meet them eventually.
• Appending twice doubles the day. 77,033 trips become 154,066, and no error is raised.
• Replacing the partition twice changes nothing. The second run is harmless, which is the property every scheduled load needs.

The guard rejects the whole file for the January partition because of those 18 rows. So decide in advance which of three policies you want:

1. Reject and alert. Right when the source should never send such rows.
2. Route. Split rows by their own date and load each day's partition. Late rows then land where they belong.
3. Partition by arrival, not by event. An ingestion-time partitioned table files rows under the load date whatever they contain, and the event time stays an ordinary column.

There is no universally correct choice. The mistake is not choosing.

Decision 5: evidence for every load

When a number looks wrong, the first question is whether every file arrived. Answer it from the files themselves. An external table over the landing prefix exposes a pseudocolumn naming the file each row came from (Google Cloud: query Cloud Storage data):

CREATE OR REPLACE EXTERNAL TABLE landing.trips_files
OPTIONS (
  format = 'PARQUET',
  uris = ['gs://example-landing/trips/*']
);

SELECT _FILE_NAME AS source_file, COUNT(*) AS row_count
FROM landing.trips_files
GROUP BY source_file;

The statement follows the documented form (Google Cloud: Cloud Storage external tables). Store that result with the load date, and compare its total with the row count of the loaded partition. A load that passes this reconciliation is finished. One that does not has told you exactly which file to look at.

Scheduling the loads

Three options, in order of effort:

• Data Transfer Service. It schedules recurring loads from Cloud Storage with no code. The destination table must already exist. In its default append mode an unmodified file is loaded only once, and the minimum interval is 15 minutes, with event-driven transfers for shorter delays (Google Cloud: Cloud Storage transfers).
• An orchestrator. A daily workflow that runs the partition load for the run date gives you retries, backfills and dependencies. It is the usual choice once several tables depend on each other.
• An event trigger. A function that starts a load when an object arrives suits irregular deliveries. It must tolerate the same event arriving twice, which the partition replacement above already does.

Whichever you choose, run the loader as its own service account with access to this bucket and this dataset only. Service accounts and workload identity explains how.

When the schema changes

Sources add columns. Decide the policy before it happens:

• Strict: the load fails on an unknown column and a person decides. This is the safer default.
• Additive: a load that appends can be allowed to add new columns, and earlier rows then hold NULL in them (Google Cloud: modifying table schemas).

Removed or retyped columns should always stop the load. No setting makes those safe.

Pitfalls

• Appending daily loads. Every retry becomes a duplicate.
• Auto-detecting the schema of CSV files in production. The inferred types can differ between deliveries.
• Deleting landing files after loading. Keep them for a defined period with a lifecycle rule, so a partition can be rebuilt. See Google Cloud Storage.
• Trusting the folder name. Check what the rows say.
• Transforming during the load. Raw first, transformation after, so that a logic change does not require the source to resend.

For the choice between batch and streaming, see loading data into BigQuery. For choosing the partition column, see partitioning and clustering.

Building ingestion layers of this kind is part of the Data Engineering course.

Sources

• Load externally partitioned data (Google Cloud Documentation) — Google Cloud
• Create partitioned tables (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
• Query Cloud Storage data in external tables (Google Cloud Documentation) — Google Cloud
• Create Cloud Storage external tables (Google Cloud Documentation) — Google Cloud
• Load Cloud Storage data into BigQuery (Google Cloud Documentation) — Google Cloud
• Modifying table schemas (Google Cloud Documentation) — Google Cloud
• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission