Gate a Dataproc PySpark Batch on Data Readiness

Run a Dataproc Serverless PySpark batch only after the upstream data is proven ready, instead of firing a scheduled Spark job blindly against missing or stale source tables. This blueprint orchestrates a BigQuery readiness check, a conditional gate, and a Dataproc Serverless PySpark submission, then reports the outcome to Discord. It solves a common data engineering problem: scheduled Spark batches that waste serverless compute, produce empty rollups, or corrupt downstream tables because they ran before today's data landed.

How it works

1. The daily io.kestra.plugin.core.trigger.Schedule trigger fires the flow on a cron cadence (0 5 * * *, shipped disabled: true so you can review it first).
2. The check_readiness task (io.kestra.plugin.gcp.bigquery.Query) runs a FETCH_ONE query that counts today's rows in the analytics.source_events table for event_date = CURRENT_DATE().
3. The gate task (io.kestra.plugin.core.flow.If) evaluates {{ outputs.check_readiness.row.cnt > 0 }}.
4. When the condition is true, submit_pyspark (io.kestra.plugin.gcp.dataproc.batches.PySparkSubmit) launches the batch from mainPythonFileUri on Dataproc Serverless, and notify_submitted posts a confirmation via io.kestra.plugin.discord.DiscordIncomingWebhook.
5. When the condition is false, notify_skipped reports the skipped run to Discord.
6. An errors handler (alert_on_failure) alerts Discord if the readiness check or batch fails. A pluginDefaults block injects projectId and serviceAccount into every GCP task.

What you get

• A readiness gate that protects expensive serverless compute from running on empty or partial data.
• Conditional branching that submits the PySpark batch only when source rows exist.
• Discord notifications for submitted, skipped, and failed runs.
• Centralized GCP credentials through pluginDefaults so individual tasks stay clean.

Who it's for

• Data engineers running scheduled Spark or PySpark batches on GCP.
• Platform teams that need dependency-aware orchestration across BigQuery and Dataproc.
• Anyone tired of Spark jobs that succeed technically but produce wrong results on late data.

Why orchestrate this with Kestra

Dataproc Serverless can run a PySpark batch, but its own scheduler cannot express "only run if the upstream BigQuery data has actually landed." Kestra adds that gate declaratively in YAML: an event or schedule trigger, a readiness query, and an If condition that branches before any compute starts. You also get retries, failure alerting, full execution lineage across BigQuery and Dataproc in one place, and the freedom to swap the schedule trigger for an event trigger without rewriting the job.

Prerequisites

• A GCP project with Dataproc Serverless, BigQuery, and Cloud Storage enabled.
• A source_events table in the analytics dataset with an event_date column.
• A PySpark job uploaded to GCS (the flow points at jobs/daily_rollup.py).
• A Discord incoming webhook for notifications.

Secrets

• GCP_PROJECT_ID: GCP project hosting Dataproc, GCS, and BigQuery.
• GCP_SERVICE_ACCOUNT: service account key JSON with Dataproc, GCS, and BigQuery access.
• GCP_REGION: region for the serverless batch.
• GCS_BUCKET: bucket holding the PySpark job file.
• DISCORD_WEBHOOK_URL: Discord incoming webhook URL.

Quick start

1. Add the five secrets above to your Kestra namespace or tenant.
2. Adjust the readiness SQL and the If condition to match your freshness rule.
3. Point mainPythonFileUri at your PySpark job in GCS.
4. Run the flow manually to validate the gate, then enable the daily schedule.

How to extend

• Replace the row count with a freshness timestamp check or a GCS _SUCCESS marker check.
• Swap the schedule trigger for an event trigger that fires when a marker file lands in GCS.
• Add post-processing tasks after submit_pyspark to load batch results back into BigQuery.
• Route notifications to Slack or email instead of (or alongside) Discord.

Links

• Plugin docs: https://kestra.io/plugins/plugin-gcp
• Orchestrate Dataproc: https://kestra.io/orchestration/dataproc