Event Driven Postgres to Databricks Pipeline with Data Quality Checks

Build an event driven lakehouse ingestion pipeline without writing a scheduler or a single polling script. This Kestra blueprint watches a Postgres staging table, and as soon as new rows appear it fans the data out to Amazon S3 and Databricks in parallel, kicks off a Databricks aggregation job, queries the aggregated result over Databricks SQL, and runs a pandas based data quality gate that flags bad records before anyone downstream consumes them.

How it works

1. The pg_trigger trigger (io.kestra.plugin.jdbc.postgresql.Trigger) polls the staging_data table every two minutes and starts an execution whenever the query returns rows, storing the result set as the trigger output.
2. The json task (io.kestra.plugin.serdes.json.IonToJson) converts the fetched rows from Ion to JSON.
3. The parallel task (io.kestra.plugin.core.flow.Parallel) lands the file twice at once: s3_store (io.kestra.plugin.aws.s3.Upload) writes a timestamped object to S3 for the raw archive, while databricks_store (io.kestra.plugin.databricks.dbfs.Upload) pushes the same file to DBFS.
4. The create_job task (io.kestra.plugin.databricks.job.CreateJob) launches your existing Databricks aggregation job. Databricks authentication is shared through pluginDefaults, so the host and token are declared once.
5. The query_data_set task (io.kestra.plugin.databricks.sql.Query) reads the aggregated table back through a SQL warehouse, and csv (io.kestra.plugin.serdes.csv.IonToCsv) converts it to CSV.
6. The check_data task (io.kestra.plugin.scripts.python.Commands) runs an inline pandas script that counts null rows and negative quantities, publishing num_errors as an output. The if task (io.kestra.plugin.core.flow.If) logs a warning when any issues are found.

What you get

• An end to end event driven pipeline from operational Postgres to the Databricks lakehouse.
• Parallel dual writes so the raw archive in S3 and the DBFS landing zone never drift apart.
• A data quality gate that runs after aggregation, not before, so you validate what consumers actually read.
• Shared Databricks credentials via pluginDefaults instead of repeating them on every task.
• Full lineage in the Kestra UI from the triggering rows to the DQ verdict.

Who it's for

• Data engineers replacing cron plus custom polling scripts with declarative event driven ingestion.
• Lakehouse teams that need operational data in Databricks minutes after it lands in Postgres.
• Analytics engineers who want automated quality checks stitched into the load path.

Why orchestrate this with Kestra

Wiring this by hand means a poller, an S3 uploader, a Databricks API client, and a QA script, each with its own retries, secrets, and logging. Kestra collapses all of it into one YAML flow: the JDBC trigger handles change detection, Parallel handles the fan out, plugin defaults centralize authentication, and every step's inputs, outputs, and logs are captured automatically. If any task fails you see exactly where, replay from that point, and keep the audit trail.

Prerequisites

• A Postgres database with a staging_data table (adapt the trigger SQL to your change detection strategy).
• An S3 bucket (replace my-data-lake-bucket) and a Databricks workspace with a job and SQL warehouse. Replace the placeholder jobId and httpPath with your real values.
• The Databricks and AWS plugins available on your Kestra instance.

Secrets

• JDBC_CONNECTION_STRING, PG_USERNAME, PG_PASSWORD: Postgres connection used by the trigger.
• AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY: credentials for the S3 upload.
• DATABRICKS_HOST, DATABRICKS_TOKEN: workspace URL and personal access token for all Databricks tasks.

Quick start

1. Add the secrets above to your Kestra namespace.
2. Replace the S3 bucket, Databricks jobId, and SQL warehouse httpPath with your own values.
3. Adapt the trigger SQL and the aggregation query to your schema.
4. Deploy the flow, insert rows into staging_data, and watch the execution run end to end.

How to extend

• Replace the inline pandas check with Great Expectations or Soda for declarative data quality suites.
• Send the warning to Slack with io.kestra.plugin.slack.notifications.SlackIncomingWebhook instead of only logging it.
• Add a cleanup task that deletes processed rows from staging_data so the trigger only sees new data.
• Swap the polling trigger for io.kestra.plugin.debezium.postgres.RealtimeTrigger for true change data capture.
• Fail the execution instead of warning by raising an error when num_errors exceeds a threshold.

Links

• Databricks plugin docs: https://kestra.io/plugins/plugin-databricks
• Postgres plugin docs: https://kestra.io/plugins/plugin-jdbc-postgres
• AWS S3 plugin docs: https://kestra.io/plugins/plugin-aws