S3 File Trigger to Postgres with dbt Transformations

Turn an S3 bucket into the front door of your warehouse. This Kestra blueprint polls a bucket prefix for new files, and whenever objects land it loads each one into Postgres with an efficient batch insert, then runs your dbt project to transform the raw rows into analytics ready models. Concurrency control keeps runs from stepping on each other, automatic retries absorb transient database hiccups, and a flow level error handler posts to Slack the moment anything breaks.

How it works

1. The s3 trigger (io.kestra.plugin.aws.s3.Trigger) checks the etl-demo- prefix of the bucket every minute. Detected objects are downloaded into Kestra's internal storage and then deleted from the bucket (action: DELETE) so files are processed exactly once.
2. Flow level concurrency with behavior: QUEUE and limit: 1 serializes executions, so overlapping trigger fires queue instead of racing on the same tables.
3. The each task (io.kestra.plugin.core.flow.ForEach) iterates over trigger.objects with concurrencyLimit: 2, processing two files at a time.
4. Per file, jsonToIon (io.kestra.plugin.serdes.json.JsonToIon) normalizes the JSON, then insert_raw_data (io.kestra.plugin.jdbc.postgresql.Batch) streams the rows into raw_events with a parameterized insert.
5. The dbt task (io.kestra.plugin.dbt.cli.Run) executes dbt run in the ghcr.io/kestra-io/dbt-postgres container, reading your dbt project from namespace files and injecting connection details from secrets into profiles.
6. A flow level retry policy retries any failed task up to three times, and the errors branch fires io.kestra.plugin.slack.notifications.SlackIncomingWebhook if the run still fails.

What you get

• Exactly once, file driven ingestion with no cron guesswork and no duplicate loads.
• Parallel per file processing with a hard cap, tuned by one property.
• Batch inserts into Postgres instead of row by row loops.
• dbt transformations chained directly after the load, in the same execution and lineage.
• Built in retries plus a Slack alert path for anything retries cannot fix.

Who it's for

• Data engineers ingesting partner or application exports that arrive as files on S3.
• Analytics teams who want dbt to run the moment fresh raw data lands, not on a timer.
• Platform teams standardizing file to warehouse pipelines with alerting included.

Why orchestrate this with Kestra

A homegrown version of this needs an S3 poller, a queue, a loader, a dbt scheduler, and alerting glue. Kestra provides all of it declaratively: the S3 trigger handles detection and cleanup, ForEach handles fan out with bounded parallelism, flow concurrency prevents overlapping runs, and retries and error handling are flow level policies rather than code. Every file, insert, and dbt log line is visible in one execution view.

Prerequisites

• An S3 bucket (replace my-landing-bucket) receiving JSON files under the configured prefix. Note that action: DELETE removes processed files, switch to MOVE if you need an archive.
• A Postgres database with the target raw table (adapt the insert statement to your schema).
• Your dbt project uploaded as namespace files in the flow's namespace.

Secrets

• AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY: credentials for the S3 trigger.
• JDBC_CONNECTION_STRING, PG_USERNAME, PG_PASSWORD, PG_HOST, PG_PORT: Postgres connection for the batch insert and the dbt profile.
• SLACK_WEBHOOK: incoming webhook URL for failure alerts.

Quick start

1. Add the secrets above to your Kestra namespace.
2. Replace the bucket, prefix, region, and insert statement with your own values.
3. Upload your dbt project to the namespace and align the profiles block with your dbt_project.yml name.
4. Deploy the flow and drop a JSON file matching the prefix into the bucket.
5. Watch the execution load the file, run dbt, and complete in the Kestra UI.

How to extend

• Add io.kestra.plugin.dbt.cli.Test after the run to gate downstream consumers on dbt tests.
• Swap action: DELETE for MOVE with a moveTo prefix to keep an archive of processed files.
• Raise concurrencyLimit for high file volumes, or remove flow concurrency if runs are independent.
• Route the error notification through Microsoft Teams, PagerDuty, or email using the corresponding notification plugin.
• Add a CSV or Avro branch by switching the serdes task per file extension with io.kestra.plugin.core.flow.Switch.

Links

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