GCS to BigQuery Ingestion with Data Quality Verification & Slack Alert

Ingesting dirty or unvalidated data into enterprise cloud data warehouses like Google BigQuery can break downstream analytics dashboards, invalidate financial reports, and result in costly backfills. Catching malformed records, null foreign keys, and anomalous records before landing into production warehouse tables is essential for robust data engineering.

This blueprint implements a production-grade data quality gate for cloud storage ingestion: it fetches inbound batch datasets, runs automated anomaly and data integrity checks, exports a quality metrics manifest, and dispatches real-time Slack notifications.

How it works

1. fetch_batch_data (io.kestra.plugin.core.http.Request): Downloads the inbound dataset batch from cloud storage (or simulated HTTP data lake endpoint).
2. analyze_data_quality (io.kestra.plugin.scripts.python.Script): Executes a lightweight containerized Python audit that evaluates records for missing required attributes, negative monetary values, and schema anomalies, writing quality_metrics.json.
3. evaluate_quality_gate (io.kestra.plugin.core.flow.If): Assesses the data quality audit result:
   • If the batch meets acceptance criteria, notify_slack_success (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) sends a confirmation card to the data operations channel.
4. alert_on_failure (errors block): Triggers immediately if any task in the ingestion flow fails, dispatching an urgent Slack incident alert with the execution ID.

What you get

• Automated data quality firewall preventing contaminated data from entering warehouse tables.
• Zero external dependencies: data profiling and quality calculation handled cleanly in Python.
• Real-time Slack notifications keeping data engineering teams informed of pipeline health.
• Auditable JSON metric outputs captured automatically in Kestra internal storage.

Who it's for

• Data Engineers building resilient ELT/ETL pipelines for Google BigQuery and cloud data warehouses.
• Analytics Engineers responsible for dataset reliability and metric accuracy.
• Data Platform teams needing centralized visibility and alerting for daily batch loads.

Why orchestrate this with Kestra

Coordinating cloud storage transfers, custom Python profiling scripts, conditional quality gates, and communication channels requires fragile cron jobs or monolithic pipeline tools. Kestra provides declarative YAML flow definitions, native Docker containerization, built-in conditional branching, secure secret injection, and comprehensive observability across every pipeline run.

Prerequisites

• Inbound dataset available via Google Cloud Storage or HTTP endpoint.
• Target BigQuery dataset and GCP service account credentials (for production warehouse insertion).
• Slack Incoming Webhook URL for data pipeline status alerts.

Secrets

• SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint for operational notifications.
• GCP_SERVICE_ACCOUNT_KEY: (Optional) JSON credentials for Google Cloud service account when writing to BigQuery.

Quick start

1. Add the SLACK_WEBHOOK_URL secret to your Kestra namespace secrets.
2. Import this blueprint into your Kestra instance.
3. Click Execute to trigger the flow and observe the data validation step and Slack alert.
4. Enable or adjust the daily_ingestion_schedule to match your batch arrival schedule.

How to extend

• Swap the download task with io.kestra.plugin.gcp.gcs.Download or io.kestra.plugin.gcp.gcs.Trigger to process Google Cloud Storage files directly upon creation.
• Append io.kestra.plugin.gcp.bigquery.Load to automatically load verified files into BigQuery tables after quality validation passes.
• Integrate Great Expectations or Soda Core inside the Python script for enterprise-scale quality assertions.

Links

• Kestra BigQuery Plugin Documentation
• Kestra Google Cloud Storage Plugin Documentation
• Kestra Slack Plugin Documentation
• This blueprint has been created by GauravPatil2515