Lab — Provision a Serverless Data Pipeline with Lambda, S3 and DynamoDB

What You'll Build

In this lab you will build a fully serverless data pipeline for the cricket analytics platform. Ball-by-ball match data JSON files are uploaded to an S3 bucket. Each upload automatically triggers a Lambda function via S3 Event Notifications. The Lambda function parses the JSON, validates the data, writes individual delivery records to DynamoDB, and publishes a processing summary to an SNS topic that emails the operations team. You will package and deploy the Lambda function using the AWS CLI, configure IAM permissions using least-privilege roles, set up the S3 event trigger, test the end-to-end pipeline with a sample match file, and monitor execution via CloudWatch Logs. This lab demonstrates the event-driven serverless architecture pattern used by virtually every modern data ingestion pipeline.

Analogy🏏Cricket

🏏 Think of it like cricket: This script is like the pre-match ground inspection conducted by the match referee, pitch curator, and captains before a Test match begins. Before Rohit Sharma and the opposition captain walk out for the toss, the curator has measured pitch moisture, taken grass length readings, and documented the surface condition — creating a baseline against which any afternoon deterioration can be measured.Just as this structured inspection prevents surprises and creates a documented record, the inventory script creates a documented baseline for a server against which future anomalies can be compared. Just as a ground inspection without a checklist might miss a drainage issue that affects the afternoon session, a server assessment without a structured script might miss a nearly-full disk that causes a midnight deployment failure.The insight is that the value of a structured inspection is not just the current findings but the reproducible method — the same script run tomorrow highlights exactly what changed.

Prerequisites

AWS CLI configured with permissions for Lambda, S3, DynamoDB, SNS, IAM and CloudWatch Logs
Python 3.11+ installed locally for writing and testing the Lambda function code
Completed M4 Lessons 1-5: IAM roles, S3 events, DynamoDB write patterns
An email address to receive SNS notifications for pipeline completion
jq installed for JSON processing in setup scripts

Setup — Resources and Project Structure

Create the project directory, S3 buckets, DynamoDB table and SNS topic. The Lambda function code is written locally, packaged as a ZIP and deployed via the AWS CLI — this is the standard deployment pattern for Lambda functions without a framework like SAM or CDK.

bash

#!/bin/bash
# Lab setup: create all required AWS resources
set -euo pipefail

ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
REGION='ap-south-1'
STACK='cricket-pipeline'

INGEST_BUCKET="${STACK}-ingest-${ACCOUNT_ID}"
SCORES_TABLE="${STACK}-scores"
SNS_TOPIC_NAME="${STACK}-alerts"
LAMBDA_NAME="${STACK}-processor"

mkdir -p ~/cricket_pipeline/lambda
cd ~/cricket_pipeline

echo '=== Creating S3 ingestion bucket ==='
aws s3api create-bucket \
    --bucket "$INGEST_BUCKET" \
    --region "$REGION" \
    --create-bucket-configuration LocationConstraint="$REGION"
aws s3api put-public-access-block \
    --bucket "$INGEST_BUCKET" \
    --public-access-block-configuration \
    'BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true'
aws s3api put-bucket-versioning \
    --bucket "$INGEST_BUCKET" \
    --versioning-configuration Status=Enabled
echo "Bucket: ${INGEST_BUCKET}"

echo
echo '=== Creating DynamoDB table ==='
aws dynamodb create-table \
    --table-name "$SCORES_TABLE" \
    --billing-mode PAY_PER_REQUEST \
    --attribute-definitions \
        AttributeName=match_id,AttributeType=S \
        AttributeName=ball_ref,AttributeType=S \
    --key-schema \
        AttributeName=match_id,KeyType=HASH \
        AttributeName=ball_ref,KeyType=RANGE \
    --sse-specification Enabled=true,SSEType=KMS 2>/dev/null || \
    echo 'DynamoDB table already exists'
echo "Table: ${SCORES_TABLE}"

echo
echo '=== Creating SNS topic ==='
SNS_ARN=$(aws sns create-topic \
    --name "$SNS_TOPIC_NAME" \
    --query 'TopicArn' --output text)
echo "SNS ARN: ${SNS_ARN}"

# Subscribe your email
read -r -p 'Enter email for pipeline alerts (or press Enter to skip): ' EMAIL
if [[ -n "$EMAIL" ]]; then
    aws sns subscribe \
        --topic-arn "$SNS_ARN" \
        --protocol email \
        --notification-endpoint "$EMAIL"
    echo "Subscription confirmation sent to ${EMAIL} — confirm to receive alerts"
fi

# Save config for subsequent steps
cat > ~/cricket_pipeline/config.env << CONFIG
export INGEST_BUCKET='${INGEST_BUCKET}'
export SCORES_TABLE='${SCORES_TABLE}'
export SNS_ARN='${SNS_ARN}'
export LAMBDA_NAME='${LAMBDA_NAME}'
export ACCOUNT_ID='${ACCOUNT_ID}'
export REGION='${REGION}'
CONFIG
echo 'Config saved to config.env'

Step 1 — Lambda Function Code and IAM Role

Write the Lambda function that processes cricket match data files. The function reads the uploaded JSON from S3, validates each delivery record, batch-writes to DynamoDB, and publishes a summary notification to SNS. Then create the least-privilege IAM execution role.

Analogy🏏Cricket

🏏 Think of it like cricket: The Lambda function is the automated match data validator — it reads each submitted scorecard file, checks that every delivery has valid bowler, batsman, runs and outcome fields (validation), enters each delivery into the official database (DynamoDB batch_write), and sends the processing summary to the operations director (SNS publish). The IAM execution role is the validator's access pass — it grants exactly the permissions needed: read from the specific ingest bucket, write to the specific DynamoDB table, publish to the specific SNS topic, and write logs to CloudWatch.

bash

# Write the Lambda function
cat > ~/cricket_pipeline/lambda/handler.py << 'PYTHON'
import json
import os
import boto3
from datetime import datetime
from typing import Any

s3       = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
sns      = boto3.client('sns')

TABLE    = os.environ['SCORES_TABLE']
SNS_ARN  = os.environ['SNS_TOPIC_ARN']

def validate_delivery(delivery: dict) -> list[str]:
    """Validate a single delivery record. Returns list of errors."""
    errors = []
    required = ['ball_ref', 'batsman_id', 'bowler_id', 'runs']
    for field in required:
        if field not in delivery:
            errors.append(f'Missing required field: {field}')
    if 'runs' in delivery:
        runs = delivery['runs']
        if not isinstance(runs, int) or runs < 0 or runs > 6:
            errors.append(f'Invalid runs value: {runs} (must be 0-6)')
    return errors

def write_deliveries(match_id: str, deliveries: list[dict]) -> tuple[int, int]:
    """Batch write deliveries to DynamoDB. Returns (written, failed)."""
    table = dynamodb.Table(TABLE)
    written = failed = 0

    # DynamoDB batch_writer handles 25-item batches automatically
    with table.batch_writer() as batch:
        for delivery in deliveries:
            errors = validate_delivery(delivery)
            if errors:
                print(f'Validation failed for {delivery.get("ball_ref","unknown")}: {errors}')
                failed += 1
                continue
            batch.put_item(Item={
                'match_id':   match_id,
                'ball_ref':   delivery['ball_ref'],
                'batsman_id': delivery['batsman_id'],
                'bowler_id':  delivery['bowler_id'],
                'runs':       delivery['runs'],
                'wicket':     delivery.get('wicket', False),
                'extras':     delivery.get('extras', 0),
                'ingested_at': datetime.utcnow().isoformat(),
            })
            written += 1
    return written, failed

def lambda_handler(event: dict, context: Any) -> dict:
    """Process S3 event — read match file and ingest deliveries."""
    results = []

    for record in event.get('Records', []):
        bucket = record['s3']['bucket']['name']
        key    = record['s3']['object']['key']
        print(f'Processing s3://{bucket}/{key}')

        try:
            # Read match file from S3
            obj = s3.get_object(Bucket=bucket, Key=key)
            match_data = json.loads(obj['Body'].read())

            match_id   = match_data.get('match_id', key.split('/')[-1].replace('.json',''))
            deliveries = match_data.get('deliveries', [])

            if not deliveries:
                print(f'Warning: no deliveries in {key}')
                continue

            written, failed = write_deliveries(match_id, deliveries)

            summary = {
                'match_id': match_id,
                'file':     key,
                'total':    len(deliveries),
                'written':  written,
                'failed':   failed,
                'status':   'SUCCESS' if failed == 0 else 'PARTIAL',
            }
            results.append(summary)
            print(f'Processed {match_id}: {written} written, {failed} failed')

        except Exception as exc:
            print(f'ERROR processing {key}: {exc}')
            results.append({'file': key, 'status': 'ERROR', 'error': str(exc)})

    # Publish summary to SNS
    sns.publish(
        TopicArn=SNS_ARN,
        Subject=f'Cricket pipeline: processed {len(results)} file(s)',
        Message=json.dumps(results, indent=2),
    )

    return {'statusCode': 200, 'body': json.dumps(results)}
PYTHON

echo 'Lambda handler written'

# Package as ZIP
cd ~/cricket_pipeline/lambda
zip -q ../function.zip handler.py
echo "Package: $(ls -lh ../function.zip | awk '{print $5}')"

Step 2 — Deploy Lambda and Configure S3 Trigger

Create the IAM execution role, deploy the Lambda function with the ZIP package, and configure the S3 bucket to send event notifications to Lambda when new JSON files are uploaded to the raw/ prefix. The S3-to-Lambda trigger requires a Lambda resource-based policy that allows S3 to invoke the function.

Analogy🏏Cricket

🏏 Think of it like cricket: The S3 event notification configuration is the scorer's filing system trigger — when a new scorecard is placed in the 'raw' tray (S3 prefix raw/), the system automatically rings the data validator's phone (invokes Lambda). The resource-based policy is the validator's published phone number — it authorises the filing system (S3) to call the validator directly, which is a separate authorisation from the validator's own credentials (IAM execution role). Both must be configured: S3 needs permission to call Lambda, and Lambda needs permission to read from S3 and write to DynamoDB.

bash

#!/bin/bash
# Step 2: IAM role, Lambda deployment and S3 trigger
source ~/cricket_pipeline/config.env
cd ~/cricket_pipeline

echo '=== Creating Lambda IAM execution role ==='
# Trust policy
ROLE_ARN=$(aws iam create-role \
    --role-name "${LAMBDA_NAME}-role" \
    --assume-role-policy-document '{
        "Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]
    }' \
    --query 'Role.Arn' --output text 2>/dev/null || \
    aws iam get-role --role-name "${LAMBDA_NAME}-role" \
        --query 'Role.Arn' --output text)

# Least-privilege permissions policy
aws iam put-role-policy \
    --role-name "${LAMBDA_NAME}-role" \
    --policy-name CricketPipelinePolicy \
    --policy-document "{
        \"Statement\": [
            {\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\"],
             \"Resource\":\"arn:aws:s3:::${INGEST_BUCKET}/*\"},
            {\"Effect\":\"Allow\",\"Action\":[\"dynamodb:BatchWriteItem\",\"dynamodb:PutItem\"],
             \"Resource\":\"arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${SCORES_TABLE}\"},
            {\"Effect\":\"Allow\",\"Action\":[\"sns:Publish\"],
             \"Resource\":\"${SNS_ARN}\"},
            {\"Effect\":\"Allow\",\"Action\":[\"logs:CreateLogGroup\",\"logs:CreateLogStream\",\"logs:PutLogEvents\"],
             \"Resource\":\"arn:aws:logs:${REGION}:${ACCOUNT_ID}:log-group:/aws/lambda/${LAMBDA_NAME}:*\"}
        ]
    }"
echo "Execution role ARN: ${ROLE_ARN}"
sleep 10  # IAM propagation

echo
echo '=== Deploying Lambda function ==='
LAMBDA_ARN=$(aws lambda create-function \
    --function-name "$LAMBDA_NAME" \
    --runtime python3.11 \
    --handler handler.lambda_handler \
    --role "$ROLE_ARN" \
    --zip-file fileb://function.zip \
    --environment "Variables={SCORES_TABLE=${SCORES_TABLE},SNS_TOPIC_ARN=${SNS_ARN}}" \
    --timeout 60 \
    --memory-size 256 \
    --query 'FunctionArn' --output text 2>/dev/null || \
    aws lambda update-function-code \
        --function-name "$LAMBDA_NAME" \
        --zip-file fileb://function.zip \
        --query 'FunctionArn' --output text)
echo "Lambda ARN: ${LAMBDA_ARN}"

echo
echo '=== Configuring S3 event notification ==='
# Allow S3 to invoke Lambda
aws lambda add-permission \
    --function-name "$LAMBDA_NAME" \
    --statement-id AllowS3Invoke \
    --action lambda:InvokeFunction \
    --principal s3.amazonaws.com \
    --source-arn "arn:aws:s3:::${INGEST_BUCKET}" \
    --source-account "$ACCOUNT_ID" 2>/dev/null || echo 'Permission already exists'

# Set S3 notification
aws s3api put-bucket-notification-configuration \
    --bucket "$INGEST_BUCKET" \
    --notification-configuration "{
        \"LambdaFunctionConfigurations\": [{
            \"LambdaFunctionArn\": \"${LAMBDA_ARN}\",
            \"Events\": [\"s3:ObjectCreated:*\"],
            \"Filter\": {
                \"Key\": {\"FilterRules\": [
                    {\"Name\": \"prefix\", \"Value\": \"raw/\"},
                    {\"Name\": \"suffix\", \"Value\": \".json\"}
                ]}
            }
        }]
    }"
echo 'S3 → Lambda trigger configured'

Step 3 — End-to-End Test and Monitoring

Upload a sample match data file to S3, wait for Lambda to process it, verify the records in DynamoDB, and inspect the CloudWatch Logs for the Lambda execution. Then introduce a validation error to confirm the function handles malformed data gracefully.

Analogy🏏Cricket

🏏 Think of it like cricket: The end-to-end test is the pre-tournament dress rehearsal — the scorer submits a test scorecard for a practice match (sample JSON upload), the validation system processes it automatically, and the operations team confirms that the official database received all deliveries correctly and the notification email arrived. Testing with a malformed record (missing bowler_id) is the equivalent of submitting a scorecard with a blank bowler column — the system should flag it with a clear error message rather than silently storing corrupt data or crashing entirely.

bash

#!/bin/bash
# Step 3: end-to-end testing
source ~/cricket_pipeline/config.env

echo '=== Creating sample match data ==='
cat > /tmp/ipl_2024_042.json << 'MATCH'
{
  "match_id": "IPL_2024_042",
  "match": "Mumbai Indians vs Chennai Super Kings",
  "venue": "Wankhede Stadium",
  "date": "2024-04-15",
  "deliveries": [
    {"ball_ref": "1.1", "batsman_id": "ROHIT_SHARMA",   "bowler_id": "JADEJA_R",  "runs": 4, "wicket": false},
    {"ball_ref": "1.2", "batsman_id": "ROHIT_SHARMA",   "bowler_id": "JADEJA_R",  "runs": 0, "wicket": false},
    {"ball_ref": "1.3", "batsman_id": "ROHIT_SHARMA",   "bowler_id": "JADEJA_R",  "runs": 6, "wicket": false},
    {"ball_ref": "1.4", "batsman_id": "ISHAN_KISHAN",   "bowler_id": "JADEJA_R",  "runs": 1, "wicket": false},
    {"ball_ref": "1.5", "batsman_id": "ROHIT_SHARMA",   "bowler_id": "JADEJA_R",  "runs": 0, "wicket": true,  "wicket_type": "caught"},
    {"ball_ref": "1.6", "batsman_id": "SURYAKUMAR_Y",   "bowler_id": "JADEJA_R",  "runs": 2, "wicket": false},
    {"ball_ref": "2.1", "batsman_id": "SURYAKUMAR_Y",   "bowler_id": "BUMRAH_J",  "runs": 4, "wicket": false},
    {"ball_ref": "2.2", "batsman_id": "ISHAN_KISHAN",   "bowler_id": "BUMRAH_J",  "runs": 0, "wicket": false},
    {"ball_ref": "2.3", "batsman_id": "ISHAN_KISHAN",   "bowler_id": "BUMRAH_J",  "runs": 1, "wicket": false},
    {"ball_ref": "2.4", "batsman_id": "SURYAKUMAR_Y",   "bowler_id": "BUMRAH_J",  "runs": 6, "wicket": false},
    {"ball_ref": "2.5", "batsman_id": "SURYAKUMAR_Y",   "bowler_id": "BUMRAH_J",  "runs": 4, "wicket": false},
    {"ball_ref": "bad1","batsman_id": "HARDIK_PANDYA",                              "runs": 2, "wicket": false}
  ]
}
MATCH

# Upload to S3 ingest bucket (triggers Lambda automatically)
echo '=== Uploading match data to S3 ==='
aws s3 cp /tmp/ipl_2024_042.json "s3://${INGEST_BUCKET}/raw/2024/ipl_2024_042.json"
echo 'Upload complete — Lambda trigger activated'

# Wait for processing
echo 'Waiting 15 seconds for Lambda to process...'
sleep 15

# Verify DynamoDB records
echo
echo '=== Verifying DynamoDB records ==='
COUNT=$(aws dynamodb query \
    --table-name "$SCORES_TABLE" \
    --key-condition-expression 'match_id = :mid' \
    --expression-attribute-values '{":mid":{"S":"IPL_2024_042"}}' \
    --select COUNT \
    --query 'Count' --output text)
echo "Records in DynamoDB for IPL_2024_042: ${COUNT} (expected: 11 — bad1 should be rejected)"

# Sample records
aws dynamodb query \
    --table-name "$SCORES_TABLE" \
    --key-condition-expression 'match_id = :mid' \
    --expression-attribute-values '{":mid":{"S":"IPL_2024_042"}}' \
    --query 'Items[*].{Ball:ball_ref.S, Batsman:batsman_id.S, Bowler:bowler_id.S, Runs:runs.N}' \
    --output table

# Check CloudWatch Logs
echo
echo '=== Lambda CloudWatch Logs (last execution) ==='
LOG_GROUP="/aws/lambda/${LAMBDA_NAME}"
LATEST_STREAM=$(aws logs describe-log-streams \
    --log-group-name "$LOG_GROUP" \
    --order-by LastEventTime \
    --descending \
    --query 'logStreams[0].logStreamName' --output text 2>/dev/null)

if [[ -n "$LATEST_STREAM" && "$LATEST_STREAM" != 'None' ]]; then
    aws logs get-log-events \
        --log-group-name "$LOG_GROUP" \
        --log-stream-name "$LATEST_STREAM" \
        --query 'events[*].message' --output text | head -20
else
    echo 'No log streams yet — Lambda may not have executed (check S3 trigger config)'
fi

Step 4 — Cleanup

bash

#!/bin/bash
# Cleanup: remove all lab resources
source ~/cricket_pipeline/config.env

echo '=== Removing S3 event notification ==='
aws s3api put-bucket-notification-configuration \
    --bucket "$INGEST_BUCKET" \
    --notification-configuration '{}'

echo '=== Deleting Lambda function ==='
aws lambda delete-function --function-name "$LAMBDA_NAME" 2>/dev/null || true

echo '=== Deleting IAM role and policy ==='
aws iam delete-role-policy \
    --role-name "${LAMBDA_NAME}-role" \
    --policy-name CricketPipelinePolicy 2>/dev/null || true
aws iam delete-role --role-name "${LAMBDA_NAME}-role" 2>/dev/null || true

echo '=== Emptying and deleting S3 bucket ==='
aws s3 rm "s3://${INGEST_BUCKET}" --recursive 2>/dev/null || true
aws s3api delete-bucket --bucket "$INGEST_BUCKET" 2>/dev/null || true

echo '=== Deleting DynamoDB table ==='
aws dynamodb delete-table --table-name "$SCORES_TABLE" 2>/dev/null || true

echo '=== Deleting SNS topic ==='
aws sns delete-topic --topic-arn "$SNS_ARN" 2>/dev/null || true

echo '=== Cleanup complete ==='
rm -f ~/cricket_pipeline/config.env

Warning: S3 buckets with versioning enabled cannot be deleted until all versions of all objects are deleted — including delete markers. The cleanup script uses aws s3 rm --recursive which removes only current versions. If the bucket still fails to delete, run: aws s3api delete-objects --bucket BUCKET --delete "$(aws s3api list-object-versions --bucket BUCKET --query '{Objects: Versions[].{Key:Key,VersionId:VersionId}}')" to delete all versions before the bucket deletion.

Extension Challenge: Extend the pipeline with three production enhancements: (1) Add a Dead Letter Queue (DLQ) — an SQS queue configured as the Lambda function's destination for failed invocations — so that files that cause Lambda errors are not silently lost; (2) Add an AWS Glue Crawler that automatically discovers the DynamoDB table schema and registers it in the Glue Data Catalog, enabling Athena SQL queries directly against DynamoDB data; (3) Add Lambda Powertools (Python library) for structured JSON logging, tracing with X-Ray and metrics — this adds production observability to the function in under 20 lines of code.

S3 event notifications require two permissions: an S3 bucket notification configuration (what to notify) and a Lambda resource-based policy (allowing S3 to invoke the function).
DynamoDB batch_writer handles the 25-item batch limit automatically — use it for bulk writes rather than calling put_item in a loop, which is slower and more expensive.
Lambda IAM execution roles must be least-privilege — the function in this lab needs only s3:GetObject (specific bucket), dynamodb:BatchWriteItem (specific table) and sns:Publish (specific topic).
CloudWatch Logs are the primary debugging tool for Lambda — every print() statement and exception traceback appears in the log stream for the function execution.
Validate data before writing to DynamoDB — invalid records should be counted and reported (not silently skipped or crash the function) so the operations team knows exactly how many deliveries failed validation.
Use environment variables for Lambda configuration (table names, SNS ARNs) — hardcoding resource identifiers makes the function non-portable between environments.