Curate GCS Files into BigQuery with Python

This blueprint implements the classic download, transform, load pattern with Google Cloud Storage as the source and BigQuery as the destination. Raw files land in a GCS bucket, but they rarely arrive analytics ready: column names are inconsistent, rows are empty, and duplicates creep in. This flow closes that gap on a schedule by pulling a raw object from GCS, cleaning it in a Python container with pandas, and loading the curated output into a clean BigQuery table, with a Discord notification to confirm each run. It is a reusable GCS to BigQuery ETL pipeline you can point at any CSV prefix.

How it works

1. The daily io.kestra.plugin.core.trigger.Schedule trigger fires on a cron cadence (0 6 * * *, shipped disabled) so the curation runs unattended.
2. The download task (io.kestra.plugin.gcp.gcs.Download) pulls the object at gs://<bucket>/{{ inputs.source_object }} into Kestra internal storage.
3. The transform task (io.kestra.plugin.scripts.python.Script) runs in a Docker container (ghcr.io/kestra-io/pydata:latest), reads the file with pandas, normalizes column names, drops empty rows, de-duplicates, and emits curated.csv.
4. The load_curated task (io.kestra.plugin.gcp.bigquery.Load) loads the curated CSV into analytics.customers_clean with autodetect, createDisposition: CREATE_IF_NEEDED, and writeDisposition: WRITE_TRUNCATE.
5. The notify task (io.kestra.plugin.discord.DiscordIncomingWebhook) confirms the load, and an alert_on_failure error handler posts to Discord if any step fails.

What you get

• A scheduled GCS to BigQuery pipeline that always reflects the latest curated state.
• Data cleaning in Python where it belongs, with orchestration declared in YAML.
• Built in success and failure notifications via Discord.
• A clean BigQuery table created automatically on the first run.

Who it's for

• Data engineers building ingestion pipelines on Google Cloud.
• Analytics engineers who need tidy, deduplicated tables to model on top of.
• Teams replacing brittle cron scripts with a declarative, observable flow.

Why orchestrate this with Kestra

GCS and BigQuery each have their own scheduling primitives, but neither coordinates a download, a Python transform, and a load as one observable unit. Kestra ties them together with event or schedule triggers, automatic retries, execution level lineage across tasks, and declarative YAML you can version control. The container based transform keeps business logic in Python while the pipeline stays reproducible, and the error handler guarantees you hear about failures, something a storage lifecycle rule or a scheduled query alone cannot deliver.

Prerequisites

A Google Cloud project with a GCS bucket and a BigQuery analytics dataset, plus a Discord incoming webhook. A raw object must exist at the source_object path (default raw/customers.csv).

Secrets

• GCP_PROJECT_ID: GCP project hosting the bucket and dataset.
• GCP_SERVICE_ACCOUNT: service account key JSON with GCS read and BigQuery job access.
• GCS_BUCKET: bucket holding the raw source files.
• DISCORD_WEBHOOK_URL: Discord incoming webhook URL.

Quick start

1. Add the secrets above to your Kestra namespace.
2. Point source_object and GCS_BUCKET at your file and confirm the analytics dataset exists.
3. Run the flow once on demand to validate the download, transform, and load.
4. Enable the daily schedule (or change the cron) to run it unattended.

How to extend

• Swap Download for Downloads to bulk-pull a whole prefix, then loop the transform over each file.
• Replace pandas with Polars or DuckDB inside the same Python script for heavier reshaping.
• Change writeDisposition to WRITE_APPEND for incremental accumulation instead of full refresh.
• Add a data quality check between transform and load, and fan out notifications to Slack or email.

Links

• Google Cloud plugin docs
• Orchestrate Google Cloud Storage with Kestra