Event-driven data ingestion to AWS S3 data lake managed by Apache

Build an event-driven data lake ingestion pipeline on AWS that reacts the moment new files land in Amazon S3. Instead of polling on a fixed schedule or running batch jobs that miss fresh data, this blueprint watches an S3 prefix, ingests each new object into an Apache Iceberg table with a Python script, and keeps the table consistent with Amazon Athena MERGE and OPTIMIZE statements. It solves the classic data lake problem of late, duplicated, and fragmented data by combining real-time S3 event detection with declarative, retryable orchestration.

How it works

1. The io.kestra.plugin.aws.s3.Trigger watches the source_prefix of the S3 bucket on a PT1S interval, picks up new files (filter: FILES), and uses action: MOVE to relocate each detected object to the destination_prefix, guaranteeing exactly-once processing.
2. Inside an io.kestra.plugin.core.flow.WorkingDirectory, the io.kestra.plugin.git.Clone task pulls the public kestra-io/scripts repository so the ingestion logic stays version controlled.
3. The io.kestra.plugin.scripts.python.Commands task runs etl/aws_iceberg_fruit.py on an io.kestra.plugin.scripts.runner.docker.Docker runner with awswrangler and kestra dependencies, passing the triggered object key (trigger.objects | jq('.[].key') | first) as the ingestion argument.
4. io.kestra.plugin.aws.athena.Query runs a MERGE INTO against the Iceberg table to upsert rows, then a second Athena query runs OPTIMIZE ... REWRITE DATA USING BIN_PACK to compact files and keep query performance high.

What you get

• Near real-time ingestion triggered by S3 object events, not cron polling.
• Idempotent processing via the trigger MOVE action so files are handled once.
• ACID upserts into Apache Iceberg through Athena MERGE.
• Automatic table maintenance with Athena OPTIMIZE.
• Version-controlled ingestion logic cloned from Git at runtime.

Who it's for

• Data engineers building lakehouse ingestion on AWS.
• Platform teams replacing brittle Lambda and cron glue with declarative pipelines.
• Analytics teams who need fresh, deduplicated Iceberg tables queryable in Athena.

Why orchestrate this with Kestra

S3 and Athena have no native scheduler that can chain event detection, containerized Python, an upsert, and table optimization into one observable run. Kestra fills that gap: the S3 event trigger starts the flow declaratively, each task gets retries and full execution logs, lineage across ingest, merge, and optimize is captured end to end, and the entire pipeline is defined in version-controllable YAML rather than scattered scripts.

Prerequisites

• An AWS account with an S3 bucket, an Apache Iceberg table, and an Athena workgroup.
• Docker available for the script task runner.

Secrets

• AWS_ACCESS_KEY_ID
• AWS_SECRET_ACCESS_KEY
• AWS_DEFAULT_REGION

Quick start

1. Add the three AWS secrets above to your Kestra instance.
2. Set the bucket, database, source_prefix, and destination_prefix variables to match your environment.
3. Create the target Iceberg table and the raw_fruits staging table in Athena.
4. Enable the flow and drop a file into the source prefix to trigger ingestion.

How to extend

• Swap aws_iceberg_fruit.py for your own ingestion script in the cloned repo.
• Adjust the trigger prefix and interval to match your file landing pattern.
• Replace the sample MERGE with your real schema and add partitioning to the OPTIMIZE step.
• Add a notification task on failure or chain a downstream dbt or analytics flow.

Links

• AWS plugin docs
• Orchestrate AWS with Kestra