Batch Data Pipeline with Medallion Architecture

Build an event-driven S3 to Snowflake pipeline using bronze/silver/gold layers.

Introduction

30 Second Summary

Every business generates raw data constantly. Sales orders, user signups, sensor readings. But raw data is messy and useless until someone cleans it, organizes it, and puts it where analysts can actually query it.

In this project, you will build a fully automated batch data pipeline that takes raw sales data, processes it through three layers (bronze, silver, gold), and loads the final result into a cloud data warehouse for querying.

What You'll Build

You run a simple Python script, and within seconds your data flows from a raw CSV upload all the way through automated cleaning, aggregation, and into a Snowflake warehouse table you can query immediately.

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

  • A data generator script you can run anytime to produce sales data and watch it appear in S3 within seconds.
  • An event-driven Lambda function that automatically detects new files and transforms raw data through cleaned and aggregated layers using the medallion architecture pattern.
  • A live Snowpipe auto-ingest connection that loads your aggregated data into Snowflake so you can query it moments after uploading.
  • Secret Mission: add data quality validation that quarantines bad records to a separate folder, keeping only clean data flowing through your pipeline.

Are there any prerequisites?

This project assumes you already have an AWS account and a Snowflake account (a free trial works perfectly). You also need Python 3.10 or higher installed on your machine and the AWS CLI configured with your credentials.

If you do not have these yet, set them up before starting Step 1.

Before We Start

Before we dive in, take a moment to lock in what you're about to build and why it matters.

Set Up Your S3 Bucket and IAM Roles

Before any data can flow through your pipeline, you need infrastructure in place to receive it.

In this step, you will create an S3 bucket with three folder prefixes (bronze, silver, gold) and an IAM role that gives Lambda permission to read and write to that bucket.

In this step, get ready to:
  • Create an S3 bucket with bronze/, silver/, and gold/ folder prefixes.
  • Create an IAM role for Lambda with S3 and CloudWatch Logs permissions.
  • Verify your AWS CLI credentials work by listing your S3 buckets.
Verify your AWS CLI credentials

Your AWS CLI needs working credentials before you can create any resources.

  • Verify your AWS CLI credentials work by listing your S3 buckets:
aws s3 ls

You should see a list of any existing buckets in your account, or an empty response if you have none yet.

Seeing an access denied or credentials error?

Make sure you have run aws configure and entered a valid Access Key ID and Secret Access Key. Check that the IAM user associated with your credentials has permission to list S3 buckets.

Help me fix my AWS CLI credentials

Create your S3 bucket with medallion prefixes

The medallion architecture organises data into layers. Each layer lives under a separate prefix (folder) in a single S3 bucket.

  • Create your S3 bucket by running this command:
aws s3 mb s3://[[BUCKET="your-bucket-name"]]

You should see a confirmation message that your bucket was created.

Why does my bucket name need to be globally unique?

S3 bucket names are shared across all AWS accounts worldwide. Choose a name that includes your initials or a random string to avoid collisions.

  • Create the three medallion layer prefixes by running these commands:
aws s3api put-object --bucket [[BUCKET="your-bucket-name"]] --key bronze/
aws s3api put-object --bucket [[BUCKET="your-bucket-name"]] --key silver/
aws s3api put-object --bucket [[BUCKET="your-bucket-name"]] --key gold/

What do these commands do?

S3 does not have real folders. A "folder" is just a zero-byte object whose key ends with a forward slash. These three commands create the bronze/, silver/, and gold/ prefixes that your pipeline will use as processing layers.

  • Confirm the prefixes exist by running:
aws s3 ls s3://[[BUCKET="your-bucket-name"]]/

You should see three entries prefixed with PRE: bronze/, silver/, and gold/.

Not seeing all three prefixes?

Make sure you replaced the bucket name placeholder in each command with your actual bucket name. Run the aws s3 ls command again after confirming each put-object command completed without errors.

Help me troubleshoot my S3 bucket prefixes

Create the IAM role for Lambda

Lambda needs an execution role that grants it permission to access your S3 bucket and write logs to CloudWatch. You will create a trust policy first, then create the role, then attach permissions.

  • Create a trust policy file that allows Lambda to assume the role by running:
cat > trust-policy.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}
EOF

What is a trust policy?

A trust policy tells AWS which service is allowed to "assume" (use) this role. By specifying lambda.amazonaws.com as the principal, you are saying only Lambda functions can use this role's permissions.

  • Create the IAM role using the trust policy by running:
aws iam create-role \
  --role-name medallion-lambda-role \
  --assume-role-policy-document file://trust-policy.json

You should see JSON output containing the role details, including an Arn field. Copy and save this ARN somewhere safe. You will need it when deploying your Lambda function in a later step.

  • Attach the S3 full access policy to the role by running:
aws iam attach-role-policy \
  --role-name medallion-lambda-role \
  --policy-arn arn:aws:iam::aws:policy/AmazonS3FullAccess
  • Attach the CloudWatch Logs full access policy to the role by running:
aws iam attach-role-policy \
  --role-name medallion-lambda-role \
  --policy-arn arn:aws:iam::aws:policy/CloudWatchLogsFullAccess

What do these policies give Lambda?

AmazonS3FullAccess lets your Lambda function read files from the bronze/ prefix and write processed files to silver/ and gold/.

CloudWatchLogsFullAccess lets Lambda write execution logs so you can debug issues later.

Before you run the verification, what two policy names do you expect to see attached to your role?

  • Verify both policies are attached by running:
aws iam list-attached-role-policies --role-name medallion-lambda-role

You should see both AmazonS3FullAccess and CloudWatchLogsFullAccess listed in the output.

Getting a "NoSuchEntity" error?

Double-check that the create-role command completed successfully before running attach-role-policy. Make sure you used medallion-lambda-role consistently across all commands.

Help me fix my IAM role creation

Your infrastructure is ready. Next up, you will write a simple data generator script that uploads fake sales data into the bronze/ layer.

Build and Run the Data Generator

Your S3 bucket is set up with its bronze, silver, and gold prefixes, and your IAM role is ready to go.

Now you need actual data flowing into the pipeline. Without files landing in the bronze layer, there's nothing for the rest of the pipeline to process. In this step, you'll write a simple Python script that generates fake sales data and uploads it straight to S3.

In this step, get ready to:
  • Write a data generator script that creates 10 rows of fake sales data and uploads them to S3.
  • Run the script and confirm the CSV file appears in your S3 bronze folder.
Create the data generator script

The data generator is intentionally tiny. It uses only Python standard library modules plus boto3 for the S3 upload.

  • Create a new file called data_generator.py in the project folder you're working from.
  • Add the following imports and configuration to the top of the file:
import csv
import random
import boto3
from datetime import datetime

# Your S3 bucket name and product list for generating fake data
BUCKET = "[[BUCKET_NAME=\"your-bucket-name\"]]"
PRODUCTS = ["Laptop", "Mouse", "Keyboard", "Monitor", "Headset"]

# Generate 10 rows of random sales data
rows = []
for i in range(10):
    rows.append({
        "order_id": f"ORD-{random.randint(1000,9999)}",
        "product": random.choice(PRODUCTS),
        "quantity": random.randint(1, 5),
        "price": round(random.uniform(10.0, 999.99), 2),
        "timestamp": datetime.now().isoformat()
    })

What does this code do?

BUCKET stores your S3 bucket name so the script knows where to upload. PRODUCTS is a list of fake product names the generator picks from randomly.

The loop creates 10 dictionaries, each representing a sales order with a random order ID, product, quantity, price, and the current timestamp.

  • Add the following code below the loop to write the data to a CSV file and upload it to S3:
# Create a timestamped CSV filename
filename = f"sales_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
with open(filename, "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["order_id", "product", "quantity", "price", "timestamp"])
    writer.writeheader()
    writer.writerows(rows)

# Upload the CSV to the bronze layer in S3
s3 = boto3.client("s3")
s3.upload_file(filename, BUCKET, f"bronze/{filename}")
print(f"Uploaded {filename} to s3://{BUCKET}/bronze/{filename}")

What does this code do?

The filename includes a timestamp (e.g. sales_20240101_120000.csv) so each run produces a unique file without overwriting previous uploads.

csv.DictWriter writes the list of dictionaries as a properly formatted CSV with headers. Then boto3 uploads the local file to the bronze/ prefix in your S3 bucket.

  • Save data_generator.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before running it.

ⓧ I'd like to double check the full code

Here's the complete data_generator.py for comparison:

import csv
import random
import boto3
from datetime import datetime

BUCKET = "[[BUCKET_NAME=\"your-bucket-name\"]]"
PRODUCTS = ["Laptop", "Mouse", "Keyboard", "Monitor", "Headset"]

rows = []
for i in range(10):
    rows.append({
        "order_id": f"ORD-{random.randint(1000,9999)}",
        "product": random.choice(PRODUCTS),
        "quantity": random.randint(1, 5),
        "price": round(random.uniform(10.0, 999.99), 2),
        "timestamp": datetime.now().isoformat()
    })

filename = f"sales_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
with open(filename, "w", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["order_id", "product", "quantity", "price", "timestamp"])
    writer.writeheader()
    writer.writerows(rows)

s3 = boto3.client("s3")
s3.upload_file(filename, BUCKET, f"bronze/{filename}")
print(f"Uploaded {filename} to s3://{BUCKET}/bronze/{filename}")
Run the script and verify the upload

The script is complete. Time to send your first batch of data into the pipeline.

Before you run this, what do you think will happen? Consider what output the script prints and where the file will end up in S3.

  • Run the data generator by executing this command:
python3 data_generator.py

You should see output like this:

Uploaded sales_20240101_120000.csv to s3://your-bucket-name/bronze/sales_20240101_120000.csv

What should I see?

The script prints a confirmation message showing the exact filename and S3 path. Your timestamp will be different from the example since it reflects the current date and time.

  • Confirm the file landed in S3 by running:
aws s3 ls s3://[[BUCKET_NAME="your-bucket-name"]]/bronze/

You should see your timestamped CSV file listed in the output, for example:

2024-01-01 12:00:00        456 sales_20240101_120000.csv

Don't see your file in S3?

  • Check that the BUCKET value in your script matches your actual S3 bucket name exactly.
  • Make sure your AWS CLI credentials are still configured correctly by running aws sts get-caller-identity.
  • If you see a NoSuchBucket error, double check the bucket name for typos.

Why isn't my data generator uploading to S3?

Your first batch of raw data is now sitting in the bronze layer. Next up, you'll build the Lambda function that automatically processes this raw data into cleaned and aggregated layers the moment it lands.

Create the Lambda Processor and Wire the S3 Trigger

You have raw sales data landing in S3, but right now it just sits there. Nobody wants to manually log in and run a script every time a file arrives.

In this step, you will write a Lambda function that automatically fires when data lands in the bronze/ prefix. It cleans the data (silver layer) and aggregates it by product (gold layer). This is event-driven architecture in action.

In this step, get ready to:
  • Write a Lambda function that reads, cleans, and aggregates CSV data across medallion layers.
  • Deploy the function to AWS using the CLI and wire an S3 event trigger.
  • Run the full pipeline end-to-end and confirm silver/ and gold/ files appear automatically.
Write the Lambda function code

The Lambda function has two jobs: clean the raw data (silver) and aggregate it by product (gold). You will write it in a single file called lambda_function.py.

  • Create a new file called lambda_function.py in the same folder as your data_generator.py.
  • Add the imports and the first half of the handler function by pasting this code:
import csv
import boto3
import io
from collections import defaultdict

s3 = boto3.client("s3")

def lambda_handler(event, context):
    # Extract bucket name and file key from the S3 event
    bucket = event["Records"][0]["s3"]["bucket"]["name"]
    key = event["Records"][0]["s3"]["object"]["key"]

    # Read the 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))

    # Clean rows: strip whitespace, validate numerics, drop bad data
    clean_rows = []
    for row in reader:
        if not row["product"].strip():
            continue
        try:
            row["quantity"] = int(row["quantity"])
            row["price"] = float(row["price"])
        except (ValueError, TypeError):
            continue
        if row["quantity"] <= 0 or row["price"] <= 0:
            continue
        clean_rows.append(row)

    if not clean_rows:
        return {"statusCode": 200, "body": "No valid rows to process"}

What does this code do?

The function receives an S3 event payload containing the bucket name and the object key (file path) of the newly uploaded file.

It downloads the CSV, then loops through each row applying three validation checks: the product name is not empty, quantity is a valid positive integer, and price is a valid positive float. Any row that fails gets silently dropped. This is the bronze-to-silver cleaning step.

  • Now add the second half of the function directly below the code you just pasted. This writes the silver and gold outputs:
    filename = key.split("/")[-1]

    # Write cleaned data to silver/
    silver_output = io.StringIO()
    writer = csv.DictWriter(silver_output, fieldnames=["order_id", "product", "quantity", "price", "timestamp"])
    writer.writeheader()
    writer.writerows(clean_rows)
    s3.put_object(Bucket=bucket, Key=f"silver/{filename}", Body=silver_output.getvalue())

    # Aggregate by product for gold/
    aggregated = defaultdict(lambda: {"total_quantity": 0, "total_revenue": 0.0})
    for row in clean_rows:
        product = row["product"]
        aggregated[product]["total_quantity"] += row["quantity"]
        aggregated[product]["total_revenue"] += row["quantity"] * row["price"]

    # Write aggregated data to gold/
    gold_output = io.StringIO()
    gold_writer = csv.DictWriter(gold_output, fieldnames=["product", "total_quantity", "total_revenue"])
    gold_writer.writeheader()
    for product, metrics in aggregated.items():
        gold_writer.writerow({
            "product": product,
            "total_quantity": metrics["total_quantity"],
            "total_revenue": round(metrics["total_revenue"], 2)
        })
    s3.put_object(Bucket=bucket, Key=f"gold/{filename}", Body=gold_output.getvalue())

    return {"statusCode": 200, "body": f"Processed {len(clean_rows)} rows into silver and gold"}

What does the silver and gold logic do?

The silver step writes all validated rows back to S3 under the silver/ prefix. This is your "cleaned" dataset.

The gold step groups rows by product name, sums up total quantity and total revenue (quantity multiplied by price), and writes one summary row per product to gold/. This is your analytics-ready dataset.

  • Save lambda_function.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved lambda_function.py and it lives in the same folder as data_generator.py.

ⓧ I'd like to double check the full code

Your lambda_function.py should contain these two chunks in order. Scroll up and confirm you have the imports and cleaning logic first, followed by the silver/gold writing logic immediately after.

Deploy the Lambda function using the AWS CLI

Lambda expects your code packaged as a .zip file. You will zip your Python file and then create the function in one command.

  • Zip your Lambda code by running this command:
zip function.zip lambda_function.py
  • Confirm the zip file was created by running:
ls function.zip

You should see function.zip listed in the output.

  • Deploy the function to AWS by running this command:
aws lambda create-function \
  --function-name medallion-processor \
  --runtime python3.13 \
  --handler lambda_function.lambda_handler \
  --role arn:aws:iam::[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]]:role/medallion-lambda-role \
  --zip-file fileb://function.zip

What does this command do?

  • --function-name medallion-processor names the function in AWS.
  • --runtime python3.13 tells Lambda to use the Python 3.13 runtime.
  • --handler lambda_function.lambda_handler tells Lambda which file and function to call (file lambda_function.py, function lambda_handler).
  • --role attaches the IAM role you created earlier so Lambda can access S3 and write logs.
  • --zip-file fileb://function.zip uploads your code package.

You should see a JSON response containing FunctionName, Runtime, and State fields confirming the function was created.

Seeing an error about the role?

Make sure you replaced the account ID placeholder with your actual 12-digit AWS account ID. You can find it by running aws sts get-caller-identity.

If you see a message about the role not being able to be assumed, wait 10 seconds and try again. IAM role propagation can take a moment.

Need more help? Ask: Why is my Lambda create-function failing with a role error?

Configure the S3 event notification to trigger Lambda

The function exists, but nothing tells S3 to call it. You need two things: a resource-based permission allowing S3 to invoke the function, and a notification configuration on the bucket.

  • Grant S3 permission to invoke your Lambda function by running:
aws lambda add-permission \
  --function-name medallion-processor \
  --statement-id s3-trigger \
  --action lambda:InvokeFunction \
  --principal s3.amazonaws.com \
  --source-arn arn:aws:s3:::[[BUCKET_NAME="your-bucket-name"]]
  • Create a file called notification.json with the S3 event notification configuration. Paste this into the file:
{
  "LambdaFunctionConfigurations": [
    {
      "Id": "BronzeTrigger",
      "LambdaFunctionArn": "arn:aws:lambda:[[AWS_REGION="your-region"]]:[[AWS_ACCOUNT_ID="your 12-digit AWS account ID"]]:function:medallion-processor",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {
              "Name": "prefix",
              "Value": "bronze/"
            }
          ]
        }
      }
    }
  ]
}

Why filter on the bronze/ prefix?

The Lambda function writes to silver/ and gold/. If the trigger fired on all object creations, your function would trigger itself in an infinite loop. Filtering to only bronze/ ensures the function only runs when new raw data arrives.

  • Apply the notification configuration to your bucket by running:
aws s3api put-bucket-notification-configuration \
  --bucket [[BUCKET_NAME="your-bucket-name"]] \
  --notification-configuration file://notification.json

This command produces no output on success. The trigger is now live.

Getting an "Unable to validate" error?

This usually means the Lambda permission was not applied correctly. Double check that the --source-arn in your add-permission command matches your bucket name exactly (arn:aws:s3:::your-bucket-name).

Also confirm the Lambda function ARN in notification.json matches your region and account ID. You can verify the function ARN by running aws lambda get-function --function-name medallion-processor.

Still stuck? Ask: Why is my S3 notification configuration failing to validate?

Test the full pipeline end-to-end

Everything is wired. Time to prove it works by generating new data and watching the pipeline process it automatically.

  • Before you run this, predict: do you think new files will appear in silver/ and gold/ within a few seconds of running the data generator?
  • Trigger the pipeline by running your data generator:
python data_generator.py

Wait about 5-10 seconds for Lambda to process the file.

  • Check that Lambda created the silver file by running:
aws s3 ls s3://[[BUCKET_NAME="your-bucket-name"]]/silver/
  • Check that Lambda created the gold file by running:
aws s3 ls s3://[[BUCKET_NAME="your-bucket-name"]]/gold/

You should see a new CSV file listed under both silver/ and gold/ with a timestamp matching the file you just uploaded to bronze/. That means your Lambda automatically processed the raw data through both transformation layers.

What should I see?

Each command shows a file like sales_20260731_143022.csv (with your actual timestamp). The silver file contains your cleaned rows. The gold file contains one row per product with aggregated totals.

Not seeing files in silver/ or gold/?

Check CloudWatch Logs for errors. Run aws logs describe-log-groups and look for a group named /aws/lambda/medallion-processor. If the group exists, the function was invoked. Check the latest log stream for error details.

If no log group exists, the S3 trigger did not fire. Verify your notification configuration is applied by running aws s3api get-bucket-notification-configuration --bucket your-bucket-name.

Need help debugging? Ask: My Lambda function isn't being triggered by the S3 upload. How do I debug this?

Your event-driven processing layer is live. Every time you run the data generator, Lambda automatically picks up the file and processes it through silver and gold. Next up, you will connect Snowflake to automatically ingest the gold-layer data for querying.

Connect Snowflake with Snowpipe Auto-Ingest

Your gold-layer data is sitting in S3, aggregated and ready for analysis. But your analytics team queries data in Snowflake, not S3.

Snowpipe auto-ingest solves this last-mile problem. It uses S3 event notifications to detect new files in the gold/ prefix and automatically loads them into a Snowflake table. No manual COPY commands, no scheduling. The moment Lambda writes a gold file, Snowpipe picks it up.

In this step, get ready to:
  • Create a Snowflake storage integration, external stage, table, and pipe with AUTO_INGEST = TRUE.
  • Configure the S3 event notification on the gold/ prefix to send to Snowflake's SQS queue.
  • Run the full end-to-end pipeline and query your data in Snowflake.
Create the Snowflake objects

Snowflake needs several objects to read from S3: a storage integration (the trust link to AWS), an external stage (the S3 path), a target table, and a pipe that auto-loads new files.

  • Open your Snowflake account in a browser and sign in.
  • Click Worksheets in the left sidebar, then click the + button to create a new SQL worksheet.
  • Make sure your role is set to ACCOUNTADMIN in the top-left role selector. Only account administrators can create storage integrations.
  • Create the database, schema, and storage integration by running this SQL in your worksheet:
-- Create the database and schema for our pipeline
CREATE DATABASE IF NOT EXISTS pipeline_db;
USE DATABASE pipeline_db;
CREATE SCHEMA IF NOT EXISTS public;
USE SCHEMA public;

-- Create the storage integration linking Snowflake to your S3 gold/ prefix
CREATE OR REPLACE STORAGE INTEGRATION s3_pipeline_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::[[ACCOUNT_ID="your 12-digit AWS account ID"]]:role/snowflake-access-role'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('s3://[[BUCKET_NAME="your-bucket-name"]]/gold/');

What does this SQL do?

The storage integration creates a trust relationship between Snowflake and your AWS account. It tells Snowflake which IAM role to assume and which S3 locations it can access.

The STORAGE_AWS_ROLE_ARN references a role you will create in the next chunk. Snowflake does not validate the role until you actually use the stage, so the order works.

  • You should see a "Statement executed successfully" message for each command.
  • Now retrieve Snowflake's IAM user ARN and external ID by running:
-- Get Snowflake's IAM user ARN and external ID for the trust policy
DESCRIBE INTEGRATION s3_pipeline_int;

You will see a table of properties. Record the values from two rows:

  • STORAGE_AWS_IAM_USER_ARN (looks like arn:aws:iam::123456789001:user/abc1-b-self1234).
  • STORAGE_AWS_EXTERNAL_ID (looks like MYACCOUNT_SFCRole=2_a123456/s0aBCDEfGHIJklmNoPq=).

Copy these values now

You will paste them into an AWS IAM trust policy in the next chunk. Keep them somewhere accessible.

Now switch to your terminal to create the IAM role that Snowflake will assume.

  • Create a file called trust-policy.json in your project folder with this content:
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "[[SNOWFLAKE_IAM_ARN="STORAGE_AWS_IAM_USER_ARN value from DESCRIBE output"]]"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": {
          "sts:ExternalId": "[[SNOWFLAKE_EXTERNAL_ID="STORAGE_AWS_EXTERNAL_ID value from DESCRIBE output"]]"
        }
      }
    }
  ]
}
  • Create the IAM role using your trust policy by running:
aws iam create-role \
  --role-name snowflake-access-role \
  --assume-role-policy-document file://trust-policy.json

You should see JSON output containing the role ARN and creation details.

  • Create a file called s3-read-policy.json to grant the role read access to your gold/ prefix:
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:GetObjectVersion"],
      "Resource": "arn:aws:s3:::[[BUCKET_NAME="your-bucket-name"]]/gold/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket", "s3:GetBucketLocation"],
      "Resource": "arn:aws:s3:::[[BUCKET_NAME="your-bucket-name"]]"
    }
  ]
}
  • Attach this policy to the role by running:
aws iam put-role-policy \
  --role-name snowflake-access-role \
  --policy-name s3-gold-read \
  --policy-document file://s3-read-policy.json

No output means success. The role now allows Snowflake to read from your gold/ prefix.

Getting an access denied error?

Make sure your AWS CLI credentials have permission to create IAM roles. Check that trust-policy.json is valid JSON with no trailing commas.

help me troubleshoot IAM role creation errors for my Snowflake storage integration

Switch back to your Snowflake worksheet to create the remaining objects.

  • Create the external stage, target table, and pipe by running this SQL:
-- Create the external stage pointing at your gold/ prefix
CREATE OR REPLACE STAGE gold_stage
  STORAGE_INTEGRATION = s3_pipeline_int
  URL = 's3://[[BUCKET_NAME="your-bucket-name"]]/gold/'
  FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '"' SKIP_HEADER = 1);

-- Create the target table for aggregated sales data
CREATE OR REPLACE TABLE gold_sales (
  product STRING,
  total_quantity INTEGER,
  total_revenue FLOAT
);

-- Create the pipe with AUTO_INGEST to load new files automatically
CREATE OR REPLACE PIPE gold_pipe
  AUTO_INGEST = TRUE
  AS
    COPY INTO gold_sales
      FROM @gold_stage
      FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '"' SKIP_HEADER = 1);

What does AUTO_INGEST do?

When AUTO_INGEST = TRUE, Snowflake creates an SQS queue behind the scenes. Any S3 event notification sent to that queue triggers Snowpipe to load the new file into gold_sales automatically.

  • Retrieve the SQS queue ARN that Snowpipe created by running:
-- Get the SQS queue ARN for S3 event notification configuration
SHOW PIPES;

In the results table, find the row for gold_pipe. Copy the value in the notification_channel column. This is the SQS queue ARN you will configure S3 to send events to.

Don't see a notification_channel value?

Make sure you set AUTO_INGEST = TRUE when creating the pipe. Without it, Snowflake does not provision an SQS queue. Re-run the CREATE OR REPLACE PIPE statement with the correct setting.

help me fix my Snowpipe that is not showing a notification_channel in SHOW PIPES output

Configure the S3 event notification for gold/

Your bucket already has an event notification on bronze/ that triggers Lambda. Now you need to add a second notification on gold/ that sends to the Snowflake SQS queue.

This command replaces all existing notifications

The put-bucket-notification-configuration command replaces the entire notification configuration for your bucket. You must include BOTH the existing Lambda trigger (for bronze/) AND the new SQS trigger (for gold/) in a single JSON file. Omitting the Lambda trigger will break your pipeline.

  • Create a file called notification.json with both notification configurations:
{
  "LambdaFunctionConfigurations": [
    {
      "LambdaFunctionArn": "[[LAMBDA_ARN="arn:aws:lambda:region:account-id:function:medallion-processor"]]",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {"Name": "prefix", "Value": "bronze/"}
          ]
        }
      }
    }
  ],
  "QueueConfigurations": [
    {
      "QueueArn": "[[SQS_ARN="notification_channel value from SHOW PIPES"]]",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {"Name": "prefix", "Value": "gold/"}
          ]
        }
      }
    }
  ]
}

What does this configuration do?

The LambdaFunctionConfigurations section preserves your existing trigger: when a file lands in bronze/, Lambda processes it. The QueueConfigurations section adds the new trigger: when a file lands in gold/, S3 notifies the Snowflake SQS queue, and Snowpipe loads it.

  • Apply the notification configuration to your bucket by running:
aws s3api put-bucket-notification-configuration \
  --bucket [[BUCKET_NAME="your-bucket-name"]] \
  --notification-configuration file://notification.json

No output means success. Both notifications are now active on your bucket.

Getting an invalid configuration error?

Check that your Lambda function ARN and SQS queue ARN are correct. The Lambda ARN must match the function you deployed earlier. The SQS ARN comes from the notification_channel column in SHOW PIPES output.

Also verify notification.json is valid JSON with no trailing commas or mismatched brackets.

help me fix my S3 put-bucket-notification-configuration error

Verify the full end-to-end pipeline

Everything is wired together. Time to trigger the entire pipeline from start to finish and confirm data arrives in Snowflake.

Before you run this, what do you expect to happen? Think about the chain: data generator uploads to bronze/, Lambda processes to silver/ and gold/, Snowpipe detects the gold/ file and loads it into gold_sales.

  • Run your data generator one more time:
python data_generator.py

You should see the familiar upload confirmation message.

  • Wait 1-2 minutes for the full pipeline to execute. Lambda processes the file, writes to gold/, and Snowpipe picks it up.
  • Switch back to your Snowflake worksheet and run:
SELECT * FROM gold_sales;

You should see rows of aggregated sales data with product names, total quantities, and total revenue values.

  • Run the data generator a second time to confirm new data flows through automatically:
python data_generator.py
  • Wait 1-2 minutes, then re-run the query in Snowflake:
SELECT * FROM gold_sales;

You should see additional rows appear. Each run of the data generator produces new aggregated data that Snowpipe loads automatically.

No data appearing in gold_sales?

First check that Lambda ran by looking at CloudWatch Logs for the medallion-processor function. Then check that gold/ has new files with aws s3 ls s3://your-bucket-name/gold/.

If gold/ has files but Snowflake is empty, the Snowpipe SQS notification might not be connected. Verify the S3 event notification is configured correctly by running aws s3api get-bucket-notification-configuration --bucket your-bucket-name and checking for the QueueConfigurations entry.

help me debug why Snowpipe is not loading data from my gold/ prefix into Snowflake

Your pipeline is fully automated. Every time you run the data generator, data flows from S3 bronze through Lambda processing all the way into a queryable Snowflake table. No manual steps required.

Secret mission

Add Data Quality Validation with Quarantine

Production pipelines encounter bad data constantly. Can you add validation that separates good records from bad, quarantining invalid rows to a separate S3 prefix with a reason for rejection? You'll modify both the data generator and the Lambda function.

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 pipeline costs almost nothing at rest. Lambda only charges when triggered, S3 storage is pennies per month for small files, and your Snowflake warehouse auto-suspends after inactivity.

Cost warning

Your Snowflake trial credits are consumed whenever the warehouse is active (running queries or loading data via Snowpipe). If you are done experimenting, choose Pause or Delete to avoid ongoing credit consumption.

Resources you used:

  • S3 bucket with bronze/, silver/, gold/, and quarantine/ prefixes.
  • Lambda function medallion-processor.
  • IAM role medallion-lambda-role (Lambda execution role).
  • IAM role snowflake-access-role (Snowflake storage integration role).
  • Snowflake objects: database pipeline_db, storage integration, stage, pipe, and table.
  • Local project files: data_generator.py, lambda_function.py, snowflake_setup.sql, and generated CSV files.
  • Python package boto3 installed via pip.

Keep everything running

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

  • Lambda only runs (and charges) when a file lands in bronze/. No triggers means no cost.
  • Snowpipe only charges when it loads a file. No new gold/ files means no Snowpipe compute.
  • The Snowflake warehouse auto-suspends after 5 minutes of inactivity by default, stopping credit consumption.
  • S3 storage charges a few cents per month for the small CSV files stored across your prefixes.
  • The boto3 package and local project files remain on your machine.

Pause - I'll come back to this later

Shut down running processes and suspend resources to free up credits while keeping everything intact for later.

  • Your Snowflake warehouse auto-suspends after 5 minutes of inactivity. No manual action is needed unless you changed the default.
  • Stop running your data generator script. Without new files in bronze/, Lambda will not trigger and Snowpipe will not load.
  • Your S3 bucket, Lambda function, IAM roles, and Snowflake objects all remain intact and ready to resume.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each group below in order.

  • Delete the S3 bucket and all its contents by running:
aws s3 rb s3://[[BUCKET="your-bucket-name"]] --force
  • Delete the Lambda function by running:
aws lambda delete-function --function-name medallion-processor
  • Delete the IAM roles. You must detach all attached policies before deleting each role. For medallion-lambda-role, list its attached policies:
aws iam list-attached-role-policies --role-name medallion-lambda-role
  • For each policy ARN returned, detach it:
aws iam detach-role-policy --role-name medallion-lambda-role --policy-arn [[POLICY_ARN="the policy ARN from the previous output"]]
  • Then delete the role:
aws iam delete-role --role-name medallion-lambda-role
  • Repeat for snowflake-access-role:
aws iam list-attached-role-policies --role-name snowflake-access-role
aws iam detach-role-policy --role-name snowflake-access-role --policy-arn [[POLICY_ARN_SF="the policy ARN from the previous output"]]
aws iam delete-role --role-name snowflake-access-role
  • Drop Snowflake objects. Run these SQL commands in Snowflake as ACCOUNTADMIN:
USE DATABASE pipeline_db;
USE SCHEMA public;
DROP PIPE IF EXISTS gold_pipe;
DROP TABLE IF EXISTS gold_sales;
DROP STAGE IF EXISTS gold_stage;
DROP STORAGE INTEGRATION IF EXISTS s3_pipeline_int;
DROP DATABASE IF EXISTS pipeline_db;
  • Delete local project files. On macOS or Linux, run:
rm -f data_generator.py lambda_function.py snowflake_setup.sql sales_*.csv
  • On Windows, run:
Remove-Item data_generator.py, lambda_function.py, snowflake_setup.sql, sales_*.csv
  • Uninstall boto3 by running:
pip uninstall boto3 -y

Nice Work!

Nice Work!

Nice work! You just built a fully automated, event-driven batch data pipeline using the medallion architecture pattern that production data teams rely on every day.

You've learned how to:

  • Build a data generator that uploads sales data to S3 and trigger an entire pipeline with a single script run.
  • Create an AWS Lambda function that automatically transforms raw data through silver (cleaned) and gold (aggregated) layers using S3 event triggers.
  • Connect Snowflake via Snowpipe auto-ingest to automatically load gold-layer data into a warehouse table for querying.
  • Secret Mission: added data quality validation with a quarantine pattern that separates bad records from clean ones before they reach downstream layers.

Ready to quiz yourself?