Building a CSV to Athena ETL pipeline on AWS

S3, Lambda, SNS, SQS, Postgres, Glue, Iceberg and Athena. One Terraform file, and every step safe to rerun.

Fleet trips arrive as CSV files. The API needs them in Postgres within seconds. Analysis needs the full history, queried without touching the API’s database. The pipeline does both: an event-driven path for every upload, and a nightly batch into an Iceberg table that Athena queries.

The upload path runs in seconds. The nightly path merges each day’s rows into the lake.

1. Upload straight to S3

The API never handles the file. It records the upload in Postgres and returns a presigned URL for one key, valid for 15 minutes.

upload_id = f"upl-{secrets.token_hex(4)}"
key = f"uploads/raw/{upload_id}/{filename}"
url = s3.generate_presigned_url(https://medium.com/%3Cbr/%3E%20%20%20%20"put_object",%3Cbr/%3E%20%20%20%20Params=%7B"Bucket":%20bucket,%20"Key":%20key%7D,%3Cbr/%3E%20%20%20%20ExpiresIn=900,%3Cbr/%3E)

The browser PUTs the file to that URL. A CORS rule on the bucket allows the PUT; the signature is what authorises it.

2. Validate every row in Lambda

An S3 event notification starts the Lambda for each new .csv under uploads/raw/.

resource "aws_s3_bucket_notification" "uploads" {
  bucket = aws_s3_bucket.uploads.id

  lambda_function {
    lambda_function_arn = aws_lambda_function.check_upload.arn
    events              = ["s3:ObjectCreated:*"]
    filter_prefix       = "uploads/raw/"
    filter_suffix       = ".csv"
  }
}

The Lambda streams the file and validates each row with the same Pydantic model the API uses, so a row gets the same verdict however it arrives.

reader = csv.DictReader(lines)
for row in reader:
    try:
        item = TripIngestItem.model_validate(row)
    except ValidationError as error:
        line = reader.line_num
        rejected.append({"line": line, "errors": error.errors()})
    else:
        passed.append(item.model_dump(mode="json"))

Good rows go to processed/<id>.jsonl, bad rows with their line numbers and reasons to rejected/<id>.jsonl. Decimals are written as strings, so 42.50 stays exact. A file is checked in about a second.

Pydantic has compiled parts, so the package is built for Lambda’s platform, Linux on ARM. Terraform runs the build, zips the folder, and redeploys the function only when the zip’s hash changes.

uv pip install --target build/ --only-binary :all: \
  --python-platform aarch64-manylinux2014 \
  --python-version 3.14 "pydantic==2.14.1"

3. One message, two consumers

The Lambda publishes one summary to SNS. With MessageStructure="json", email subscribers get readable text and every other subscriber gets JSON.

message = {"default": json.dumps(summary), "email": email_text}
sns.publish(
    TopicArn=topic_arn,
    Subject=f"{file} processed: {rows} rows, {bad} bad",
    MessageStructure="json",
    Message=json.dumps(message),
)

An SQS queue subscribed to the topic receives the JSON. After five failed attempts a message moves to a dead-letter queue.

4. Load into Postgres with an SQS worker

The worker runs in a container next to the API on EC2 and long-polls the queue, so nothing has to connect into the server. For each message it maps plates and driver codes to ids, then inserts:

insert(Trip).values(trips).on_conflict_do_nothing(
    index_elements=["trip_ref"]
).returning(Trip.trip_ref)

The unique trip_ref makes a repeated upload a no-op, and RETURNING separates new rows from duplicates. Rows with an unknown plate or driver are turned away with a reason. The outcome is stored on the upload's row in Postgres, where the upload page reads it.

Get surajit002’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Finally the worker writes every row Postgres now holds to uploads/loaded/<date>/ and deletes the message. If it crashes before the delete, the message comes back and the rerun changes nothing.

5. Merge nightly into Iceberg with Glue

EventBridge Scheduler starts a Step Functions state machine at 02:00. The first state works out yesterday’s date with JSONata. The second runs the Glue job and waits for it to finish.

"MergeTrips": {
  "Type": "Task",
  "Resource": "arn:aws:states:::glue:startJobRun.sync",
  "Arguments": {
    "JobName": "trips-merge",
    "Arguments": { "--date": "{% $states.input.date %}" }
  },
  "Retry": [{
    "ErrorEquals": ["States.ALL"],
    "IntervalSeconds": 60,
    "MaxAttempts": 2
  }]
}

The Glue job, Spark 3.5 on two G.1X workers, reads the day’s folder, casts each column to its type, drops duplicate trip_refs and merges:

MERGE INTO glue_catalog.fleet.trips AS t
USING new_rows AS n
ON t.trip_ref = n.trip_ref
WHEN NOT MATCHED THEN INSERT *

The table is Apache Iceberg, partitioned by day(started_at) and created by Terraform. Each MERGE writes new Parquet files, then commits a new snapshot to the Glue Data Catalog in one atomic step. Readers see the old version or the new one, never half of a write. A run takes about two minutes and costs a few cents.

6. Query with Athena

The analysis page runs one fixed query. User input never reaches the SQL text: the plate and the week’s bounds are sent as execution parameters.

SELECT
  count(*) AS trips,
  sum(distance_km) AS km,
  sum(litres_used) AS litres,
  round(sum(distance_km) / sum(litres_used), 2) AS km_per_litre
FROM fleet.trips
WHERE plate_number = ?
  AND started_at >= CAST(? AS timestamp)
  AND started_at < CAST(? AS timestamp)

athena.start_query_execution(
    QueryString=SQL,
    ExecutionParameters=[
        "'KAA 123A'",
        "'2026-10-18 21:00:00'",
        "'2026-10-25 21:00:00'",
    ],
    WorkGroup="crmapi",
)

The time filter prunes partitions, so a week’s query reads only that week’s files. The page polls GetQueryExecution and shows each state, the queue, planning and execution times, the bytes scanned and the cost: $5 per TB, rounded up to the MB, with a 10 MB minimum.

Every step is safe to rerun

• S3: each upload gets its own key.
• Lambda: the same file always produces the same output keys.
• Worker: ON CONFLICT DO NOTHING, the first result is kept, and the message is deleted last.
• Glue: MERGE … WHEN NOT MATCHED never inserts a trip twice.
• Step Functions: any day can be reloaded by starting it with {"date": "YYYY-MM-DD"}.

All of it is Terraform

The buckets, Lambda, SNS topic, queues, IAM roles, Glue database, Iceberg table and job, state machine, schedule and Athena workgroup are defined in one main.tf. Each role can do only its own step. One command, make deploy, applies it and ships the API and worker containers.

The stack

• Ingest: FastAPI, S3 presigned URLs, Lambda (Python 3.14, Pydantic)
• Messaging: SNS, SQS with a dead-letter queue
• Serving store: Postgres
• Lake: Glue 5.0 (Spark), Apache Iceberg, Glue Data Catalog
• Orchestration: EventBridge Scheduler, Step Functions
• Query: Athena
• Infrastructure: Terraform