Serverless Vaccination Data Pipeline

Build an S3-triggered Lambda pipeline that validates and loads health data into DynamoDB.

Introduction

30 Second Summary

Every vaccination campaign generates thousands of records per day across dozens of sites. Health systems need automated pipelines that validate, transform, and store this data the moment it arrives.

In this project, you will build a serverless data ingestion pipeline on AWS that automatically processes vaccination campaign CSV files when they land in cloud storage. You'll implement schema validation, dead-letter routing for bad records, and batch loading into a NoSQL database.

What You'll Build

An event-driven serverless pipeline deployed with AWS SAM that triggers on S3 file uploads, validates records against an expected schema, and loads them into DynamoDB.

By the end of this project, you'll have:

  • A SAM template defining S3, Lambda, and DynamoDB resources as infrastructure-as-code.
  • A Python Lambda function that validates, transforms, and batch-loads vaccination records into DynamoDB.
  • End-to-end proof of the pipeline using a synthetic dataset of ~41K Philippine vaccination events verified through CloudWatch logs and DynamoDB queries.
  • Secret Mission: Add a cold chain risk monitor that flags vaccine batches where potency has degraded below a safety threshold.

Are there any prerequisites?

You'll need an AWS account with the AWS CLI already configured, Python installed on your machine, and Git installed. Familiarity with basic AWS services like S3 and Lambda is helpful but not required.

Set Up Your Tools and Dataset

Your serverless pipeline needs data to ingest. In this project, you will process vaccination campaign records from the Philippines.

To get that data, you will clone a synthetic dataset generator that produces realistic vaccination event records across 17 Philippine regions. Once you have the generator and its dependencies installed locally, you will be ready to build and deploy the pipeline in later steps.

In this step, get ready to:
  • Clone the Philippine vaccination campaign dataset generator repository.
  • Install the Python dependencies the generator needs to run.
Clone the dataset generator repository

The dataset generator is an open-source repository that uses statistical models to produce synthetic vaccination event data. It outputs records that mirror the structure and volume of real campaign data, making it ideal for testing your ingestion pipeline.

  • In your terminal, move to your Desktop by running this command:
cd ~/Desktop
  • Clone the dataset generator repository by running:
git clone https://github.com/cs-loliva/philippine-vaccination-campaign-synthetic-dataset.git
  • Confirm the folder was created by running:
ls

You should see philippine-vaccination-campaign-synthetic-dataset listed in the output.

Clone failed or repository not found?

Make sure you have an internet connection. The repository is public, so no authentication is needed.

Check that git is installed by running git --version in your terminal. If it returns a version number, Git is working correctly.

Help me troubleshoot why git clone is failing

Install the generator's Python dependencies

The dataset generator relies on scientific computing libraries like NumPy, Pandas, and SciPy to produce statistically realistic data. You need to install these before you can generate the dataset.

  • Move into the cloned repository by running:
cd philippine-vaccination-campaign-synthetic-dataset
  • Install the Python dependencies by running:
pip install -r requirements.txt

What is this installing?

The requirements.txt file lists six libraries the generator depends on: NumPy, Pandas, SciPy, scikit-learn, statsmodels, and Faker. These power the statistical models that produce realistic vaccination event data.

These are local-only dependencies for generating the dataset. They are not used by your Lambda function.

The installation may take a minute or two. When it completes, you should see Successfully installed followed by the package names and versions.

Seeing permission errors or install failures?

If you see a permissions error, try running the command with --user at the end: pip install -r requirements.txt --user.

If a specific package fails to build, make sure your Python version is 3.11 or higher by running python --version.

Help me fix Python dependency installation errors

Your dataset generator is ready to go. Next up, you will define the infrastructure for your serverless pipeline using AWS SAM templates.

Define the Infrastructure

With all your tools installed and verified, it is time to define the cloud resources your pipeline needs. Instead of clicking through the AWS console, you will declare everything in a single YAML file that SAM deploys for you.

This approach is called infrastructure-as-code. Your template.yaml file becomes the blueprint for an S3 bucket, a DynamoDB table, and a Lambda function. Change the file, redeploy, and your cloud resources update automatically.

In this step, get ready to:
  • Create the SAM project directory structure with the required files.
  • Write the SAM template defining the S3 bucket and DynamoDB table.
  • Add the Lambda function with IAM policies and an S3 event trigger.
Create the SAM project directory structure

Your SAM project needs a specific layout. The root folder holds template.yaml (the infrastructure blueprint), and a src/ subfolder holds the Lambda function code and its dependencies.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your SAM project folder and move into it by running these commands:
mkdir vaccination-pipeline
cd vaccination-pipeline
  • Create the src/ subfolder and the required files by running these commands:
mkdir src
touch template.yaml
touch src/requirements.txt
  • Confirm the files exist by running:
ls
ls src/

You should see template.yaml in the first listing, and requirements.txt in the second.

Don't see the files?

Make sure you ran mkdir src before creating files inside it. If touch is not recognized on Windows, use New-Item instead (e.g. New-Item template.yaml).

help me create the SAM project directory structure on my operating system

  • Write the Lambda dependency file by opening src/requirements.txt in Cursor and pasting this content:
boto3>=1.35.0

Why pin boto3 in requirements.txt?

The Lambda runtime already includes boto3, but pinning it in requirements.txt ensures reproducible builds. When SAM packages your function, it installs exactly the version you specify rather than relying on whatever ships with the runtime.

  • Save src/requirements.txt.
Write the SAM template with S3 and DynamoDB resources

The SAM template declares your cloud resources using CloudFormation syntax with serverless extensions. You will start with the template header, global defaults, and the two data-layer resources: an S3 bucket for incoming files and a DynamoDB table for validated records.

  • Open template.yaml in Cursor and paste the following content:
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Serverless Vaccination Campaign Data Ingestion Pipeline

Globals:
  Function:
    Timeout: 60
    MemorySize: 256
    Runtime: python3.12
    Architectures:
      - x86_64

Resources:
  IngestionBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub vaccination-ingestion-${AWS::AccountId}

  VaccinationEventsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: VaccinationEvents
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: region_date
          AttributeType: S
        - AttributeName: event_id
          AttributeType: S
      KeySchema:
        - AttributeName: region_date
          KeyType: HASH
        - AttributeName: event_id
          KeyType: RANGE

What does this code do?

The Transform line tells CloudFormation this is a SAM template, which unlocks serverless-specific resource types like AWS::Serverless::Function.

The Globals section sets defaults for all Lambda functions in the stack: 60-second timeout, 256 MB memory, and the python3.12 runtime.

IngestionBucket creates an S3 bucket with a unique name derived from your AWS account ID. VaccinationEventsTable creates a DynamoDB table with a composite key: region_date as the partition key and event_id as the sort key.

Why PAY_PER_REQUEST billing?

PAY_PER_REQUEST means DynamoDB scales automatically and you only pay for actual reads and writes. For this project's data volume, it stays comfortably within the always-free tier (25 GB storage, 200M requests/month).

Do not save yet. You will add the Lambda function next.

Add the Lambda function with IAM policies

The Lambda function is the core of your pipeline. It needs permissions to read CSV files from S3, write dead-letter records back to S3, and batch-write valid records to DynamoDB.

  • In template.yaml, add this below the VaccinationEventsTable resource (directly after the RANGE line, maintaining the same 2-space indentation level as the other resources):
  IngestionFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: handler.lambda_handler
      Policies:
        - S3ReadPolicy:
            BucketName: !Sub vaccination-ingestion-${AWS::AccountId}
        - S3CrudPolicy:
            BucketName: !Sub vaccination-ingestion-${AWS::AccountId}
        - DynamoDBCrudPolicy:
            TableName: !Ref VaccinationEventsTable
      Environment:
        Variables:
          TABLE_NAME: !Ref VaccinationEventsTable
          DEAD_LETTER_PREFIX: dead-letter/
      Events:
        S3Upload:
          Type: S3
          Properties:
            Bucket: !Ref IngestionBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: raw/

Outputs:
  BucketName:
    Description: S3 bucket for vaccination data ingestion
    Value: !Ref IngestionBucket
  TableName:
    Description: DynamoDB table for vaccination events
    Value: !Ref VaccinationEventsTable

What do these IAM policies do?

SAM policy templates save you from writing raw IAM JSON. Each one grants the minimum permissions your function needs:

  • S3ReadPolicy allows the Lambda to read objects (the uploaded CSV files).
  • S3CrudPolicy allows writing dead-letter files back to the same bucket.
  • DynamoDBCrudPolicy allows batch-writing validated records to the table.

How does the S3 event trigger work?

The Events section creates an S3 event notification automatically. Any file uploaded to the raw/ prefix triggers this Lambda. Files written to other prefixes (like dead-letter/) do not trigger it, which prevents recursive invocations.

  • Save template.yaml.
  • Validate your template by running this command in the terminal from your vaccination-pipeline folder:
sam validate

You should see a message confirming the template is valid.

Seeing a validation error?

YAML is whitespace-sensitive. Make sure IngestionFunction is indented at the same level as IngestionBucket and VaccinationEventsTable (2 spaces under Resources:). Check that you do not have tabs mixed with spaces.

help me fix a SAM template validation error

✔️ Awesome, I've got everything!

Great. Double check you have saved template.yaml and that sam validate passes before moving on.

ⓧ I'd like to double check the full code

Here is the complete template.yaml file:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Serverless Vaccination Campaign Data Ingestion Pipeline

Globals:
  Function:
    Timeout: 60
    MemorySize: 256
    Runtime: python3.12
    Architectures:
      - x86_64

Resources:
  IngestionBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub vaccination-ingestion-${AWS::AccountId}

  VaccinationEventsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: VaccinationEvents
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: region_date
          AttributeType: S
        - AttributeName: event_id
          AttributeType: S
      KeySchema:
        - AttributeName: region_date
          KeyType: HASH
        - AttributeName: event_id
          KeyType: RANGE

  IngestionFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: handler.lambda_handler
      Policies:
        - S3ReadPolicy:
            BucketName: !Sub vaccination-ingestion-${AWS::AccountId}
        - S3CrudPolicy:
            BucketName: !Sub vaccination-ingestion-${AWS::AccountId}
        - DynamoDBCrudPolicy:
            TableName: !Ref VaccinationEventsTable
      Environment:
        Variables:
          TABLE_NAME: !Ref VaccinationEventsTable
          DEAD_LETTER_PREFIX: dead-letter/
      Events:
        S3Upload:
          Type: S3
          Properties:
            Bucket: !Ref IngestionBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: raw/

Outputs:
  BucketName:
    Description: S3 bucket for vaccination data ingestion
    Value: !Ref IngestionBucket
  TableName:
    Description: DynamoDB table for vaccination events
    Value: !Ref VaccinationEventsTable

And here is src/requirements.txt:

boto3>=1.35.0

Your infrastructure blueprint is complete. Next up, you will write the Python Lambda function that reads CSV files, validates each record, and loads the good data into DynamoDB.

Build the Ingestion Lambda

Your infrastructure template defines what resources exist. Now you need to write the code that actually runs when a CSV file lands in S3.

The Lambda function is the brain of this pipeline. It needs to read raw CSV data, decide which records are valid, transform the good ones into a queryable format, and route the bad ones somewhere safe for inspection.

In this step, get ready to:
  • Write the S3 event handler that reads uploaded CSV files.
  • Implement schema validation that catches invalid records.
  • Transform and load valid records into DynamoDB with dead-letter routing for failures.
Write the S3 event handler

When S3 triggers your Lambda, it sends an event payload containing the bucket name and object key of the uploaded file. Your handler needs to extract that information, fetch the file, and parse it as CSV.

  • Create a new file called handler.py inside the src/ folder of your SAM project.
  • Add the module setup (imports, clients, environment variables, and constants) by pasting this code:
import csv
import io
import json
import os
import urllib.parse
from datetime import datetime

import boto3

# Initialize AWS service clients
s3 = boto3.client("s3")
dynamodb = boto3.resource("dynamodb")

# Read environment variables set by SAM template
TABLE_NAME = os.environ["TABLE_NAME"]
DEAD_LETTER_PREFIX = os.environ["DEAD_LETTER_PREFIX"]

# Columns that every CSV row must contain
REQUIRED_COLUMNS = [
    "date",
    "region",
    "vaccine_type",
    "site_id",
    "person_id",
    "dose_sequence",
]

# The 17 valid Philippine regions for this dataset
VALID_REGIONS = [
    "NCR",
    "CAR",
    "Region I",
    "Region II",
    "Region III",
    "Region IV-A",
    "Region IV-B",
    "Region V",
    "Region VI",
    "Region VII",
    "Region VIII",
    "Region IX",
    "Region X",
    "Region XI",
    "Region XII",
    "Region XIII",
    "BARMM",
]

What does this code do?

This module setup establishes everything the handler needs before any function runs.

  • s3 = boto3.client("s3") creates an S3 client for reading uploaded files and writing dead-letter output.
  • dynamodb = boto3.resource("dynamodb") creates a higher-level DynamoDB resource that provides the convenient batch_writer() API.
  • TABLE_NAME and DEAD_LETTER_PREFIX are pulled from the environment variables you defined in template.yaml.
  • VALID_REGIONS lists all 17 Philippine administrative regions. Any record with a region not in this list is flagged as invalid.
  • Save src/handler.py.
  • You won't see any output yet. The next chunks will add the functions that use these constants.
Implement schema validation

Real-world data is messy. Dates might be in the wrong format, required fields might be blank, and regions might be misspelled. Your validation function needs to catch all of these problems and collect error messages explaining exactly what went wrong.

  • In src/handler.py, add the validate_record function below the VALID_REGIONS list by pasting this code:
def validate_record(record, row_index):
    errors = []

    # Check date field exists and matches YYYY-MM-DD format
    date_val = record.get("date", "").strip()
    if not date_val:
        errors.append("missing date")
    else:
        try:
            datetime.strptime(date_val, "%Y-%m-%d")
        except ValueError:
            errors.append(f"invalid date format: {date_val}")

    # Check region is one of the 17 valid Philippine regions
    region_val = record.get("region", "").strip()
    if not region_val:
        errors.append("missing region")
    elif region_val.title() not in [r.title() for r in VALID_REGIONS]:
        errors.append(f"invalid region: {region_val}")

    # Check dose_sequence is a positive integer
    dose_val = record.get("dose_sequence", "").strip()
    if not dose_val:
        errors.append("missing dose_sequence")
    else:
        try:
            dose_int = int(dose_val)
            if dose_int < 1:
                errors.append(f"dose_sequence must be positive: {dose_val}")
        except ValueError:
            errors.append(f"dose_sequence not an integer: {dose_val}")

    # Check remaining required fields are not empty
    for col in ["vaccine_type", "site_id", "person_id"]:
        if not record.get(col, "").strip():
            errors.append(f"missing {col}")

    return errors

What does this code do?

This function inspects a single CSV row and returns a list of everything wrong with it. An empty list means the record passed all checks.

  • Date validation uses datetime.strptime to confirm the format is exactly YYYY-MM-DD. A date like 13/01/2024 would be flagged.
  • Region validation normalizes both sides to title case before comparing, so "ncr" still matches "NCR".
  • Dose sequence must be a positive integer. A value of 0 or a non-numeric string like "first" would be caught.
  • Save src/handler.py.
  • You won't see a change yet. The next chunk will add the transformation and orchestration logic that uses this function.

Seeing a red underline in your editor?

If your editor shows a warning about VALID_REGIONS or datetime being undefined, make sure this function is placed below the constants and imports from the first chunk. The function references those names.

Why isn't my validate_record function recognized by my editor?

Transform valid records and load them into DynamoDB

Validated records need to be shaped into the format your DynamoDB table expects. Each record needs a composite partition key combining region and date, and a sort key combining person and dose. This key design lets you later query all vaccination events for a specific region on a specific day.

  • In src/handler.py, add the transform_record function below validate_record by pasting this code:
def transform_record(record):
    # Normalize region name to title case for consistency
    region = record["region"].strip().title()
    date_str = record["date"].strip()

    # Create composite keys for DynamoDB
    region_date = f"{region}#{date_str}"
    event_id = f"{record['person_id'].strip()}#{record['dose_sequence'].strip()}"

    return {
        "region_date": region_date,
        "event_id": event_id,
        "date": date_str,
        "region": region,
        "vaccine_type": record["vaccine_type"].strip(),
        "site_id": record["site_id"].strip(),
        "person_id": record["person_id"].strip(),
        "dose_sequence": int(record["dose_sequence"].strip()),
    }

What does this code do?

This function takes a validated CSV row and reshapes it into a DynamoDB item.

  • region_date combines the region and date with a # separator (e.g., Ncr#2024-03-15). This is the partition key that groups all events in the same region on the same day.
  • event_id combines person and dose to create a unique sort key within that partition.
  • All string fields are stripped of whitespace. The dose_sequence is cast to an integer for numeric queries.
  • Save src/handler.py.

Now you have the two helper functions. The final piece is the main handler that orchestrates everything: reading the file, running validation, batch-writing valid records, and routing invalid ones to dead-letter storage.

  • Add the lambda_handler function below transform_record by pasting this code:
def lambda_handler(event, context):
    # Extract bucket and key from the S3 event notification
    bucket = event["Records"][0]["s3"]["bucket"]["name"]
    key = urllib.parse.unquote_plus(
        event["Records"][0]["s3"]["object"]["key"], encoding="utf-8"
    )

    print(f"Processing file: s3://{bucket}/{key}")

    # Read the uploaded CSV file from S3
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response["Body"].read().decode("utf-8")

    reader = csv.DictReader(io.StringIO(content))

    # Fail fast if required columns are missing from the CSV header
    missing_cols = [col for col in REQUIRED_COLUMNS if col not in reader.fieldnames]
    if missing_cols:
        error_msg = f"Missing required columns: {missing_cols}"
        print(error_msg)
        raise ValueError(error_msg)

    valid_records = []
    invalid_records = []

    # Validate each row and sort into valid or invalid
    for idx, record in enumerate(reader):
        errors = validate_record(record, idx)
        if errors:
            invalid_records.append(
                {"row": idx + 2, "errors": errors, "data": dict(record)}
            )
        else:
            valid_records.append(transform_record(record))

    print(
        f"Validation complete: {len(valid_records)} valid, "
        f"{len(invalid_records)} invalid"
    )

    # Batch-write valid records to DynamoDB
    if valid_records:
        table = dynamodb.Table(TABLE_NAME)
        with table.batch_writer() as batch:
            for record in valid_records:
                batch.put_item(Item=record)
        print(f"Loaded {len(valid_records)} records to DynamoDB")

    # Write invalid records to dead-letter prefix for inspection
    if invalid_records:
        timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
        dead_letter_key = f"{DEAD_LETTER_PREFIX}{timestamp}_errors.json"
        s3.put_object(
            Bucket=bucket,
            Key=dead_letter_key,
            Body=json.dumps(invalid_records, indent=2),
            ContentType="application/json",
        )
        print(
            f"Wrote {len(invalid_records)} invalid records to s3://{bucket}/{dead_letter_key}"
        )

    return {
        "statusCode": 200,
        "body": json.dumps(
            {
                "valid_count": len(valid_records),
                "invalid_count": len(invalid_records),
                "source_file": key,
            }
        ),
    }

What does this code do?

This is the entry point that Lambda calls when an S3 upload triggers the function. It orchestrates the full ingestion flow.

  • urllib.parse.unquote_plus decodes the S3 object key. S3 URL-encodes special characters like spaces, so this ensures you get the actual filename.
  • The column-existence check (the missing_cols check) raises an error immediately if the CSV is missing expected headers. This prevents confusing validation errors downstream.
  • batch_writer() automatically chunks writes into batches of 25 items (DynamoDB's limit per request). You don't need to manage batching yourself.
  • Invalid records are written as a timestamped JSON file under the dead-letter/ prefix. This doesn't re-trigger the Lambda because the S3 event filter only watches the raw/ prefix.

Why doesn't the dead-letter write cause an infinite loop?

Your SAM template's S3 event trigger is filtered to the raw/ prefix only. When the Lambda writes error files to dead-letter/, that upload does not match the prefix filter. The Lambda is never re-invoked.

  • Save src/handler.py.
  • You won't see output from this code until you deploy it in the next step. For now, confirm the file is saved with no syntax errors highlighted by your editor.

Seeing a syntax error in your editor?

Check that the f-strings use matching quote types. The print statements inside lambda_handler use f-strings with curly braces. Make sure you haven't accidentally mixed single and double quotes inside a string.

Also verify that the validate_record and transform_record functions are defined above lambda_handler in the file. Python reads top to bottom.

Help me debug a syntax error in my handler.py file.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/handler.py and that your editor shows no red underlines.

ⓧ I'd like to double check the full code

Here is the complete src/handler.py file. Compare yours against it to make sure nothing is missing or out of order.

import csv
import io
import json
import os
import urllib.parse
from datetime import datetime

import boto3

s3 = boto3.client("s3")
dynamodb = boto3.resource("dynamodb")

TABLE_NAME = os.environ["TABLE_NAME"]
DEAD_LETTER_PREFIX = os.environ["DEAD_LETTER_PREFIX"]

REQUIRED_COLUMNS = [
    "date",
    "region",
    "vaccine_type",
    "site_id",
    "person_id",
    "dose_sequence",
]

VALID_REGIONS = [
    "NCR",
    "CAR",
    "Region I",
    "Region II",
    "Region III",
    "Region IV-A",
    "Region IV-B",
    "Region V",
    "Region VI",
    "Region VII",
    "Region VIII",
    "Region IX",
    "Region X",
    "Region XI",
    "Region XII",
    "Region XIII",
    "BARMM",
]


def validate_record(record, row_index):
    errors = []

    date_val = record.get("date", "").strip()
    if not date_val:
        errors.append("missing date")
    else:
        try:
            datetime.strptime(date_val, "%Y-%m-%d")
        except ValueError:
            errors.append(f"invalid date format: {date_val}")

    region_val = record.get("region", "").strip()
    if not region_val:
        errors.append("missing region")
    elif region_val.title() not in [r.title() for r in VALID_REGIONS]:
        errors.append(f"invalid region: {region_val}")

    dose_val = record.get("dose_sequence", "").strip()
    if not dose_val:
        errors.append("missing dose_sequence")
    else:
        try:
            dose_int = int(dose_val)
            if dose_int < 1:
                errors.append(f"dose_sequence must be positive: {dose_val}")
        except ValueError:
            errors.append(f"dose_sequence not an integer: {dose_val}")

    for col in ["vaccine_type", "site_id", "person_id"]:
        if not record.get(col, "").strip():
            errors.append(f"missing {col}")

    return errors


def transform_record(record):
    region = record["region"].strip().title()
    date_str = record["date"].strip()
    region_date = f"{region}#{date_str}"
    event_id = f"{record['person_id'].strip()}#{record['dose_sequence'].strip()}"

    return {
        "region_date": region_date,
        "event_id": event_id,
        "date": date_str,
        "region": region,
        "vaccine_type": record["vaccine_type"].strip(),
        "site_id": record["site_id"].strip(),
        "person_id": record["person_id"].strip(),
        "dose_sequence": int(record["dose_sequence"].strip()),
    }


def lambda_handler(event, context):
    bucket = event["Records"][0]["s3"]["bucket"]["name"]
    key = urllib.parse.unquote_plus(
        event["Records"][0]["s3"]["object"]["key"], encoding="utf-8"
    )

    print(f"Processing file: s3://{bucket}/{key}")

    response = s3.get_object(Bucket=bucket, Key=key)
    content = response["Body"].read().decode("utf-8")

    reader = csv.DictReader(io.StringIO(content))

    missing_cols = [col for col in REQUIRED_COLUMNS if col not in reader.fieldnames]
    if missing_cols:
        error_msg = f"Missing required columns: {missing_cols}"
        print(error_msg)
        raise ValueError(error_msg)

    valid_records = []
    invalid_records = []

    for idx, record in enumerate(reader):
        errors = validate_record(record, idx)
        if errors:
            invalid_records.append(
                {"row": idx + 2, "errors": errors, "data": dict(record)}
            )
        else:
            valid_records.append(transform_record(record))

    print(
        f"Validation complete: {len(valid_records)} valid, "
        f"{len(invalid_records)} invalid"
    )

    if valid_records:
        table = dynamodb.Table(TABLE_NAME)
        with table.batch_writer() as batch:
            for record in valid_records:
                batch.put_item(Item=record)
        print(f"Loaded {len(valid_records)} records to DynamoDB")

    if invalid_records:
        timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
        dead_letter_key = f"{DEAD_LETTER_PREFIX}{timestamp}_errors.json"
        s3.put_object(
            Bucket=bucket,
            Key=dead_letter_key,
            Body=json.dumps(invalid_records, indent=2),
            ContentType="application/json",
        )
        print(
            f"Wrote {len(invalid_records)} invalid records to s3://{bucket}/{dead_letter_key}"
        )

    return {
        "statusCode": 200,
        "body": json.dumps(
            {
                "valid_count": len(valid_records),
                "invalid_count": len(invalid_records),
                "source_file": key,
            }
        ),
    }

Your Lambda function is complete. It knows how to read CSV files from S3, validate every row against the expected schema, transform valid records into DynamoDB items, and safely quarantine bad data. Next up, you'll deploy this function and watch it fire automatically.

Deploy the Pipeline

Your Lambda function code is written and your infrastructure template is ready. Now it is time to deploy everything to AWS as a single, managed stack.

In this step, you will use AWS SAM CLI to build the Lambda deployment package, deploy the full CloudFormation stack to your AWS account, and generate the synthetic vaccination dataset that you will feed into the pipeline.

In this step, get ready to:
  • Build the Lambda deployment package with SAM.
  • Deploy the CloudFormation stack with all resources to AWS.
  • Generate the synthetic vaccination campaign dataset locally.
Build the SAM application

Before deploying, SAM needs to package your Lambda function code along with its dependencies into a deployment-ready artifact. The sam build command reads your template.yaml and src/requirements.txt, then bundles everything into a .aws-sam directory.

  • Make sure your terminal is in the vaccination-pipeline project folder (the one containing template.yaml).
  • Build the Lambda deployment package by running this command:
sam build

What does this command do?

The sam build command creates a .aws-sam directory with packaged artifacts. It installs the dependencies listed in src/requirements.txt (like boto3) alongside your handler.py so that everything the Lambda needs is in one bundle.

  • You should see a success message ending with Build Succeeded.

Build failed?

  • Make sure you are running the command from the directory that contains template.yaml. SAM looks for this file in the current working directory.
  • Check that src/requirements.txt exists and contains boto3>=1.35.0. A missing or empty requirements file can cause build errors.
  • Confirm Python is available on your PATH by running python --version. SAM needs Python to install the dependencies.

Help me troubleshoot my SAM build failure.

Deploy the CloudFormation stack

With the build ready, you can deploy all three resources (S3 bucket, DynamoDB table, and Lambda function) to AWS in one command. SAM will walk you through an interactive guided flow on your first deploy.

  • Deploy the stack by running this command:
sam deploy --guided

What prompts will I see?

SAM will ask you a series of configuration questions. Here is how to answer each one:

  • Stack Name: enter vaccination-ingestion-pipeline (or any name you prefer).
  • AWS Region: press Enter to accept your default region, or type a region like us-east-1.
  • Confirm changes before deploy: press Enter to accept the default (Y).
  • Allow SAM CLI IAM role creation: press Enter to accept Y. This lets SAM create the IAM roles your Lambda needs.
  • Disable rollback: press Enter to accept the default (N).
  • Save arguments to configuration file: press Enter to accept Y. This saves your choices to samconfig.toml for future deploys.
  • SAM configuration file and environment: press Enter twice to accept the defaults.
  • SAM will show you a changeset preview listing the resources it will create. When prompted to confirm, type y and press Enter to deploy.
  • Wait for the deployment to complete. This typically takes 1-2 minutes.

You will see a table of events as each resource is provisioned. Look for CREATE_COMPLETE next to each resource name.

  • Once the deployment finishes, look at the Outputs section at the bottom. Note the BucketName value. You will need it in the next step to upload your CSV file.

Save the bucket name

The bucket name follows the pattern vaccination-ingestion-{YourAccountId}. Copy it somewhere handy. You will use it to upload data in the next step.

Deployment failed?

  • If you see an error about credentials, confirm your AWS CLI is configured by running aws sts get-caller-identity. You should see your account ID in the response.
  • If the S3 bucket name is already taken (bucket names are globally unique), check whether you already have a stack from a previous attempt. Run sam delete to remove it, then re-run sam deploy --guided.
  • If you see a rollback, check the events table in the terminal output for the specific resource that failed. The status reason column tells you what went wrong.

Help me debug my SAM deployment failure.

Generate the synthetic dataset

You need a realistic CSV file to feed into the pipeline. The Philippine vaccination campaign dataset generator you cloned earlier can produce roughly 41,000 vaccination event records across 17 regions.

  • Switch to the terminal where you cloned the dataset generator repository.
  • Install nbconvert so you can run the notebook from the command line:
pip install nbconvert
  • Run the dataset generator notebook by executing it headlessly:
jupyter nbconvert --to notebook --execute --ExecutePreprocessor.timeout=-1 03_dataset_generator.ipynb

What does this command do?

The jupyter nbconvert --execute command runs every cell in the notebook without opening a browser. The --ExecutePreprocessor.timeout=-1 flag removes the default 30-second-per-cell time limit, which is necessary because the generator cells run for several minutes while creating synthetic records.

The notebook creates a data/ folder containing multiple CSV files. The one you need is fact_vaccination_events.csv with approximately 41,000 records.

  • This may take a few minutes to complete. When it finishes, confirm the output file exists by running:
ls data/fact_vaccination_events.csv
  • You should see the file path printed. This confirms the dataset was generated successfully.

Notebook execution failed?

  • Ensure all dependencies were installed in Step 1. Run pip install -r requirements.txt from the dataset generator repo folder to reinstall them.
  • If you see a kernel-related error, make sure you have ipykernel installed by running pip install ipykernel.
  • If the notebook times out despite the flag, try running it again. Transient memory issues can sometimes cause a failure on the first attempt.

Help me fix the dataset generator notebook execution error.

Your infrastructure is live and your test data is ready. Next up, you will upload the CSV to S3 and watch the entire pipeline fire automatically.

Verify the End-to-End Pipeline

Your infrastructure is deployed and your Lambda function is waiting for data. The true test of any event-driven pipeline is whether the entire flow fires automatically when data arrives.

In this step, you will upload your synthetic vaccination dataset to the S3 bucket's raw/ prefix and watch the pipeline execute end-to-end. The Lambda function should trigger, validate records, load them into DynamoDB, and route any invalid records to the dead-letter prefix.

In this step, get ready to:
  • Upload the vaccination CSV to trigger the Lambda automatically.
  • Verify the Lambda execution via CloudWatch Logs.
  • Query DynamoDB to confirm records landed with the correct schema.
Upload the dataset to S3

The S3 event notification you configured in template.yaml listens for any object created under the raw/ prefix. As soon as you upload a file there, Lambda triggers automatically.

  • Upload the generated CSV to your S3 bucket by running this command:
aws s3 cp [[CSV_PATH="path/to/fact_vaccination_events.csv"]] s3://[[BUCKET_NAME="vaccination-ingestion-YOUR-ACCOUNT-ID"]]/raw/fact_vaccination_events.csv

What does this command do?

The aws s3 cp command copies a local file to an S3 location. The raw/ prefix in the destination key is what triggers your Lambda function via the S3 event notification filter.

Replace the first path with wherever you saved fact_vaccination_events.csv from the dataset generator. Replace the bucket name with the BucketName output from your sam deploy step.

  • You should see a progress bar followed by a confirmation that the file was uploaded.

What happens next?

The upload fires an S3 event notification that invokes your IngestionFunction Lambda. The function reads the CSV, validates each row, transforms valid records, and batch-writes them to DynamoDB. This all happens automatically in the background.

It may take 5 to 10 seconds for the Lambda to finish processing ~41K records. Give it a moment before checking the logs.

Upload failed or access denied?

Make sure the bucket name matches the exact output from sam deploy. It follows the pattern vaccination-ingestion-YOUR-ACCOUNT-ID.

If you see an access denied error, confirm your AWS CLI is configured with the same account and region you deployed to.

Help me troubleshoot my S3 upload. I'm getting an error when running aws s3 cp to upload my vaccination CSV to the raw/ prefix.

Check CloudWatch Logs for the Lambda execution

CloudWatch Logs is the default destination for all Lambda function output. Your print() statements in handler.py write directly here.

  • Open the AWS Lambda console in your browser.
  • Select your function from the Functions list (it contains "IngestionFunction" in its name).
  • Choose the Monitor tab.
  • Choose View CloudWatch logs to open the CloudWatch console.
  • Select the most recent Log stream (the one at the top of the list).

In the log stream, you should see output from your Lambda including the file it processed, validation counts, and how many records were loaded.

What should I see in the logs?

Your Lambda prints three key messages during execution:

  • Processing file: s3://... confirms the trigger fired correctly.
  • Validation complete: X valid, Y invalid shows how many records passed schema validation.
  • Loaded X records to DynamoDB confirms the batch write succeeded.
  • Check the dead-letter/ prefix for any invalid record files by running:
aws s3 ls s3://[[BUCKET_NAME="vaccination-ingestion-YOUR-ACCOUNT-ID"]]/dead-letter/

If any records failed validation, you will see a JSON file listed with a timestamp in its name. If all records were valid, you will see no output.

Don't see any log streams?

It may take 5 to 10 minutes for logs to appear after a function invocation. Wait a moment and refresh the page.

If no log group exists at all, the Lambda may not have triggered. Confirm your uploaded file is at the raw/ prefix (not the bucket root) by running:

aws s3 ls s3://[[BUCKET_NAME="vaccination-ingestion-YOUR-ACCOUNT-ID"]]/raw/

Help me debug why my Lambda didn't trigger after uploading a file to S3.

Query DynamoDB to confirm records are present

The final proof is querying the VaccinationEvents table directly to confirm records arrived with the correct schema.

  • Scan the DynamoDB table to retrieve a sample of records by running:
aws dynamodb scan --table-name VaccinationEvents --max-items 5

What does this command do?

The scan operation reads every item in the table. The --max-items 5 flag limits the response to 5 records so your terminal stays readable.

You should see JSON output containing items with these fields:

  • region_date (partition key) in the format Region#YYYY-MM-DD (e.g., Ncr#2023-05-14).
  • event_id (sort key) in the format person_id#dose_sequence.
  • region normalized to title case.
  • dose_sequence stored as a number (not a string).

Seeing zero items in the scan?

If the scan returns an empty Items array, the Lambda may still be processing. With ~41K records, the batch write can take 30-60 seconds. Wait a moment and run the scan again.

If you still see zero items, check CloudWatch Logs for error messages. A common cause is the CSV having different column names than the Lambda expects.

Help me figure out why my DynamoDB table is empty after the Lambda ran.

Secret mission

Cold Chain Risk Monitor

Vaccines lose potency when storage temperatures drift outside the safe range. In this mission, you will add a second event-driven Lambda that monitors cold chain shipment data and flags inventory records where vaccine potency has degraded below a safety threshold.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses serverless resources that only incur charges when triggered, so there are no ongoing idle costs within the AWS Free Tier.

Resources you used:

  • 1 CloudFormation stack containing: 1 S3 bucket, 2 Lambda functions, 1 DynamoDB table, and IAM roles.
  • Local project files (SAM project directory and cloned dataset generator repository).

Keep everything running

No action needed. Choose this if you are still actively building or want to upload more CSV files later.

  • Your Lambda functions only execute when a file is uploaded to the S3 bucket. No uploads means $0 in compute costs.
  • DynamoDB in PAY_PER_REQUEST mode only charges for reads and writes. Stored data stays within the 25 GB always-free tier for this project's volume.
  • S3 storage for the uploaded CSVs and generated JSON files stays well within the 5 GB free tier.
  • The AWS SAM CLI remains installed on your machine. It takes minimal disk space and does not run in the background.

Pause - I'll come back to this later

Serverless resources have no idle cost. There is nothing to pause for this project.

  • Your Lambda functions sleep until a new file upload triggers them. No ongoing charges accrue while they are idle.
  • DynamoDB and S3 only charge for storage and requests. With no new activity, costs stay at $0 within free tier limits.
  • When you want to resume, upload a new CSV to the raw/ or raw-inventory/ prefix and the pipeline fires automatically.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes all cloud resources and local files permanently.

  • First, empty the S3 bucket so CloudFormation can delete it. Run this command:
aws s3 rm s3://vaccination-ingestion-[[AWS_ACCOUNT_ID="your-12-digit-aws-account-id"]] --recursive
  • Delete the entire CloudFormation stack by running this command from your SAM project directory:
sam delete
  • SAM will ask you to confirm the stack deletion and whether to delete the template file in S3. Enter y for both prompts.
  • Delete your local SAM project folder and the cloned dataset generator repository. On macOS or Linux, run:
cd ~/Desktop
rm -rf [[SAM_PROJECT_DIR="your-sam-project-folder-name"]]
rm -rf philippine-vaccination-campaign-synthetic-dataset
  • On Windows, run:
cd ~\Desktop
Remove-Item -Recurse -Force [[SAM_PROJECT_DIR="your-sam-project-folder-name"]]
Remove-Item -Recurse -Force philippine-vaccination-campaign-synthetic-dataset

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade, event-driven data ingestion pipeline using AWS SAM and serverless services.

You've learned how to:

  • Deploy a SAM application with infrastructure-as-code defining S3, Lambda, and DynamoDB resources in a single template.yaml.
  • Build an event-driven Lambda that automatically fires on S3 uploads, validates records against a schema, and routes failures to a dead-letter prefix.
  • Transform and load 41K+ vaccination event records into DynamoDB using batch writes and verify the full pipeline end-to-end with synthetic health data.
  • Secret Mission: Added a cold chain risk monitor that flags degraded vaccine batches from inventory shipment data.

Ready to quiz yourself?