Detect New Files in S3 and process them in Python

Turn an Amazon S3 bucket into an event source for your data pipelines. This blueprint watches an S3 prefix and, the moment a new object arrives, runs a Python script against it inside a containerized environment, then archives the processed file. It removes the need for cron-based polling jobs, manual downloads, or glue code to wire S3 events to your processing logic, giving you an event-driven file ingestion pipeline that scales from a single CSV to a steady stream of inbound data.

How it works

1. The io.kestra.plugin.aws.s3.Trigger trigger polls the bucket defined in vars.bucket (in vars.region) every second (interval: PT1S), looking only at files (filter: FILES) under the prefix: inbound/.
2. When a new object is detected, the trigger fires an execution and processes one object at a time (maxKeys: 1).
3. After detection the trigger moves the object (action: MOVE) to the archive/ key in the same bucket, so each file is processed exactly once and the inbound prefix stays clean.
4. The process_data task of type io.kestra.plugin.scripts.python.Commands runs on a Docker taskRunner, reads the triggered object into an inputFile named input.csv via {{ read(trigger.objects[0].uri) }}, installs pandas, and executes python process_data.py from your namespace files.
5. The script emits data.csv as an outputFile, captured in Kestra internal storage for downstream tasks or inspection.

What you get

• Event-driven ingestion: pipelines start on file arrival, not on a fixed schedule.
• Exactly-once handling through the move-to-archive pattern.
• Reproducible Python execution in Docker with pinned dependencies.
• Versioned, reusable logic via namespaceFiles (your process_data.py).
• Output files persisted in internal storage and ready to pass downstream.

Who it's for

• Data engineers building S3 landing-zone ingestion pipelines.
• Analysts who drop CSVs into a bucket and want automatic processing.
• Platform teams replacing ad hoc Lambda plus cron glue with declarative orchestration.

Why orchestrate this with Kestra

S3 has no native scheduler that can run and observe arbitrary Python with retries, logging, and lineage. Kestra closes that gap: the event trigger reacts to file arrival, executions are fully traced, tasks can retry on failure, and the entire pipeline is declarative YAML you can version and review. You get observability over every file, every run, and every output, instead of opaque, fire-and-forget functions.

Prerequisites

• An AWS account with an S3 bucket and read/write access to the inbound/ and archive/ prefixes.
• A Docker-enabled Kestra worker for the Python task runner.
• A process_data.py script uploaded as a namespace file.

Secrets

• AWS_ACCESS_KEY_ID: the AWS access key id used by the S3 trigger.
• AWS_SECRET_KEY_ID: the AWS secret key used by the S3 trigger.

Quick start

1. Set the AWS_ACCESS_KEY_ID and AWS_SECRET_KEY_ID secrets in your Kestra instance.
2. Update vars.bucket and vars.region to match your S3 bucket.
3. Add a process_data.py namespace file that reads input.csv and writes data.csv.
4. Save the flow and drop a file into the inbound/ prefix to trigger an execution.

How to extend

• Change prefix or filter to watch different folders or object types.
• Swap the Python logic for any transformation, validation, or enrichment.
• Add downstream tasks to load data.csv into a warehouse such as Snowflake or BigQuery.
• Add notifications on failure, or fan out with multiple maxKeys and a loop.

Links

• AWS plugin documentation
• Orchestrate AWS with Kestra