Snowflake Stream Trigger to Postgres with dbt Transformations

React to changes in Snowflake instead of guessing when data arrives. This Kestra blueprint polls a Snowflake stream, and whenever new rows appear it starts an execution that loads each row batch into Postgres with efficient batch inserts and then runs your dbt project to transform the raw data downstream. Because Snowflake streams only return changed rows since the last consumption, the flow processes deltas rather than full table scans, and flow level concurrency, retries, and Slack alerting make the pipeline safe to run unattended.

How it works

1. The sf trigger (io.kestra.plugin.jdbc.snowflake.Trigger) connects with key pair authentication every five minutes and runs select * from streams.my_snowflake_stream on the COMPUTE_WH warehouse. When the stream returns rows, an execution starts with the rows stored as trigger output.
2. Flow level concurrency (behavior: QUEUE, limit: 1) makes sure two polls never process overlapping data at the same time.
3. The each task (io.kestra.plugin.core.flow.ForEach) iterates over trigger.rows with concurrencyLimit: 2.
4. Per batch, jsonToIon (io.kestra.plugin.serdes.json.JsonToIon) normalizes the payload and insert_raw_data (io.kestra.plugin.jdbc.postgresql.Batch) streams it into raw_events with a parameterized insert.
5. The dbt task (io.kestra.plugin.dbt.cli.Run) runs your dbt project from namespace files inside the ghcr.io/kestra-io/dbt-postgres container, with the Postgres profile rendered from secrets.
6. A constant retry policy retries failed tasks three times, and the errors branch posts to Slack through io.kestra.plugin.slack.notifications.SlackIncomingWebhook if the execution still fails.

What you get

• Delta based ingestion from Snowflake, driven by the stream rather than a blind schedule.
• Key pair authentication to Snowflake with every credential kept in the secret store.
• Bounded parallel loading into Postgres with batch inserts.
• dbt transformations chained in the same execution, sharing one lineage view.
• Queueing, retries, and Slack alerting configured as flow level policies.

Who it's for

• Data engineers syncing curated Snowflake data into operational Postgres databases.
• Teams doing reverse ETL who want change driven syncs instead of hourly full refreshes.
• Platform engineers standardizing warehouse to database pipelines with built in alerting.

Why orchestrate this with Kestra

Consuming a Snowflake stream correctly means tracking offsets, scheduling polls, handling partial failures, and alerting, all of which is bespoke code without an orchestrator. Kestra's JDBC trigger encapsulates the polling loop, ForEach gives bounded fan out, flow concurrency prevents overlapping consumption, and retries plus the errors branch turn failure handling into configuration. Every execution shows exactly which rows were picked up and what dbt did with them.

Prerequisites

• A Snowflake account with a stream (adapt streams.my_snowflake_stream) and a user configured for key pair authentication.
• A Postgres database with the target raw table matching the insert statement.
• Your dbt project uploaded as namespace files in the flow's namespace.

Secrets

• SNOWFLAKE_URL: JDBC connection URL for your Snowflake account.
• SNOWFLAKE_PRIVATE_KEY, SNOWFLAKE_PRIVATE_KEY_PASSWORD: key pair credentials for the trigger.
• JDBC_CONNECTION_STRING, PG_USERNAME, PG_PASSWORD, PG_HOST, PG_PORT: Postgres connection for the load and the dbt profile.
• SLACK_WEBHOOK: incoming webhook URL for failure alerts.

Quick start

1. Add the secrets above to your Kestra namespace.
2. Point the trigger SQL at your stream and set your warehouse name.
3. Adapt the insert statement to your Postgres schema and upload your dbt project to the namespace.
4. Deploy the flow, insert rows into the stream's source table, and watch the next poll pick them up.

How to extend

• Lower the interval for fresher syncs, or switch to a Snowflake task plus webhook for push semantics.
• Add io.kestra.plugin.dbt.cli.Test after the run to validate models before consumers read them.
• Route rows to different tables with io.kestra.plugin.core.flow.Switch based on a type column.
• Send the alert to PagerDuty or Opsgenie for on call escalation instead of a Slack channel.
• Track load metrics by appending a Postgres Query task that counts inserted rows per execution.

Links

• Snowflake plugin docs: https://kestra.io/plugins/plugin-jdbc-snowflake
• dbt plugin docs: https://kestra.io/plugins/plugin-dbt
• Postgres plugin docs: https://kestra.io/plugins/plugin-jdbc-postgres