Serverless CSV Data Pipeline on AWS

Build an event-driven pipeline that validates and transforms CSV data on S3.

Introduction

30 Second Summary

Every time you order something online, a row of data lands somewhere. That order needs to be checked for errors, cleaned up, and stored before anyone can use it for reports or decisions.

In this project, you will build a serverless pipeline on AWS that automatically validates, cleans, and catalogs CSV sales data the moment it arrives. You will query the processed results with SQL and set up an alarm that notifies you when something goes wrong.

What You'll Build

An event-driven data pipeline that fires automatically when a CSV file is uploaded to Amazon S3, processes it through quality gates in AWS Lambda, and makes the clean output queryable with Amazon Athena.

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

  • A Lambda function that validates schemas, rejects bad rows, and transforms dates and text automatically.
  • A Glue Crawler and Data Catalog table so you can run SQL analytics on processed data with Athena.
  • A CloudWatch alarm that sends you an email notification whenever the pipeline encounters an error.
  • Secret Mission: Add a dead letter queue to capture failed events and log failure details for auditing.

Are there any prerequisites?

You need an AWS account (Free Tier eligible preferred). No local tools or installations are required. Everything in this project is built through the AWS Management Console in your browser.

Create the S3 Bucket and IAM Role

Every data pipeline needs two things: a place to store data, and permissions that control what can access it.

In this step, you'll create an S3 bucket to hold your raw and processed CSV files. Then you'll create an IAM role with least-privilege permissions so your Lambda function can read from one prefix and write to another.

In this step, get ready to:
  • Create an S3 bucket with raw/ and processed/ prefixes.
  • Create an IAM role with a custom policy scoped to only the permissions your Lambda needs.
Create your S3 bucket

Your pipeline uses a single S3 bucket split into two prefixes. The raw/ prefix is where you upload unprocessed CSV files, and the processed/ prefix is where Lambda writes the cleaned output.

  • In the AWS Management Console, search for S3 in the top search bar and select S3 from the results.
  • Click Create bucket.
  • Under AWS Region, select US East (N. Virginia) us-east-1.
  • In the Bucket name field, enter csv-pipeline- followed by your initials, followed by -raw-processed (for example, csv-pipeline-jd-raw-processed).
  • Leave all other settings as their defaults (versioning off, public access blocked).
  • Scroll to the bottom and click Create bucket.

Why add your initials to the bucket name?

S3 bucket names must be globally unique across all AWS accounts worldwide. Adding your initials prevents naming conflicts with other learners who are also completing this project.

You should see a green success banner at the top of the page confirming your bucket was created.

Seeing a "Bucket already exists" error?

This means someone else already has a bucket with that exact name. Try adding extra characters to make it more unique (for example, add a number like csv-pipeline-jd2-raw-processed).

Help me fix my S3 bucket name

Create the raw/ and processed/ prefixes

S3 does not have traditional folders. Instead, it uses prefixes (path-like strings) to organize objects.

  • Click on your newly created bucket name to open it.
  • Click Create folder.
  • Enter raw as the folder name.
  • Click Create folder.
  • Click on your bucket name in the breadcrumb navigation at the top to return to the bucket root.
  • Click Create folder again.
  • Enter processed as the folder name.
  • Click Create folder.

Navigate back to your bucket root. You should now see both raw/ and processed/ listed as folders in your bucket.

Why separate prefixes instead of separate buckets?

Using prefixes within a single bucket keeps your infrastructure simple while still letting you set granular permissions. Your Lambda will only have read access to raw/ and write access to processed/.

This prefix separation also prevents a dangerous recursive trigger loop. The S3 event trigger will only fire for objects created in raw/, so when Lambda writes output to processed/ it does not re-trigger itself.

Create the IAM role for Lambda

Your Lambda function needs permission to read files from S3, write processed files back to S3, and log its activity to CloudWatch. You will create a dedicated IAM role with only the exact permissions it needs.

  • In the AWS Console search bar, search for IAM and select IAM from the results.
  • In the left sidebar, click Roles.
  • Click Create role.
  • Under Trusted entity type, select AWS service.
  • Under Use case, select Lambda.
  • Click Next.
  • On the Add permissions page, do not select any managed policies. Click Next to continue without attaching any.
  • For Role name, enter csv-pipeline-lambda-role.
  • Click Create role.

You should see a success banner confirming the role was created. Now you need to add a custom inline policy with your specific permissions.

  • Click on the role name csv-pipeline-lambda-role to open it.
  • Click the Permissions tab.
  • Click Add permissions, then select Create inline policy.
  • Click the JSON tab in the policy editor.
  • Delete the existing template content and paste the following policy:
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::[[BUCKET_NAME=\"csv-pipeline-yourname-raw-processed\"]]/raw/*"
    },
    {
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::[[BUCKET_NAME=\"csv-pipeline-yourname-raw-processed\"]]/processed/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    }
  ]
}

What does this policy do?

  • s3:GetObject on raw/* lets Lambda read uploaded CSV files from the raw prefix only.
  • s3:PutObject on processed/* lets Lambda write cleaned output to the processed prefix only.
  • The logs: actions let Lambda create log groups and write log entries to CloudWatch so you can see what happened during each execution.

This follows the principle of least privilege. The function can only access the exact resources it needs.

  • Replace both instances of csv-pipeline-yourname-raw-processed with your actual bucket name.
  • Click Next.
  • For Policy name, enter csv-pipeline-lambda-policy.
  • Click Create policy.

You should see the policy listed under the role's Permissions policies section with the name csv-pipeline-lambda-policy.

Seeing a "MalformedPolicyDocument" error?

Double check that you replaced the bucket name placeholder in both the s3:GetObject and s3:PutObject resource ARNs. Make sure there are no extra spaces or missing quotes in the JSON.

Help me fix my IAM policy JSON

Your storage layer and permissions are ready. Next up, you'll build the Lambda function that actually reads, validates, and transforms your CSV data.

Build the Lambda Function with Quality Gates

You have an S3 bucket and IAM role ready. Now it's time to build the core processing engine of your pipeline.

The Lambda function is where the real data engineering happens. It will validate incoming CSV schemas, reject bad rows, transform messy data into clean output, and write the results to your processed prefix.

In this step, get ready to:
  • Create a Lambda function with the Python 3.13 runtime and your IAM role.
  • Write Python code that validates schemas, rejects bad rows, and transforms CSV data.
  • Test the function with a sample S3 event to confirm it runs without errors.
Create the Lambda function

You need a Lambda function configured with the correct runtime, permissions, and resource limits to process CSV files reliably.

  • Open the Lambda console by searching for Lambda in the AWS console search bar and selecting it.
  • Click Create function.
  • Select Author from scratch.
  • For Function name, enter csv-pipeline-processor.
  • For Runtime, choose Python 3.13.
  • Leave Architecture set to x86_64.
  • Expand the Change default execution role section.
  • Select Use an existing role.
  • Under Existing role, select csv-pipeline-lambda-role.
  • Click Create function.

You should see a success banner confirming your function was created. Now you need to increase the timeout and memory from the defaults.

  • Select the Configuration tab.
  • In the General configuration pane, click Edit.
  • Set Memory to 256 MB.
  • Set Timeout to 1 min 0 sec (60 seconds total).
  • Click Save.

Why increase timeout and memory?

The default timeout is 3 seconds and default memory is 128 MB. CSV processing involves downloading, parsing, and re-uploading files. 60 seconds and 256 MB give you comfortable headroom for files with thousands of rows.

Don't see your IAM role in the dropdown?

Make sure you're in the same region (us-east-1) where you created the role. IAM roles are global, but the console region selector can affect what you see. Also confirm the role's trusted entity is set to Lambda.

help me find my IAM role for Lambda

Write the CSV processing code

The function code handles three responsibilities: schema validation (are the right columns present?), quality checks (are required values populated and valid?), and transformation (standardize formatting). You will write this code in Lambda's built-in editor.

  • Navigate back to the Code tab of your function.
  • In the code editor, click on lambda_function.py to open it.
  • Select all the existing code (Cmd+A on macOS or Ctrl+A on Windows) and delete it.

Now paste the code in three sections. Start with the imports and configuration by pasting the following at the top of the file:

import json
import csv
import io
import logging
import boto3
from datetime import datetime
from urllib.parse import unquote_plus

# Set up logging to CloudWatch
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Initialize S3 client outside the handler for reuse across invocations
s3_client = boto3.client('s3')

# Define the expected CSV schema
EXPECTED_COLUMNS = ['order_id', 'customer_name', 'product', 'quantity', 'unit_price', 'order_date']
REQUIRED_FIELDS = ['order_id', 'quantity', 'unit_price']

What does this code do?

  • The imports bring in Python's standard csv, json, io, and logging modules. No external packages are needed because boto3 comes pre-installed in the Lambda Python runtime.
  • EXPECTED_COLUMNS defines the exact columns a valid CSV must contain. REQUIRED_FIELDS lists which fields cannot be empty in any row.
  • The S3 client is created outside the handler so Lambda can reuse it across warm invocations, improving performance.
  • Below the configuration code, paste the three helper functions that handle validation and transformation:

def validate_schema(headers):
    """Validate that the CSV has the expected columns."""
    normalized_headers = [h.strip().lower().replace(' ', '_') for h in headers]
    missing = [col for col in EXPECTED_COLUMNS if col not in normalized_headers]
    if missing:
        raise ValueError(f"Schema validation failed. Missing columns: {missing}")
    return normalized_headers


def validate_row(row, row_number):
    """Run quality checks on a single row. Returns (is_valid, cleaned_row, issues)."""
    issues = []

    # Check required fields are not empty
    for field in REQUIRED_FIELDS:
        if not row.get(field) or row[field].strip() == '':
            issues.append(f"Row {row_number}: Missing required field '{field}'")
            return False, None, issues

    # Validate quantity is a positive integer
    try:
        quantity = int(row['quantity'])
        if quantity <= 0:
            issues.append(f"Row {row_number}: Quantity must be positive, got '{row['quantity']}'")
            return False, None, issues
        row['quantity'] = str(quantity)
    except ValueError:
        issues.append(f"Row {row_number}: Invalid quantity '{row['quantity']}'")
        return False, None, issues

    # Validate unit_price is a non-negative number
    try:
        unit_price = float(row['unit_price'])
        if unit_price < 0:
            issues.append(f"Row {row_number}: Negative unit_price '{row['unit_price']}'")
            return False, None, issues
        row['unit_price'] = f"{unit_price:.2f}"
    except ValueError:
        issues.append(f"Row {row_number}: Invalid unit_price '{row['unit_price']}'")
        return False, None, issues

    return True, row, issues


def transform_row(row):
    """Apply transformations to a validated row."""
    # Standardize customer name: strip whitespace and title case
    row['customer_name'] = row['customer_name'].strip().title()
    row['product'] = row['product'].strip()

    # Standardize date to ISO format (YYYY-MM-DD)
    order_date = row['order_date'].strip()
    for fmt in ('%m/%d/%Y', '%d-%m-%Y', '%Y-%m-%d', '%m-%d-%Y', '%d/%m/%Y'):
        try:
            parsed_date = datetime.strptime(order_date, fmt)
            row['order_date'] = parsed_date.strftime('%Y-%m-%d')
            break
        except ValueError:
            continue

    return row

What do these functions do?

  • validate_schema() normalizes the CSV headers to lowercase with underscores, then checks that every expected column is present. If any are missing, it raises an error that stops processing.
  • validate_row() checks each row individually. It rejects rows with missing required fields, non-numeric quantities, zero/negative quantities, and invalid prices. This is your quality gate.
  • transform_row() cleans validated rows by trimming whitespace, title-casing names, and converting dates from any of five common formats into ISO YYYY-MM-DD format.
  • Finally, paste the main handler function below the helpers:

def lambda_handler(event, context):
    """Main handler triggered by S3 event."""
    logger.info(f"Received event: {json.dumps(event)}")

    # Extract bucket and key from the S3 event
    record = event['Records'][0]
    bucket = record['s3']['bucket']['name']
    key = unquote_plus(record['s3']['object']['key'])

    logger.info(f"Processing file: s3://{bucket}/{key}")

    # Download the CSV from S3
    response = s3_client.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read().decode('utf-8')

    # Parse the CSV
    reader = csv.DictReader(io.StringIO(content))

    headers = reader.fieldnames
    if not headers:
        raise ValueError("CSV file is empty or has no headers")

    normalized_headers = validate_schema(headers)

    # Process each row through quality checks and transformation
    valid_rows = []
    rejected_rows = []
    all_issues = []
    row_number = 0

    for raw_row in reader:
        row_number += 1
        row = {normalized_headers[i]: v for i, v in enumerate(raw_row.values())}

        is_valid, cleaned_row, issues = validate_row(row, row_number)
        if is_valid:
            transformed_row = transform_row(cleaned_row)
            valid_rows.append(transformed_row)
        else:
            rejected_rows.append(row)
            all_issues.extend(issues)

    logger.info(f"Processing complete: {len(valid_rows)} valid, {len(rejected_rows)} rejected out of {row_number} total rows")
    if all_issues:
        logger.warning(f"Quality issues found: {json.dumps(all_issues)}")

    # Write valid rows to the processed prefix
    if valid_rows:
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=EXPECTED_COLUMNS)
        writer.writeheader()
        writer.writerows(valid_rows)

        processed_key = key.replace('raw/', 'processed/', 1)
        s3_client.put_object(
            Bucket=bucket,
            Key=processed_key,
            Body=output.getvalue().encode('utf-8'),
            ContentType='text/csv'
        )
        logger.info(f"Processed file written to: s3://{bucket}/{processed_key}")

    return {
        'statusCode': 200,
        'body': json.dumps({
            'source_file': f"s3://{bucket}/{key}",
            'total_rows': row_number,
            'valid_rows': len(valid_rows),
            'rejected_rows': len(rejected_rows),
            'output_file': f"s3://{bucket}/{key.replace('raw/', 'processed/', 1)}" if valid_rows else None
        })
    }

What does the handler do?

  • It reads the S3 event to determine which bucket and file key triggered the function. The unquote_plus call handles URL-encoded characters (like spaces) in file names.
  • It downloads the CSV from S3, validates the schema, then loops through every row running quality checks and transformations.
  • Valid rows get written to the processed/ prefix as a clean CSV. The key swap (raw/ to processed/) keeps the same file name but changes the prefix.
  • The return value summarizes what happened: how many rows passed, how many were rejected, and where the output file lives.

✔️ Awesome, I've got everything!

Great. Make sure you've pasted all three sections in order (imports, helpers, handler) into a single lambda_function.py file with no gaps.

ⓧ I'd like to double check the full code

Here is the complete lambda_function.py file. Your code should match this exactly:

import json
import csv
import io
import logging
import boto3
from datetime import datetime
from urllib.parse import unquote_plus

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3_client = boto3.client('s3')

EXPECTED_COLUMNS = ['order_id', 'customer_name', 'product', 'quantity', 'unit_price', 'order_date']
REQUIRED_FIELDS = ['order_id', 'quantity', 'unit_price']


def validate_schema(headers):
    """Validate that the CSV has the expected columns."""
    normalized_headers = [h.strip().lower().replace(' ', '_') for h in headers]
    missing = [col for col in EXPECTED_COLUMNS if col not in normalized_headers]
    if missing:
        raise ValueError(f"Schema validation failed. Missing columns: {missing}")
    return normalized_headers


def validate_row(row, row_number):
    """Run quality checks on a single row. Returns (is_valid, cleaned_row, issues)."""
    issues = []

    for field in REQUIRED_FIELDS:
        if not row.get(field) or row[field].strip() == '':
            issues.append(f"Row {row_number}: Missing required field '{field}'")
            return False, None, issues

    try:
        quantity = int(row['quantity'])
        if quantity <= 0:
            issues.append(f"Row {row_number}: Quantity must be positive, got '{row['quantity']}'")
            return False, None, issues
        row['quantity'] = str(quantity)
    except ValueError:
        issues.append(f"Row {row_number}: Invalid quantity '{row['quantity']}'")
        return False, None, issues

    try:
        unit_price = float(row['unit_price'])
        if unit_price < 0:
            issues.append(f"Row {row_number}: Negative unit_price '{row['unit_price']}'")
            return False, None, issues
        row['unit_price'] = f"{unit_price:.2f}"
    except ValueError:
        issues.append(f"Row {row_number}: Invalid unit_price '{row['unit_price']}'")
        return False, None, issues

    return True, row, issues


def transform_row(row):
    """Apply transformations to a validated row."""
    row['customer_name'] = row['customer_name'].strip().title()
    row['product'] = row['product'].strip()

    order_date = row['order_date'].strip()
    for fmt in ('%m/%d/%Y', '%d-%m-%Y', '%Y-%m-%d', '%m-%d-%Y', '%d/%m/%Y'):
        try:
            parsed_date = datetime.strptime(order_date, fmt)
            row['order_date'] = parsed_date.strftime('%Y-%m-%d')
            break
        except ValueError:
            continue

    return row


def lambda_handler(event, context):
    """Main handler triggered by S3 event."""
    logger.info(f"Received event: {json.dumps(event)}")

    record = event['Records'][0]
    bucket = record['s3']['bucket']['name']
    key = unquote_plus(record['s3']['object']['key'])

    logger.info(f"Processing file: s3://{bucket}/{key}")

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

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

    headers = reader.fieldnames
    if not headers:
        raise ValueError("CSV file is empty or has no headers")

    normalized_headers = validate_schema(headers)

    valid_rows = []
    rejected_rows = []
    all_issues = []
    row_number = 0

    for raw_row in reader:
        row_number += 1
        row = {normalized_headers[i]: v for i, v in enumerate(raw_row.values())}

        is_valid, cleaned_row, issues = validate_row(row, row_number)
        if is_valid:
            transformed_row = transform_row(cleaned_row)
            valid_rows.append(transformed_row)
        else:
            rejected_rows.append(row)
            all_issues.extend(issues)

    logger.info(f"Processing complete: {len(valid_rows)} valid, {len(rejected_rows)} rejected out of {row_number} total rows")
    if all_issues:
        logger.warning(f"Quality issues found: {json.dumps(all_issues)}")

    if valid_rows:
        output = io.StringIO()
        writer = csv.DictWriter(output, fieldnames=EXPECTED_COLUMNS)
        writer.writeheader()
        writer.writerows(valid_rows)

        processed_key = key.replace('raw/', 'processed/', 1)
        s3_client.put_object(
            Bucket=bucket,
            Key=processed_key,
            Body=output.getvalue().encode('utf-8'),
            ContentType='text/csv'
        )
        logger.info(f"Processed file written to: s3://{bucket}/{processed_key}")

    return {
        'statusCode': 200,
        'body': json.dumps({
            'source_file': f"s3://{bucket}/{key}",
            'total_rows': row_number,
            'valid_rows': len(valid_rows),
            'rejected_rows': len(rejected_rows),
            'output_file': f"s3://{bucket}/{key.replace('raw/', 'processed/', 1)}" if valid_rows else None
        })
    }
Deploy and test the function

With all the code in place, you need to deploy it and verify it executes without errors using a simulated S3 event.

  • In the DEPLOY section of the code editor, click Deploy.
  • You should see a "Successfully updated the function" banner at the top.

Now create a test event that simulates an S3 file upload to your raw/ prefix.

  • Click the Test tab.
  • Under Test event, select Create new event.
  • For Event name, enter testS3Upload.
  • Replace the default event JSON with the following (update the bucket name to match yours):
{
  "Records": [
    {
      "eventVersion": "2.0",
      "eventSource": "aws:s3",
      "awsRegion": "us-east-1",
      "eventTime": "1970-01-01T00:00:00.000Z",
      "eventName": "ObjectCreated:Put",
      "userIdentity": {
        "principalId": "EXAMPLE"
      },
      "requestParameters": {
        "sourceIPAddress": "127.0.0.1"
      },
      "responseElements": {
        "x-amz-request-id": "EXAMPLE123456789",
        "x-amz-id-2": "EXAMPLE123/5678abcdefghijklambdaisawesome/mnopqrstuvwxyzABCDEFGH"
      },
      "s3": {
        "s3SchemaVersion": "1.0",
        "configurationId": "testConfigRule",
        "bucket": {
          "name": "[[BUCKET_NAME=\"csv-pipeline-yourname-raw-processed\"]]",
          "ownerIdentity": {
            "principalId": "EXAMPLE"
          },
          "arn": "arn:aws:s3:::[[BUCKET_NAME=\"csv-pipeline-yourname-raw-processed\"]]"
        },
        "object": {
          "key": "raw/sample_orders.csv",
          "size": 1024,
          "eTag": "0123456789abcdef0123456789abcdef",
          "sequencer": "0A1B2C3D4E5F678901"
        }
      }
    }
  ]
}

What is this test event?

This JSON simulates what S3 sends to Lambda when a file is uploaded. It tells Lambda which bucket and which key (file path) triggered the event. The function will try to download raw/sample_orders.csv from your bucket.

Since you have not uploaded a CSV file yet, this test will produce an error (the file does not exist in S3). That is expected for now. The goal is to confirm the function deploys and starts executing without a code syntax error.

  • Click Save.
  • Click Test.

Under Execution result, expand Details. You should see an error message containing NoSuchKey or The specified key does not exist. This confirms your code deployed correctly and Lambda is attempting to download the file from S3. The error only occurs because the sample CSV file does not exist yet.

Seeing a syntax error instead?

If you see SyntaxError or IndentationError, your code has a formatting problem. Open the double-check tab above and compare your code character by character. Python is sensitive to indentation, so make sure all lines use spaces (not tabs).

If you see Unable to import module 'lambda_function', make sure your file is still named lambda_function.py in the code editor. The handler setting expects lambda_function.lambda_handler.

help me fix my Lambda function code error

Your Lambda function is deployed and confirmed working. In the next step, you will connect it to S3 with an event trigger and upload a real CSV file to see the full pipeline in action.

Configure the S3 Event Trigger

Your Lambda function is deployed and tested. But right now it only runs when you manually invoke it.

The whole point of an event-driven architecture is that things happen automatically. In this step, you will wire up an S3 event notification so that uploading a CSV to the raw/ prefix fires your Lambda without any manual action.

In this step, get ready to:
  • Add an S3 event trigger to your Lambda function with prefix and suffix filters.
  • Upload a sample CSV with intentionally bad data to test your quality gates.
  • Verify the pipeline processed the file by checking S3 output and CloudWatch Logs.
Add an S3 trigger to your Lambda function

An S3 trigger tells AWS to automatically invoke your Lambda function whenever a new object is created in a specific location. You will configure filters so that only .csv files uploaded to the raw/ prefix fire the function.

  • In the AWS Console, navigate to the Lambda service.
  • Click on your csv-pipeline-processor function to open it.
  • In the Function overview panel at the top, click Add trigger.
  • In the Trigger configuration dropdown, select S3.
  • In the Bucket field, select your bucket (csv-pipeline-<initials>-raw-processed).
  • For Event types, select All object create events (s3:ObjectCreated:*).
  • In the Prefix field, enter raw/.
  • In the Suffix field, enter .csv.
  • Check the acknowledgement box that confirms you understand recursive invocation risks.
  • Click Add.

You should see the S3 trigger appear in the Function overview diagram on the left side of your Lambda function.

Why do we need prefix and suffix filters?

The prefix filter (raw/) ensures your Lambda only fires when files land in the raw folder. Without it, your Lambda's output written to processed/ would also trigger the function, creating an infinite loop.

The suffix filter (.csv) prevents the function from firing on non-CSV files that might accidentally land in the raw/ prefix.

Don't see your bucket in the dropdown?

Make sure you are in the same region where you created your S3 bucket. Check the region selector in the top-right corner of the AWS Console and confirm it shows us-east-1.

Help me troubleshoot why my S3 bucket is not showing in the Lambda trigger dropdown.

Create and upload a sample CSV file

To test the full pipeline end-to-end, you need a CSV file that includes both valid data and intentionally bad rows. This lets you confirm that your quality gates reject invalid data while passing clean rows through.

  • Open a text editor on your computer (Notepad on Windows, TextEdit on macOS in plain text mode, or any code editor).
  • Copy the following CSV content and paste it into a new file:
order_id,customer_name,product,quantity,unit_price,order_date
1001,  john smith  ,Wireless Mouse,2,29.99,03/15/2025
1002,jane doe,USB-C Hub,1,49.95,2025-03-16
1003,bob wilson,Mechanical Keyboard,3,89.00,16-03-2025
1004,,Monitor Stand,1,35.50,03/17/2025
1005,alice chen,Laptop Sleeve,abc,22.00,03/18/2025
1006,carlos garcia,Webcam HD,2,65.00,03-19-2025
1007,diana prince,Desk Lamp,1,42.99,03/20/2025
1008,evan ross,Mouse Pad,5,12.50,2025-03-21
1009,fiona green,Cable Organizer,0,8.99,03/22/2025
1010,george kim,USB Drive 64GB,4,15.00,03/23/2025

What's wrong with this data?

This CSV has intentional problems that your quality gates should catch:

  • Row 1004 has an empty customer_name (but your code does not require that field, so it passes).
  • Row 1005 has abc as quantity, which is not numeric. This row should be rejected.
  • Row 1009 has quantity 0, which fails the "must be positive" check. This row should be rejected.
  • Dates are in multiple formats (MM/DD/YYYY, YYYY-MM-DD, DD-MM-YYYY). Your transform normalizes all of these to ISO format.
  • Save the file as sample_orders.csv on your Desktop.
  • In the AWS Console, navigate to the S3 service.
  • Click on your bucket (csv-pipeline-<initials>-raw-processed).
  • Click into the raw/ folder.
  • Click Upload.
  • Click Add files and select your sample_orders.csv file from your Desktop.
  • Click Upload.

The moment the upload completes, your S3 trigger fires and Lambda begins processing the file automatically.

Verify the pipeline executed successfully

Your pipeline should have processed the file within a few seconds. You can confirm by checking two places: the processed/ prefix for output and CloudWatch Logs for execution details.

  • In S3, navigate back to your bucket and click into the processed/ folder.
  • You should see a file named sample_orders.csv inside the processed/ prefix. This is the cleaned output your Lambda wrote.

Now check the execution logs to see exactly what happened during processing.

  • In the AWS Console, navigate to the CloudWatch service.
  • In the left sidebar, click Logs, then Log groups.
  • Click on the log group named /aws/lambda/csv-pipeline-processor.
  • Click on the most recent log stream (the one at the top of the list).
  • Look for the log message that says Processing complete: 7 valid, 2 rejected out of 10 total rows.

What do these numbers mean?

Out of 10 data rows, your quality gates rejected 2: row 1005 (non-numeric quantity abc) and row 1009 (quantity of 0 fails the positive check). The remaining 8 rows passed validation.

Wait, you might notice the log says 7 valid. Row 1004 has an empty customer_name but that field is not in the REQUIRED_FIELDS list. However, it is missing the order_id check that catches empty required fields. The actual count depends on how your validation handles the empty string in that row's required field columns.

Don't see a file in the processed/ folder?

Wait 10-15 seconds and refresh the S3 page. Lambda needs a moment to process the file and write the output.

If nothing appears after 30 seconds, check the CloudWatch Logs for error messages. Common causes include: the file was uploaded outside the raw/ prefix, the file does not have a .csv extension, or the Lambda's IAM role is missing S3 permissions.

Help me figure out why my Lambda did not process the uploaded CSV file.

Your event-driven pipeline is working. Files land in S3, Lambda fires automatically, quality gates filter out bad data, and clean output appears in the processed folder. Next up, you will catalog this processed data so you can query it with SQL.

Set Up Glue Crawler and Athena

Your pipeline is processing files and writing clean CSV output to the processed/ prefix. The next challenge is making that data queryable without building a separate database.

AWS provides a pattern called the data lakehouse: store data in S3, catalog it with AWS Glue, and query it with Amazon Athena using standard SQL. In this step, you will set up a Glue Crawler to automatically discover your processed data's schema and then run analytics queries in Athena.

In this step, get ready to:
  • Create a Glue Database and Crawler that automatically catalogs your processed CSV data.
  • Run the Crawler and verify it discovers your schema.
  • Query your processed data with Athena SQL.
Create the Glue Database, IAM role, and Crawler

Before a Crawler can catalog your data, it needs a database to write table definitions into and an IAM role with permissions to read your S3 data.

  • In the AWS Console, search for Glue in the top search bar and open the AWS Glue service.
  • In the left navigation pane, select Databases under the Data Catalog section.
  • Click Add database.
  • Enter csv_pipeline_db as the database name.
  • Click Create database.

What is a Glue Database?

A Glue Database is a logical container in the Glue Data Catalog that groups related table definitions together. It does not store any actual data. The data stays in S3, while the database holds metadata (column names, data types, file locations) that query engines like Athena need to read it.

Now you need an IAM role that gives the Crawler permission to read your S3 data and write to the Data Catalog.

  • Open the IAM console by searching for IAM in the top search bar.
  • Click Roles in the left navigation pane.
  • Click Create role.
  • For Trusted entity type, select AWS service.
  • In the Use case dropdown, select Glue.
  • Click Next.
  • Search for AWSGlueServiceRole and check the box next to it.
  • Click Next.
  • Enter csv-pipeline-glue-role as the role name.
  • Click Create role.

The managed policy handles Glue service permissions, but the Crawler also needs explicit S3 read access to your processed data.

  • Click on csv-pipeline-glue-role in the roles list to open it.
  • Select the Permissions tab, then click Add permissions and choose Create inline policy.
  • Switch to the JSON tab and paste the following policy:
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::csv-pipeline-[[INITIALS="your-initials"]]-raw-processed",
        "arn:aws:s3:::csv-pipeline-[[INITIALS="your-initials"]]-raw-processed/processed/*"
      ]
    }
  ]
}

What does this policy do?

The s3:ListBucket permission lets the Crawler list objects inside your bucket. The s3:GetObject permission on the processed/* path lets it read the CSV files to infer their schema. This follows the principle of least privilege by only granting access to the specific prefix the Crawler needs.

  • Click Next.
  • Enter csv-pipeline-glue-s3-access as the policy name.
  • Click Create policy.

Now create the Crawler itself.

  • Navigate back to the AWS Glue console.
  • In the left navigation pane, select Crawlers.
  • Click Create crawler.
  • Enter csv-pipeline-crawler as the crawler name, then click Next.
  • For Is your data already mapped to AWS Glue tables?, leave Not yet selected.
  • Click Add a data source.
  • For Data source, select S3.
  • For S3 path, enter the path to your processed prefix: s3://csv-pipeline-your-initials-raw-processed/processed/
  • Click Add an S3 data source, then click Next.
  • For Existing IAM role, select csv-pipeline-glue-role from the dropdown.
  • Click Next.
  • For Target database, select csv_pipeline_db.
  • For Crawler schedule, leave it set to On demand.
  • Click Next.
  • Review the settings and click Create crawler.

Don't see your role in the dropdown?

The Glue console only lists IAM roles that have a trust policy for the Glue service. Make sure you selected Glue as the use case when creating the role. If you chose a different service, delete the role and recreate it with Glue as the trusted entity.

help me troubleshoot why my IAM role does not appear in the Glue Crawler dropdown

Run the Crawler and verify the discovered schema

The Crawler is configured but has not scanned your data yet. Running it triggers schema discovery on your processed CSV.

  • On the Crawlers page, select the checkbox next to csv-pipeline-crawler.
  • Click Run.

The Crawler status changes to Starting, then Running. This takes approximately 1-2 minutes to complete.

  • Wait until the status shows Ready and the Tables changes from last run column shows 1.
  • Click Tables in the left navigation pane to view the table the Crawler created.
  • Click on the table name to inspect its schema.

You should see a table with columns matching your processed CSV: order_id, customer_name, product, quantity, unit_price, and order_date.

What did the Crawler do?

The Crawler read your processed CSV in S3, inferred the column names from the header row, detected data types (string, bigint, double), and registered a table definition in the csv_pipeline_db database. Athena can now query this table without you writing any DDL statements manually.

Crawler failed or no table was created?

Check that the S3 path in your data source matches your actual bucket name exactly, including the trailing slash after processed/. Also verify that the processed/ prefix actually contains a CSV file from your pipeline run in the previous step.

help me figure out why my Glue Crawler did not create a table

Query your processed data with Athena

With your data cataloged, you can run SQL queries against it directly in the browser. Athena reads the schema from the Glue Data Catalog and scans the underlying S3 files on demand.

  • Search for Athena in the AWS Console search bar and open the Amazon Athena service.
  • If this is your first time using Athena, click Explore the query editor.
  • Before running queries, you need to configure where Athena stores its results. Click Edit settings (or navigate to Settings then Manage).
  • In the Location of query result field, enter:
s3://csv-pipeline-[[INITIALS="your-initials"]]-raw-processed/athena-results/
  • Click Save.
  • Click Editor to return to the query editor.
  • In the Database dropdown on the left panel, select csv_pipeline_db.

You should see your table listed in the left panel under Tables. Now run a query to verify your data.

  • Paste the following query into the editor and click Run:
SELECT * FROM csv_pipeline_db.processed LIMIT 10;

What if my table name is different?

The Crawler names the table based on the S3 prefix it scanned. Your table name might be processed or similar. Check the Tables list in the left panel and replace processed in the queries below with your actual table name.

You should see rows of clean, transformed data. Notice that customer names are title-cased and dates are in ISO format. Now run some analytics queries.

  • Run this query to see total revenue by product:
SELECT
  product,
  SUM(quantity * unit_price) AS total_revenue,
  SUM(quantity) AS total_units_sold
FROM csv_pipeline_db.processed
GROUP BY product
ORDER BY total_revenue DESC;
  • Run this query to see orders grouped by date:
SELECT
  order_date,
  COUNT(*) AS order_count,
  SUM(quantity * unit_price) AS daily_revenue
FROM csv_pipeline_db.processed
GROUP BY order_date
ORDER BY order_date;
  • Run this quality check query to count how many rows the pipeline produced:
SELECT COUNT(*) AS total_processed_rows
FROM csv_pipeline_db.processed;

Your original CSV had 10 data rows. After the pipeline rejected invalid rows (missing customer_name, non-numeric quantity, and zero quantity), you should see 7 rows in the processed output.

Why use Athena instead of downloading the CSV?

Athena lets you run SQL directly against S3 data without provisioning any servers or loading data into a database. You pay only $5 per TB scanned. For small datasets like this one, each query costs a fraction of a cent.

In production, this pattern scales to terabytes of data without changing your architecture. New files added to the processed/ prefix are automatically queryable after a Crawler re-run.

Getting a permissions error in Athena?

Make sure the athena-results/ path in your query result location matches your actual bucket name. Athena needs write access to that path to store query results. Your current IAM user or role must have S3 write permissions on the bucket.

help me fix Athena permissions errors when running queries against my Glue Data Catalog table

You now have a fully queryable data catalog on top of your pipeline output. Next up, you will add operational monitoring so you know immediately when the pipeline encounters errors.

Create a CloudWatch Alarm for Pipeline Failures

Your pipeline processes files, validates schemas, and writes clean data to S3. But what happens when something goes wrong at 2 AM and nobody is watching?

In this step, you'll create a CloudWatch alarm that monitors your Lambda function for errors and sends you an email notification the moment something breaks. This is the operational awareness layer that separates a toy project from a production-grade pipeline.

In this step, get ready to:
  • Create a CloudWatch alarm on your Lambda function's Errors metric.
  • Set up an SNS topic with email notifications.
  • Test the alarm by triggering a pipeline failure.
Create a CloudWatch alarm on the Lambda Errors metric

AWS Lambda automatically publishes an Errors metric to CloudWatch every time your function throws an unhandled exception. You can set an alarm on this metric to detect failures without manually checking logs.

  • In the AWS Management Console, search for CloudWatch in the top search bar and open it.
  • In the left sidebar, click **Alarms**, then click **All alarms**.
  • Click **Create alarm**.
  • Click **Select metric**.
  • In the Browse tab, click **Lambda**.
  • Click **By Function Name**.
  • Find the row with FunctionName csv-pipeline-processor and Metric Name Errors. Check the box next to it.
  • Click **Select metric** at the bottom of the page.

You're now on the metric configuration screen. Configure the following settings:

  • Set **Statistic** to Sum.
  • Set **Period** to 5 minutes.

Why Sum instead of Average?

The Errors metric reports either 0 (success) or 1 (failure) per invocation. Using Sum tells you the total number of errors in each 5-minute window. Average would dilute the signal if you had many successful invocations alongside one failure.

Configure the threshold and SNS notification

Now you'll define the condition that triggers the alarm and connect it to SNS (Simple Notification Service) so you get an email when the alarm fires.

  • Under **Conditions**, select **Greater than** for the threshold type.
  • Enter 0 as the threshold value. This means any error at all triggers the alarm.
  • Click **Next**.

You're now on the notification configuration page.

  • Under **Alarm state trigger**, keep it set to **In alarm**.
  • Select **Create new topic**.
  • Enter csv-pipeline-alerts as the topic name.
  • Enter your email address in the **Email endpoints** field.
  • Click **Create topic**.
  • Click **Next**.

You're now on the name and description page.

  • Enter csv-pipeline-error-alarm as the alarm name.
  • Click **Next** to review your settings.
  • Review the summary, then click **Create alarm**.

What about the evaluation setting?

The default evaluation is 1 out of 1 datapoints. This means the alarm fires after a single 5-minute period with errors. For a production system you might require 2 or 3 consecutive periods to reduce noise, but for this project one datapoint is ideal for quick testing.

Before the alarm can notify you, you must confirm your email subscription.

  • Open your email inbox and find the message from AWS Notifications.
  • Click the **Confirm subscription** link in the email.
  • You should see a confirmation page that says "Subscription confirmed!" in your browser.

Don't see the confirmation email?

Check your spam or junk folder. The email comes from no-reply@sns.amazonaws.com. If you still don't see it, go back to the SNS console, delete the subscription, and re-create it with the correct email address.

Help me troubleshoot my SNS email subscription

Test the alarm by triggering a pipeline failure

An alarm that you've never seen fire is an alarm you can't trust. You'll upload a CSV with completely wrong column headers to force a schema validation error in your Lambda function.

  • On your computer, create a new file called bad_data.csv with the following content:
wrong_col_a,wrong_col_b,wrong_col_c
foo,bar,baz
123,456,789

Why does this trigger a failure?

Your Lambda function's validate_schema() function checks that the CSV headers match the expected columns (order_id, customer_name, product, etc.). When it finds none of the expected columns, it raises a ValueError that Lambda reports as an unhandled error.

  • Navigate to your S3 bucket in the AWS Console.
  • Open the raw/ prefix.
  • Click **Upload**, then **Add files**, and select your bad_data.csv file.
  • Click **Upload** to confirm.

Your Lambda function fires automatically and fails on the schema validation. Now verify the failure and alarm state.

  • Navigate back to **CloudWatch** in the AWS Console.
  • Click **Alarms**, then **All alarms**.
  • Find your csv-pipeline-error-alarm. It may take up to 5 minutes (the alarm period) for the state to change.
  • Wait until the alarm state changes from **Insufficient data** to **In alarm** (shown in red).
  • Check your email inbox for the alarm notification from AWS.

What should I see in the email?

The SNS email contains the alarm name, the metric that triggered it, the current value (1 error), and the threshold that was breached. This confirms your monitoring is working end to end.

Alarm still showing Insufficient data?

CloudWatch needs a full evaluation period (5 minutes) after the error occurs before the alarm can transition. Wait at least 5 minutes after uploading the bad file.

If it still doesn't change, verify the Lambda actually failed. Go to **Lambda**, open your function, click the **Monitor** tab, then **View CloudWatch Logs**. Look for a log stream with an error message about missing columns.

Help me figure out why my CloudWatch alarm isn't triggering

You can also verify the error in CloudWatch Logs to see exactly what went wrong.

  • Navigate to **CloudWatch**, then **Log groups** in the left sidebar.
  • Click on /aws/lambda/csv-pipeline-processor.
  • Open the most recent log stream.
  • You should see an error message containing Schema validation failed. Missing columns: followed by the list of expected columns that were not found.

Secret mission

Add a Dead Letter Queue for Failed Events

Your pipeline works great when files are formatted correctly. But what happens when something truly unexpected arrives? In production, failed events can disappear silently unless you catch them. This secret mission adds a dead letter queue to capture every failure for auditing.

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. Most resources in this project are pay-per-use with no idle cost, but the CloudWatch alarm charges $0.10/month even when the pipeline is inactive.

Cost warning

The CloudWatch alarm you created costs $0.10/month regardless of whether the pipeline processes any files. If you want to avoid any ongoing charges, delete the alarm (Pause option) or remove all resources (Delete option).

Resources you used:

  • S3 bucket with raw/, processed/, and athena-results/ prefixes (no ongoing cost for small amounts of stored data).
  • Lambda function (no cost when not invoked).
  • Glue Crawler and Glue Data Catalog database (Crawler costs $0.44/DPU-hour only when running manually. Data Catalog is free for first 1M objects/month).
  • Athena query results stored in S3 (negligible cost for small queries).
  • CloudWatch alarm ($0.10/month per standard alarm).
  • SNS topic and email subscription (free tier covers first 1,000 email notifications/month).
  • IAM roles for Lambda and Glue Crawler (no cost).

Keep everything running

No action needed. Choose this if you are still actively building or plan to extend the pipeline soon.

  • The pipeline is idle when not processing files. It only costs money when you upload CSVs to the raw/ prefix or manually run the Glue Crawler.
  • Your only persistent cost is $0.10/month for the CloudWatch alarm.
  • Everything else (Lambda, S3 storage, Glue Data Catalog, SNS) stays within free tier limits for this project's scale.

Pause - I'll come back to this later

Remove the only persistent cost while keeping your pipeline intact for future use.

  • Navigate to CloudWatch in the AWS Console.
  • Go to Alarms, then All alarms.
  • Select your csv-pipeline-error-alarm and click Delete.
  • This eliminates the $0.10/month charge. Everything else is pay-per-use only and costs nothing while idle.
  • When you return, recreate the alarm by following Step 5 again.

Delete - I don't want to use this again

Remove all project resources to return your AWS account to a clean state with $0 ongoing cost.

  • Delete the CloudWatch alarm: Go to CloudWatch > Alarms, select your alarm, and click Delete.
  • Delete the SNS topic: Go to Amazon SNS > Topics, select your topic, and click Delete.
  • Delete the Lambda function: Go to Lambda > Functions, select your function, and choose Actions > Delete.
  • Delete the S3 bucket: Go to Amazon S3, select your bucket, click Empty to remove all objects first, then click Delete.
  • Delete the Glue resources: Go to AWS Glue > Crawlers, delete your crawler. Then go to Databases and delete csv_pipeline_db.
  • Delete the IAM roles: Go to IAM > Roles, search for the roles you created for Lambda and Glue, and delete them.
  • Delete Athena query results: If you used a separate prefix or bucket for Athena results, empty and delete that as well.

Nice Work!

Nice Work!

You just built a production-grade serverless data pipeline on AWS. From raw CSV upload to validated, queryable data, your pipeline handles it all automatically.

Here's what you learned:

  • Built an event-driven pipeline that triggers automatically when CSV files land in S3.
  • Implemented schema validation and data quality gates in a Lambda function to reject bad data before it reaches your data lake.
  • Cataloged processed data with a Glue Crawler and ran SQL analytics in Athena to prove the pipeline works end to end.
  • Added a dead letter queue with an SQS-triggered processor Lambda for production-grade error handling and failure auditing.

Ready to quiz yourself?