Event-Driven Pipelines with Lambda and S3

A scheduled pipeline asks, every hour, "is there anything new?". An event-driven pipeline never asks. A file arrives, the storage service announces it, and processing starts within seconds. On AWS the simplest version is two services: Amazon S3 emits an event when an object is created, and an AWS Lambda function receives it.

The pattern is easy to set up and easy to get subtly wrong. The details that matter are in the documentation as single sentences, and each one corresponds to a bug that someone has shipped.

How the trigger works

You add a notification configuration to a bucket, naming the event types to publish and where to send them. S3 can deliver to an SNS topic, an SQS queue, a Lambda function or Amazon EventBridge (S3 docs: event notifications).

With Lambda as the destination, S3 invokes the function asynchronously with an event describing the object: the bucket name, the key, the size and an eTag (Lambda docs: S3 notifications). The event does not contain the file. The function must fetch it.

A configuration, written from the API and not run:

s3.put_bucket_notification_configuration(
    Bucket="demo-landing",
    NotificationConfiguration={"LambdaFunctionConfigurations": [{
        "LambdaFunctionArn": "arn:aws:lambda:us-east-1:123456789012:function:ingest-zones",
        "Events": ["s3:ObjectCreated:*"],
        "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": "incoming/"},
                                           {"Name": "suffix", "Value": ".csv"}]}},
    }]},
)

The filter limits the trigger to .csv objects under incoming/. Filters work on the key's prefix and suffix only; wildcards are not allowed.

Four sentences from the documentation

"Designed to be delivered at least once." The same event can arrive twice. A handler that appends rows will then append them twice.

"The object key name value is URL encoded." A file called red flower.jpg arrives in the event as red+flower.jpg. Pass that string straight to get_object and the file is not found.

"It could cause an execution loop." If the function writes to the bucket that triggers it, each output is a new object, which triggers the function again, without end.

"Lambda attempts to run it two more times." After an error, an asynchronous invocation is retried twice by default, and then the event is discarded unless a failure destination exists (Lambda docs: asynchronous error handling).

A handler that survives all four

The handler below receives an object-created event, reads a CSV file, validates it, removes placeholder rows, and writes JSON lines to a second bucket. The file is the New York taxi zone lookup table, uploaded under a key with spaces in it. The code runs against moto, a local emulator of S3, and the handler is called directly, twice, with the same event, to imitate a duplicate delivery.

import csv, io, json, os
from urllib.parse import unquote_plus
import boto3
from botocore.exceptions import ClientError
from moto import mock_aws

os.environ.update(AWS_ACCESS_KEY_ID="testing", AWS_SECRET_ACCESS_KEY="testing",
                  AWS_DEFAULT_REGION="us-east-1")
LANDING, LAKE = "demo-landing", "demo-lake"
REQUIRED = ["LocationID", "Borough", "Zone", "service_zone"]
s3 = None                                   # created once per execution environment


def handler(event, context):
    results = []
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = unquote_plus(record["s3"]["object"]["key"])          # keys arrive URL-encoded
        etag = record["s3"]["object"]["eTag"]
        out_key = f"curated/zones/{os.path.basename(key).rsplit('.', 1)[0]}.{etag[:8]}.jsonl"

        try:                                                       # already done? then do nothing
            s3.head_object(Bucket=LAKE, Key=out_key)
            results.append({"key": key, "status": "skipped, already processed"})
            continue
        except ClientError as e:
            if e.response["Error"]["Code"] != "404":
                raise

        body = s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode("utf-8")
        rows = list(csv.DictReader(io.StringIO(body)))
        if not rows or any(c not in rows[0] for c in REQUIRED):
            raise ValueError(f"{key}: missing required columns")
        good = [r for r in rows if r["Borough"] not in ("Unknown", "N/A")]
        s3.put_object(Bucket=LAKE, Key=out_key,
                      Body="\n".join(json.dumps(r) for r in good).encode("utf-8"))
        results.append({"key": key, "status": "processed", "rows_in": len(rows), "rows_out": len(good)})
    return results


@mock_aws
def main():
    global s3
    s3 = boto3.client("s3")
    s3.create_bucket(Bucket=LANDING)
    s3.create_bucket(Bucket=LAKE)

    key = "incoming/taxi zone lookup 2024.csv"                     # note the spaces
    s3.upload_file("taxi_zone_lookup.csv", LANDING, key)
    etag = s3.head_object(Bucket=LANDING, Key=key)["ETag"].strip('"')

    # The notification S3 would send for that upload (only the fields the handler reads)
    event = {"Records": [{"eventSource": "aws:s3", "eventName": "ObjectCreated:Put",
                          "s3": {"bucket": {"name": LANDING},
                                 "object": {"key": "incoming/taxi+zone+lookup+2024.csv", "eTag": etag}}}]}

    print("first delivery :", handler(event, None))
    print("second delivery:", handler(event, None))
    print("objects in lake:", [o["Key"] for o in s3.list_objects_v2(Bucket=LAKE)["Contents"]])


main()

first delivery : [{'key': 'incoming/taxi zone lookup 2024.csv', 'status': 'processed', 'rows_in': 265, 'rows_out': 263}]
second delivery: [{'key': 'incoming/taxi zone lookup 2024.csv', 'status': 'skipped, already processed'}]
objects in lake: ['curated/zones/taxi zone lookup 2024.c6064b7c.jsonl']

How it handles each hazard:

• Duplicates. The output key is derived from the source file name and its eTag, so the same input always maps to the same output. Before doing any work the handler checks whether that output exists. The second delivery is skipped, and the lake holds one object, not two. Even without the check, writing to the same key would overwrite, not duplicate.
• Encoded keys. unquote_plus turns taxi+zone+lookup+2024.csv back into the real key.
• Loops. Input and output are different buckets. The S3 guide's advice is exactly that: use two buckets, or restrict the trigger to a prefix used only for incoming objects.
• Bad files. A file without the required columns raises an error instead of producing a wrong output. That hands the decision to the retry and failure settings.

Of 265 rows, two placeholder rows were removed. The client is created outside the handler so that warm invocations reuse it.

Where failed events should go

Raising an error is only useful if someone finds out. For asynchronous invocations, configure one of:

• An on-failure destination, such as an SQS queue or SNS topic, which receives a record of the failed invocation.
• A dead-letter queue for the discarded event.

Then alarm on that queue having messages. A file that fails validation three times needs a person, and the queue is where they will find it. Retrying will not help a malformed file; the value of the retry is for transient faults such as a throttled request.

When to put a queue or a bus in between

Direct S3-to-Lambda is the right start. Three variations solve specific problems:

The last row is the most common in data engineering. Lambda's limits make it a poor place to transform a large file, and a good place to notice one, check its name and size, and start the job that will process it. AWS Step Functions covers the workflow side.

Operational habits

• One function, one job. Validation and conversion of one dataset, not a general file processor full of branches.
• Log the key, the eTag and the outcome for every record, so a missing file can be traced.
• Set the timeout and memory deliberately, with the largest expected file in mind.
• Limit concurrency if the function writes to something that cannot take parallel load.
• Handle late and out-of-order events. Events for the same key can arrive out of order; the sequencer field in the event exists to compare them.
• Test with real event shapes. Keep sample events in the repository and call the handler with them, as above.

Event-driven ingestion and scheduled batch are not rivals. Most platforms use events to land and validate data promptly, and schedules for the heavier modelling that follows. Idempotency in data pipelines explains the property that makes both safe to retry; AWS Lambda explained covers the function model; and building a serverless analytics stack on AWS shows where this trigger fits in a whole platform.

Amazon S3, AWS data services and pipeline design are covered in the Data Engineering course.

Sources

• Amazon S3 Event Notifications (Amazon S3 User Guide) — AWS Documentation
• Process Amazon S3 event notifications with Lambda (AWS Lambda Developer Guide) — AWS Documentation
• Event message structure (Amazon S3 User Guide) — AWS Documentation
• Configuring event notifications using object key name filtering (Amazon S3 User Guide) — AWS Documentation
• How Lambda handles errors and retries with asynchronous invocation (AWS Lambda Developer Guide) — AWS Documentation