Batch-Enrich BigQuery Rows with Vertex AI

Run scheduled LLM enrichment of BigQuery rows with a Gemini model on Vertex AI, without writing a one-off Python notebook or babysitting a loop. This Kestra blueprint pulls the rows that still need a label, fans them out to io.kestra.plugin.gcp.vertexai.ChatCompletion under a concurrency cap, writes the model response back to BigQuery, and emails the team on success or failure. It turns ad-hoc LLM batch jobs into a governed, retryable, observable pipeline that respects Vertex AI quotas and keeps costs predictable.

How it works

1. A io.kestra.plugin.core.trigger.Schedule trigger (nightly, cron 0 2 * * *, disabled by default) launches the batch on a cadence you control.
2. select_rows runs io.kestra.plugin.gcp.bigquery.Query with fetchType: FETCH to pull rows from analytics.reviews where sentiment IS NULL, limited to 50.
3. enrich is a io.kestra.plugin.core.flow.ForEach over outputs.select_rows.rows with concurrencyLimit: 4, capping parallel calls to Vertex AI.
4. Inside the loop, classify calls io.kestra.plugin.gcp.vertexai.ChatCompletion with region: us-central1 and modelId: gemini-1.5-flash, asking Gemini for a one-word sentiment.
5. store runs another io.kestra.plugin.gcp.bigquery.Query that updates the row with outputs.classify.predictions[0].
6. notify sends an io.kestra.plugin.email.MailSend summary with the batch size and execution id.
7. An errors handler (alert_on_failure) emails the team if any step fails.
8. pluginDefaults injects projectId and serviceAccount into every io.kestra.plugin.gcp task from secrets.

What you get

• Scheduled, concurrency-capped LLM enrichment over BigQuery.
• Per-row write-back, so the next run only processes unlabeled rows.
• Success and failure emails for every batch.
• Centralized GCP credentials via pluginDefaults.
• Full execution history, logs, and retries in the Kestra UI.

Who it's for

• Data and analytics engineers running LLM enrichment over warehouse tables.
• ML and AI teams operationalizing Gemini on Vertex AI for classification, tagging, or extraction.
• Platform teams that need governed, auditable batch LLM jobs instead of one-off notebooks.

Why orchestrate this with Kestra

Vertex AI gives you the model, BigQuery gives you the table, but neither schedules a row-by-row enrichment loop with concurrency control, retries, write-back, and alerting. Kestra adds the missing layer: declarative YAML, event or cron triggers, concurrencyLimit to stay under model quotas, automatic retries on transient errors, lineage across BigQuery and Vertex AI, and a dedicated error handler. You get one place to see what ran, what it cost, and what failed, instead of stitching together Cloud Scheduler, Cloud Functions, and ad-hoc Python.

Prerequisites

• A GCP project with BigQuery and the Vertex AI API enabled.
• A reviews table (analytics.reviews) with id, review, and a nullable sentiment column.
• A Gemini model available in your chosen region (default us-central1, gemini-1.5-flash).
• SMTP credentials for the notification emails.

Secrets

• GCP_PROJECT_ID: GCP project hosting BigQuery and Vertex AI.
• GCP_SERVICE_ACCOUNT: service account key JSON with BigQuery and Vertex AI access.
• SMTP_HOST: SMTP server hostname.
• SMTP_USERNAME: SMTP user.
• SMTP_PASSWORD: SMTP password.

Quick start

1. Add the secrets above to your Kestra namespace.
2. Point the select_rows SQL at your own table and filter, and update the store UPDATE statement to match.
3. Adjust region and modelId in classify to the Gemini model you have enabled.
4. Update from, to, and the email subjects in notify and alert_on_failure.
5. Enable the nightly trigger (set disabled: false) or run the flow manually from the UI.

How to extend

• Swap the sentiment prompt for tagging, summarization, PII redaction, or structured JSON extraction.
• Raise concurrencyLimit for higher throughput, or lower it to stay safely under Vertex AI quotas.
• Add a Slack notification next to MailSend for richer alerting.
• Replace the Schedule trigger with a Pub/Sub or BigQuery event trigger to enrich rows as they arrive.
• Add an If on emitted token-count metrics to stop a batch when a cost threshold is crossed.
• Persist raw model responses to GCS for auditability before the write-back.

Links

• Plugin docs: https://kestra.io/plugins/plugin-gcp
• Orchestration hub: https://kestra.io/orchestration/vertex-ai
• Orchestration hub: https://kestra.io/orchestration/bigquery