Customer Behavior Analytics Pipeline
Build an ETL pipeline merging Redshift and MongoDB data for BI insights.
Introduction
30 Second Summary
Every time you browse a store, click a marketing email, or buy a ticket online, you generate data across multiple systems. Businesses that can combine these scattered signals into a single picture unlock powerful insights about what drives customer decisions.
In this project, you will build an end-to-end analytics pipeline that pulls customer data from two different databases, merges it into a unified dataset, runs statistical tests to identify spending patterns, and visualizes the results in an interactive dashboard.
What You'll Build
A Python ETL pipeline that integrates transactional sales data from Amazon Redshift with customer interaction data from MongoDB Atlas, applies statistical hypothesis testing, and delivers insights through an Amazon QuickSight dashboard.
By the end of this project, you'll have:
- A working Python ETL pipeline that extracts data from Amazon Redshift and MongoDB Atlas, transforms and merges it into an analytics-ready dataset.
- Statistical test results (t-test and Chi-Squared) that identify significant customer behavior patterns from the merged data.
- An interactive Amazon QuickSight dashboard connected to Redshift displaying KPIs, sales trends, and customer segment analysis.
- Secret Mission: Build an RFM customer segmentation model and add a segmentation heatmap to your QuickSight dashboard.
Are there any prerequisites?
This project assumes you have an AWS account with Redshift Serverless free trial eligibility, a MongoDB Atlas account with a cluster running, Python 3.10 installed, and the AWS CLI configured.
You do not need prior experience with Redshift, QuickSight, or statistical testing. The project guides you through each tool step by step.
Before We Start
Every real business stores its data across multiple systems. Sales transactions live in a data warehouse optimized for fast analytics. Customer behavior data (clicks, email opens, page views) lives in a flexible document store. To answer meaningful questions about customer behavior, you need to bring these data sources together.
In this step, you will install the Python packages needed for data extraction and analysis, set up your Amazon Redshift Serverless data warehouse with sample data, and verify your MongoDB Atlas connection from Python.
In this step, get ready to:
- Install Python packages for data extraction, transformation, and statistical analysis.
- Create a Redshift Serverless workgroup and load the tickit sample dataset.
- Verify MongoDB Atlas connectivity and create your analytics database.
Install Python packages and create project structure
Your pipeline needs libraries to connect to both databases, manipulate data, and run statistical tests. You will install all of them now so they are ready when you start building.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir customer-analytics
cd customer-analytics- Install the required Python packages by running:
pip install redshift-connector pymongo pandas numpy scipyWhat do these packages do?
Each package handles a different part of the pipeline:
- redshift-connector connects Python directly to Amazon Redshift Serverless.
- pymongo connects Python to MongoDB Atlas.
- pandas and numpy handle data manipulation and numerical operations.
- scipy provides the statistical tests (t-test, Chi-Squared) you will run later.
- Verify all packages installed correctly by running:
python3 -c "import redshift_connector; import pymongo; import pandas; import numpy; import scipy; print('All packages installed successfully')"- You should see All packages installed successfully printed in your terminal.
- Create the two Python files you will use throughout this project by running:
touch generate_interactions.py analytics_pipeline.py- Confirm the files exist by running:
ls- You should see generate_interactions.py and analytics_pipeline.py in the output.
Import error or package not found?
- Make sure you used pip that matches your Python 3.10 installation. Try pip3 instead of pip if your system has multiple Python versions.
- If you see a permissions error, try adding --user to the end of the pip install command.
- help me troubleshoot my Python package installation
Set up Redshift Serverless and load sample data
Amazon Redshift Serverless is a cloud data warehouse that charges only when queries run. You will use its built-in tickit sample dataset, which contains sales transaction data for events like concerts and shows.
What is Amazon Redshift Serverless?
Redshift Serverless is a columnar data warehouse designed for fast analytical queries over large datasets. Unlike a traditional database that stores data row by row, Redshift stores data by column. This makes aggregation queries (SUM, AVG, COUNT) extremely fast.
The "Serverless" part means you do not manage any infrastructure. Redshift automatically scales compute up when queries run and pauses when idle. You are billed per RPU-hour (Redshift Processing Unit) on a per-second basis, with no charge when idle.
- Open the AWS Management Console in your browser and sign in to your AWS account.
- In the search bar at the top, type Redshift and select Amazon Redshift from the results.
- Click Try Redshift Serverless Free Trial.
- Under Configuration, choose Use default settings. This creates a default namespace and a default workgroup automatically.
- Click Save configuration.
- Wait for setup to complete (usually under 3 minutes), then click Continue to go to your Serverless dashboard.
What is the free trial?
The Redshift Serverless free trial gives you $300 in credits valid for 90 days. This project's queries will use only a tiny fraction of that credit. You must delete resources after the project to avoid charges once the trial expires.
Now load the tickit sample dataset so you have sales data to work with.
- In the left navigation panel, click Query editor v2.
- In the left panel of Query Editor v2, expand the sample_data_dev database.
- Select the tickit schema icon to load the sample data.
- Wait for the data to load. Once complete, you can expand the tickit schema to see tables like users, sales, event, and category.
Next, set a usage limit to prevent unexpected charges during the project.
- Go back to the main Redshift console (click the Amazon Redshift logo or navigate back).
- In the left navigation, click Workgroup configuration.
- Click your workgroup name (default-workgroup).
- Click the Limits tab.
- Click Manage usage limits.
- Set the Frequency to Daily.
- Set the usage limit to 1 RPU hour.
- Set the action to Turn off user queries.
- Click Save changes.
Why set a usage limit?
Redshift Serverless charges per RPU-hour when queries are running. Setting a daily limit of 1 RPU-hour means Redshift will automatically stop processing queries if you exceed that threshold in a single day. This protects you from unexpected costs while completing the project.
Workgroup creation failed?
- If you see an error about too few subnets, your default VPC may not have subnets in enough Availability Zones. Try a different region that has at least 2 AZs in your VPC.
- If you do not see the free trial option, your account may have previously used Redshift Serverless. Check the Redshift Serverless free trial eligibility requirements.
- help me troubleshoot my Redshift Serverless workgroup creation
Verify MongoDB Atlas connection
Your MongoDB Atlas cluster will store customer interaction data. Before writing the data generation script, confirm that Python can connect to your cluster and create the database you need.
- In your MongoDB Atlas dashboard, click Connect on your cluster.
- Choose Drivers and copy your connection string. It looks like mongodb+srv://your-username:your-password@your-cluster.mongodb.net/.
- Make sure your current IP address is in the Atlas Network Access allow list. If not, add it now.
- Back in your terminal, open a Python shell by running:
python3- Test the connection and create your database by running this code in the Python shell:
import pymongo
# Replace with your actual MongoDB Atlas connection string
client = pymongo.MongoClient("mongodb+srv://your-username:your-password@your-cluster.mongodb.net/")
# Verify connection by listing databases
print(client.list_database_names())
# Create the database and collection
db = client["customer_analytics"]
collection = db["interactions"]
# Insert a test document to force collection creation
collection.insert_one({"test": True})
print("Collection created successfully!")
# Clean up the test document
collection.delete_one({"test": True})
print("Connection verified and collection ready!")
client.close()What does this code do?
- pymongo.MongoClient(...) establishes a connection to your MongoDB Atlas cluster using the connection string.
- client.list_database_names() confirms the connection works by listing existing databases.
- MongoDB creates databases and collections lazily. Inserting a test document forces the customer_analytics database and interactions collection to actually exist on the server.
- You should see your database names printed, followed by Collection created successfully! and Connection verified and collection ready!.
- Exit the Python shell by typing:
exit()Connection refused or timeout?
- Check that your IP address is in the Atlas Network Access allow list. Atlas blocks connections from IPs not on the list.
- Verify your username and password in the connection string are correct. Special characters in the password may need URL encoding.
- Make sure you replaced the entire placeholder connection string with your actual one from the Atlas dashboard.
- help me connect to MongoDB Atlas from Python
Your environment is ready. You have Python packages installed, a Redshift data warehouse loaded with sales data, and a MongoDB database waiting for interaction data. Next up, you will generate realistic customer interaction records and load them into MongoDB.
Generate and Load Customer Interaction Data
Your data infrastructure is in place. Redshift has the tickit sales data, MongoDB Atlas is connected, and your Python packages are ready.
Now it's time to bridge the two systems. In this step, you'll generate realistic customer interaction data linked to real Redshift user IDs and load it into MongoDB Atlas.
In this step, get ready to:
- Query Redshift for real user IDs from the tickit dataset.
- Generate 50,000 synthetic customer interaction records in Python.
- Bulk insert all records into MongoDB Atlas.
Query Redshift for user IDs
Your synthetic interaction data needs to reference real users from the tickit dataset. By pulling user IDs directly from Redshift, every generated interaction will map back to an actual user record for clean merging later.
- Open generate_interactions.py in your code editor.
- Add the imports and configuration block at the top of the file:
import random
import datetime
import uuid
import redshift_connector
import pymongo
import pandas as pd
# --- Configuration ---
REDSHIFT_HOST = "your-workgroup.your-account.your-region.redshift-serverless.amazonaws.com"
REDSHIFT_DATABASE = "sample_data_dev"
REDSHIFT_USER = "admin"
REDSHIFT_PASSWORD = "your-password-here"
MONGO_URI = "mongodb+srv://your-username:your-password@your-cluster.mongodb.net/"
MONGO_DB = "customer_analytics"
MONGO_COLLECTION = "interactions"What does this configuration do?
The imports bring in the libraries you installed in Step 1. The configuration block defines connection details for both data sources in one place.
REDSHIFT_HOST is your workgroup endpoint (found in the Redshift Serverless console under your workgroup's details). MONGO_URI is your Atlas connection string (found in the Atlas console under Connect).
- Replace the placeholder values in REDSHIFT_HOST, REDSHIFT_USER, REDSHIFT_PASSWORD, and MONGO_URI with your actual credentials.
- Your Redshift host endpoint looks like workgroup-name.account-id.region.redshift-serverless.amazonaws.com.
Finding your Redshift endpoint
In the AWS Console, go to Amazon Redshift, select Serverless dashboard, then click your workgroup name. The endpoint is listed under General information.
- Below the configuration block, add the Redshift query code:
# --- Step 1: Get user IDs from Redshift ---
print("Connecting to Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DATABASE,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
cursor.execute("SELECT userid FROM tickit.users")
user_ids_df = cursor.fetch_dataframe()
user_ids = user_ids_df["userid"].tolist()
print(f"Retrieved {len(user_ids)} user IDs from Redshift")
conn.close()What does this code do?
redshift_connector.connect() opens a connection to your Redshift Serverless workgroup using the credentials you configured above.
cursor.fetch_dataframe() pulls the query results directly into a pandas DataFrame. The .tolist() call converts the user IDs into a plain Python list that the data generator can sample from.
- Save generate_interactions.py.
Connection refused or timeout?
Make sure your Redshift Serverless workgroup has Publicly accessible enabled in the workgroup settings. Also verify that your endpoint, username, and password are correct.
help me debug my Redshift connection error
Generate synthetic interaction data
Customer behavior data includes page views, cart activity, and email engagement. You'll simulate a full year of this activity across 50,000 records with weighted event probabilities that mirror real user behavior.
- Below conn.close(), add the data generation code:
# --- Step 2: Generate synthetic interaction data ---
print("Generating synthetic interaction data...")
EVENT_TYPES = ["page_view", "add_to_cart", "purchase", "email_open", "email_click"]
EVENT_WEIGHTS = [0.45, 0.20, 0.10, 0.15, 0.10]
DEVICES = ["desktop", "mobile", "tablet"]
REGIONS = ["US-East", "US-West", "EU-West", "APAC"]
CAMPAIGNS = [None, "summer_sale", "loyalty_program", "new_arrivals", "flash_deal"]
NUM_RECORDS = 50000
interactions = []
start_date = datetime.datetime(2024, 1, 1)
end_date = datetime.datetime(2024, 12, 31)
date_range_seconds = int((end_date - start_date).total_seconds())
for _ in range(NUM_RECORDS):
user_id = random.choice(user_ids)
event_type = random.choices(EVENT_TYPES, weights=EVENT_WEIGHTS, k=1)[0]
timestamp = start_date + datetime.timedelta(seconds=random.randint(0, date_range_seconds))
session_id = str(uuid.uuid4())[:8]
device = random.choice(DEVICES)
region = random.choice(REGIONS)
campaign_id = random.choice(CAMPAIGNS)
interaction = {
"user_id": int(user_id),
"event_type": event_type,
"timestamp": timestamp,
"session_id": session_id,
"device": device,
"region": region,
"campaign_id": campaign_id,
}
interactions.append(interaction)
print(f"Generated {len(interactions)} interaction records")What does this code do?
EVENT_WEIGHTS makes the distribution realistic. Page views are most common (45%), purchases are rare (10%), and email events fill the middle. This mirrors real e-commerce funnels.
Each record gets a random user ID pulled from the Redshift users you just queried. This creates the link between MongoDB interactions and Redshift transactions that your ETL pipeline will merge later.
campaign_id is sometimes None because not every interaction comes from a marketing campaign. This adds realistic sparsity to the data.
- Save generate_interactions.py.
Script running slowly?
Generating 50,000 records in a loop takes a few seconds. If it seems stuck, check that NUM_RECORDS is set to 50000 and not a much larger number.
help me understand why my data generation is slow
Insert into MongoDB Atlas
With 50,000 records ready in memory, you'll bulk insert them into MongoDB in batches. Batching prevents timeout issues and lets you track insertion progress.
- Below the print(f"Generated {len(interactions)} interaction records") line, add the MongoDB insertion code:
# --- Step 3: Insert into MongoDB ---
print("Connecting to MongoDB Atlas...")
mongo_client = pymongo.MongoClient(MONGO_URI)
db = mongo_client[MONGO_DB]
collection = db[MONGO_COLLECTION]
# Clear existing data
collection.delete_many({})
# Bulk insert
BATCH_SIZE = 5000
for i in range(0, len(interactions), BATCH_SIZE):
batch = interactions[i:i + BATCH_SIZE]
collection.insert_many(batch)
print(f" Inserted batch {i // BATCH_SIZE + 1}/{(len(interactions) // BATCH_SIZE) + 1}")
count = collection.count_documents({})
print(f"MongoDB now contains {count} interaction documents")
mongo_client.close()
print("Done!")What does this code do?
collection.delete_many({}) clears any existing documents first. This makes the script idempotent, so you can safely re-run it without duplicating data.
The loop splits 50,000 records into batches of 5,000. Each call to insert_many() sends one batch to MongoDB Atlas. The print statement shows progress so you know it's working.
- Save generate_interactions.py.
- Run the script by executing this command in your terminal:
python generate_interactions.pyYou should see output showing the Redshift connection, data generation progress, and 10 batch insertion confirmations. The final line should confirm MongoDB contains 50,000 interaction documents.
Seeing an authentication error for MongoDB?
Double check that your MONGO_URI includes the correct username and password. Also verify your Atlas cluster allows connections from your current IP address (check Network Access in the Atlas console).
help me fix my MongoDB Atlas authentication or network error
✔️ Awesome, I've got everything!
Great. Double check you've saved your file and the terminal output shows 50,000 documents in MongoDB.
ⓧ I'd like to double check the full code
import random
import datetime
import uuid
import redshift_connector
import pymongo
import pandas as pd
# --- Configuration ---
REDSHIFT_HOST = "your-workgroup.your-account.your-region.redshift-serverless.amazonaws.com"
REDSHIFT_DATABASE = "sample_data_dev"
REDSHIFT_USER = "admin"
REDSHIFT_PASSWORD = "your-password-here"
MONGO_URI = "mongodb+srv://your-username:your-password@your-cluster.mongodb.net/"
MONGO_DB = "customer_analytics"
MONGO_COLLECTION = "interactions"
# --- Step 1: Get user IDs from Redshift ---
print("Connecting to Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DATABASE,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
cursor.execute("SELECT userid FROM tickit.users")
user_ids_df = cursor.fetch_dataframe()
user_ids = user_ids_df["userid"].tolist()
print(f"Retrieved {len(user_ids)} user IDs from Redshift")
conn.close()
# --- Step 2: Generate synthetic interaction data ---
print("Generating synthetic interaction data...")
EVENT_TYPES = ["page_view", "add_to_cart", "purchase", "email_open", "email_click"]
EVENT_WEIGHTS = [0.45, 0.20, 0.10, 0.15, 0.10]
DEVICES = ["desktop", "mobile", "tablet"]
REGIONS = ["US-East", "US-West", "EU-West", "APAC"]
CAMPAIGNS = [None, "summer_sale", "loyalty_program", "new_arrivals", "flash_deal"]
NUM_RECORDS = 50000
interactions = []
start_date = datetime.datetime(2024, 1, 1)
end_date = datetime.datetime(2024, 12, 31)
date_range_seconds = int((end_date - start_date).total_seconds())
for _ in range(NUM_RECORDS):
user_id = random.choice(user_ids)
event_type = random.choices(EVENT_TYPES, weights=EVENT_WEIGHTS, k=1)[0]
timestamp = start_date + datetime.timedelta(seconds=random.randint(0, date_range_seconds))
session_id = str(uuid.uuid4())[:8]
device = random.choice(DEVICES)
region = random.choice(REGIONS)
campaign_id = random.choice(CAMPAIGNS)
interaction = {
"user_id": int(user_id),
"event_type": event_type,
"timestamp": timestamp,
"session_id": session_id,
"device": device,
"region": region,
"campaign_id": campaign_id,
}
interactions.append(interaction)
print(f"Generated {len(interactions)} interaction records")
# --- Step 3: Insert into MongoDB ---
print("Connecting to MongoDB Atlas...")
mongo_client = pymongo.MongoClient(MONGO_URI)
db = mongo_client[MONGO_DB]
collection = db[MONGO_COLLECTION]
# Clear existing data
collection.delete_many({})
# Bulk insert
BATCH_SIZE = 5000
for i in range(0, len(interactions), BATCH_SIZE):
batch = interactions[i:i + BATCH_SIZE]
collection.insert_many(batch)
print(f" Inserted batch {i // BATCH_SIZE + 1}/{(len(interactions) // BATCH_SIZE) + 1}")
count = collection.count_documents({})
print(f"MongoDB now contains {count} interaction documents")
mongo_client.close()
print("Done!")Why generate synthetic data instead of using real data?
In real data engineering, you often need to test pipelines before production data is available. Generating synthetic data tied to real IDs (the Redshift user IDs) lets you build and validate your entire pipeline end-to-end.
This approach also gives you control over the schema and volume. You know exactly what fields exist and can guarantee meaningful joins between your two data sources.
Your MongoDB Atlas collection is now loaded with 50,000 interaction records linked to real Redshift users. Next up, you'll build the ETL pipeline that merges this behavioral data with transactional sales data into a unified analytics dataset.
Build the ETL Pipeline
Your MongoDB Atlas collection now holds 50,000+ interaction records tied to real user IDs from Redshift. The raw data exists in two separate systems, but it tells one story about customer behavior.
In this step, you will build an ETL pipeline that extracts transactional sales data from Amazon Redshift and behavioral interaction data from MongoDB Atlas, transforms both into per-user aggregates, merges them into a single analytics DataFrame, and writes the result back to Redshift for downstream visualization.
In this step, get ready to:
- Extract sales data from the tickit dataset in Redshift using SQL JOINs.
- Extract and aggregate interaction data from MongoDB Atlas.
- Merge both sources into a unified analytics DataFrame and write it back to Redshift.
Extract sales data from Redshift
The tickit dataset in Redshift contains sales transactions, event details, and user profiles spread across multiple tables. A single SQL query with JOINs pulls all of this together into one result set.
- Open analytics_pipeline.py in your code editor and add the imports and configuration block:
import redshift_connector
import pymongo
import pandas as pd
import numpy as np
from scipy import stats
# --- Configuration ---
REDSHIFT_HOST = "your-workgroup.your-account.your-region.redshift-serverless.amazonaws.com"
REDSHIFT_DB_SAMPLE = "sample_data_dev"
REDSHIFT_DB_DEV = "dev"
REDSHIFT_USER = "admin"
REDSHIFT_PASSWORD = "your-password-here"
MONGO_URI = "mongodb+srv://your-username:your-password@your-cluster.mongodb.net/"
MONGO_DB = "customer_analytics"
MONGO_COLLECTION = "interactions"What does this code do?
The imports bring in your database connectors (redshift_connector and pymongo), data manipulation libraries (pandas and numpy), and statistical testing (scipy.stats).
The configuration section defines connection details for both Redshift and MongoDB. Notice there are two Redshift databases: REDSHIFT_DB_SAMPLE for reading the tickit data, and REDSHIFT_DB_DEV for writing your analytics results.
- Replace the placeholder values for REDSHIFT_HOST, REDSHIFT_PASSWORD, and MONGO_URI with your actual credentials (same values you used in generate_interactions.py).
- Below the configuration, add the Redshift extraction code:
# ============================================================
# STEP 3: ETL PIPELINE
# ============================================================
# --- Extract from Redshift ---
print("=" * 60)
print("STEP 3: ETL PIPELINE")
print("=" * 60)
print("\nExtracting sales data from Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DB_SAMPLE,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
sales_query = """
SELECT
s.buyerid AS user_id,
s.qtysold,
s.pricepaid,
s.commission,
s.saletime,
e.catid,
e.eventname,
u.city,
u.state
FROM tickit.sales s
JOIN tickit.event e ON s.eventid = e.eventid
JOIN tickit.users u ON s.buyerid = u.userid
"""
cursor.execute(sales_query)
sales_df = cursor.fetch_dataframe()
print(f" Extracted {len(sales_df)} sales records from Redshift")
conn.close()What does this query do?
The SQL query joins three tickit tables to create a complete picture of each sale. The sales table provides transaction amounts. The event table adds the event category. The users table adds geographic info.
The cursor.fetch_dataframe() method is a built-in feature of redshift-connector that returns the query results directly as a pandas DataFrame.
- Save analytics_pipeline.py.
Getting a connection error?
Double-check that your REDSHIFT_HOST matches the endpoint shown in the Redshift Serverless console under your workgroup details. Make sure the workgroup has Publicly accessible turned on in its settings.
If you see an authentication error, confirm REDSHIFT_USER and REDSHIFT_PASSWORD match what you set during workgroup creation. Help me troubleshoot my Redshift connection error.
Extract interaction data from MongoDB
The interaction data in MongoDB is stored as individual event documents. You need to pull all of them into Python, then aggregate them per user so each user has a single row of behavioral metrics.
- Add the MongoDB extraction code below the Redshift section:
# --- Extract from MongoDB ---
print("\nExtracting interaction data from MongoDB...")
mongo_client = pymongo.MongoClient(MONGO_URI)
db = mongo_client[MONGO_DB]
collection = db[MONGO_COLLECTION]
interactions = list(collection.find({}, {"_id": 0}))
interactions_df = pd.DataFrame(interactions)
print(f" Extracted {len(interactions_df)} interaction records from MongoDB")
mongo_client.close()Why exclude _id?
MongoDB automatically adds an _id field to every document. The projection {"_id": 0} tells MongoDB to exclude it from the results. This keeps your DataFrame clean with only the fields you inserted.
- Now add the interaction aggregation code that rolls up events per user:
# --- Transform: Aggregate interactions per user ---
interaction_agg = interactions_df.groupby("user_id").agg(
page_views=("event_type", lambda x: (x == "page_view").sum()),
cart_additions=("event_type", lambda x: (x == "add_to_cart").sum()),
email_opens=("event_type", lambda x: (x == "email_open").sum()),
email_clicks=("event_type", lambda x: (x == "email_click").sum()),
total_sessions=("session_id", "nunique"),
).reset_index()
interaction_agg["email_engagement"] = interaction_agg["email_opens"] + interaction_agg["email_clicks"]What does this aggregation produce?
Each user now has a single row with counts of page views, cart additions, email opens, email clicks, and unique sessions. The email_engagement column combines opens and clicks into one metric for easier comparison later.
- Save analytics_pipeline.py.
Seeing zero records from MongoDB?
Confirm that you successfully ran generate_interactions.py in the previous step and saw the count printed as 50,000. Also verify that the MONGO_URI, MONGO_DB, and MONGO_COLLECTION values match what you used in the generation script. Help me figure out why my MongoDB extraction returns zero records.
Transform and merge the datasets
You now have raw sales data in sales_df and aggregated interaction metrics in interaction_agg. The sales data still needs to be aggregated per user before you can merge the two sources together.
- Add the sales aggregation code that computes per-user spending metrics:
# --- Transform: Aggregate sales per user ---
print("\nTransforming data...")
sales_agg = sales_df.groupby("user_id").agg(
total_spent=("pricepaid", "sum"),
purchase_count=("pricepaid", "count"),
avg_ticket_price=("pricepaid", "mean"),
total_commission=("commission", "sum"),
).reset_index()
# Get most common category per user
top_category = sales_df.groupby("user_id")["catid"].agg(
lambda x: x.mode().iloc[0] if not x.mode().empty else None
).reset_index()
top_category.columns = ["user_id", "preferred_category"]
sales_agg = sales_agg.merge(top_category, on="user_id", how="left")
# Get user region from sales data
user_region = sales_df.groupby("user_id")["state"].first().reset_index()
user_region.columns = ["user_id", "user_state"]
sales_agg = sales_agg.merge(user_region, on="user_id", how="left")What does this aggregation produce?
Each user gets a total_spent, purchase_count, avg_ticket_price, and total_commission from their transaction history.
The preferred_category captures which event category (concerts, sports, etc.) the user spent the most on using the statistical mode. The user_state gives their geographic location for regional analysis.
- Add the merge and derived features code:
# --- Merge ---
print("\nMerging datasets...")
analytics_df = sales_agg.merge(interaction_agg, on="user_id", how="inner")
# Derived features
analytics_df["conversion_rate"] = np.where(
analytics_df["page_views"] > 0,
analytics_df["purchase_count"] / analytics_df["page_views"],
0,
)
analytics_df["engagement_score"] = (
analytics_df["email_engagement"] * 0.3
+ analytics_df["page_views"] * 0.3
+ analytics_df["cart_additions"] * 0.4
)
print(f" Merged dataset: {len(analytics_df)} users with both transactional and behavioral data")
print(f" Columns: {list(analytics_df.columns)}")What are the derived features?
The conversion_rate divides purchases by page views to measure how effectively a user converts browsing into buying. The np.where guard prevents division by zero for users with no page views.
The engagement_score is a weighted composite that combines email engagement (30%), page views (30%), and cart additions (40%). This single number summarizes overall behavioral engagement for each user.
- Finally, add the code that writes the merged results back to Redshift:
# --- Write back to Redshift ---
print("\nWriting aggregated results to Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DB_DEV,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS public.customer_analytics_summary")
cursor.execute("""
CREATE TABLE public.customer_analytics_summary (
user_id INTEGER,
total_spent DECIMAL(10,2),
purchase_count INTEGER,
avg_ticket_price DECIMAL(10,2),
preferred_category INTEGER,
user_state VARCHAR(50),
page_views INTEGER,
cart_additions INTEGER,
email_engagement INTEGER,
total_sessions INTEGER,
conversion_rate DECIMAL(8,4),
engagement_score DECIMAL(10,2)
)
""")
# Write using cursor.write_dataframe
write_df = analytics_df[[
"user_id", "total_spent", "purchase_count", "avg_ticket_price",
"preferred_category", "user_state", "page_views", "cart_additions",
"email_engagement", "total_sessions", "conversion_rate", "engagement_score"
]].copy()
cursor.write_dataframe(write_df, "customer_analytics_summary")
conn.commit()
print(f" Written {len(write_df)} rows to customer_analytics_summary")
conn.close()Why write back to Redshift?
The customer_analytics_summary table stores the merged, analytics-ready data in Redshift's dev database. This is the table you will connect to Amazon QuickSight in a later step to build your dashboard.
The conn.commit() call is critical. Without it, Redshift treats the CREATE TABLE and INSERT as uncommitted transactions that other sessions (like Query Editor v2) cannot see.
- Save analytics_pipeline.py.
- Run the pipeline by executing this command in your terminal:
python analytics_pipeline.py- You should see output showing the number of records extracted from each source, the merged dataset size, and a confirmation of rows written to customer_analytics_summary.
- To verify the table was created, switch to the Redshift Query Editor v2 in your AWS Console.
- Connect to the dev database and run:
SELECT COUNT(*) FROM public.customer_analytics_summary;- You should see a row count matching the number printed by your script.
Table not showing in Query Editor v2?
Make sure you are connected to the dev database in Query Editor v2, not sample_data_dev. The script writes to dev because that is the default writable database for Redshift Serverless.
If the query returns an error about the relation not existing, confirm your script printed the commit confirmation line. A missing conn.commit() means the table exists only in the script's session and is invisible to other connections. Help me troubleshoot why my customer_analytics_summary table is not visible in Redshift Query Editor.
✔️ Awesome, I've got everything!
Great. Double check you have saved analytics_pipeline.py and confirmed the table row count in Redshift.
ⓧ I'd like to double check the full code
import redshift_connector
import pymongo
import pandas as pd
import numpy as np
from scipy import stats
# --- Configuration ---
REDSHIFT_HOST = "your-workgroup.your-account.your-region.redshift-serverless.amazonaws.com"
REDSHIFT_DB_SAMPLE = "sample_data_dev"
REDSHIFT_DB_DEV = "dev"
REDSHIFT_USER = "admin"
REDSHIFT_PASSWORD = "your-password-here"
MONGO_URI = "mongodb+srv://your-username:your-password@your-cluster.mongodb.net/"
MONGO_DB = "customer_analytics"
MONGO_COLLECTION = "interactions"
# ============================================================
# STEP 3: ETL PIPELINE
# ============================================================
# --- Extract from Redshift ---
print("=" * 60)
print("STEP 3: ETL PIPELINE")
print("=" * 60)
print("\nExtracting sales data from Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DB_SAMPLE,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
sales_query = """
SELECT
s.buyerid AS user_id,
s.qtysold,
s.pricepaid,
s.commission,
s.saletime,
e.catid,
e.eventname,
u.city,
u.state
FROM tickit.sales s
JOIN tickit.event e ON s.eventid = e.eventid
JOIN tickit.users u ON s.buyerid = u.userid
"""
cursor.execute(sales_query)
sales_df = cursor.fetch_dataframe()
print(f" Extracted {len(sales_df)} sales records from Redshift")
conn.close()
# --- Extract from MongoDB ---
print("\nExtracting interaction data from MongoDB...")
mongo_client = pymongo.MongoClient(MONGO_URI)
db = mongo_client[MONGO_DB]
collection = db[MONGO_COLLECTION]
interactions = list(collection.find({}, {"_id": 0}))
interactions_df = pd.DataFrame(interactions)
print(f" Extracted {len(interactions_df)} interaction records from MongoDB")
mongo_client.close()
# --- Transform: Aggregate sales per user ---
print("\nTransforming data...")
sales_agg = sales_df.groupby("user_id").agg(
total_spent=("pricepaid", "sum"),
purchase_count=("pricepaid", "count"),
avg_ticket_price=("pricepaid", "mean"),
total_commission=("commission", "sum"),
).reset_index()
# Get most common category per user
top_category = sales_df.groupby("user_id")["catid"].agg(
lambda x: x.mode().iloc[0] if not x.mode().empty else None
).reset_index()
top_category.columns = ["user_id", "preferred_category"]
sales_agg = sales_agg.merge(top_category, on="user_id", how="left")
# Get user region from sales data
user_region = sales_df.groupby("user_id")["state"].first().reset_index()
user_region.columns = ["user_id", "user_state"]
sales_agg = sales_agg.merge(user_region, on="user_id", how="left")
# --- Transform: Aggregate interactions per user ---
interaction_agg = interactions_df.groupby("user_id").agg(
page_views=("event_type", lambda x: (x == "page_view").sum()),
cart_additions=("event_type", lambda x: (x == "add_to_cart").sum()),
email_opens=("event_type", lambda x: (x == "email_open").sum()),
email_clicks=("event_type", lambda x: (x == "email_click").sum()),
total_sessions=("session_id", "nunique"),
).reset_index()
interaction_agg["email_engagement"] = interaction_agg["email_opens"] + interaction_agg["email_clicks"]
# --- Merge ---
print("\nMerging datasets...")
analytics_df = sales_agg.merge(interaction_agg, on="user_id", how="inner")
# Derived features
analytics_df["conversion_rate"] = np.where(
analytics_df["page_views"] > 0,
analytics_df["purchase_count"] / analytics_df["page_views"],
0,
)
analytics_df["engagement_score"] = (
analytics_df["email_engagement"] * 0.3
+ analytics_df["page_views"] * 0.3
+ analytics_df["cart_additions"] * 0.4
)
print(f" Merged dataset: {len(analytics_df)} users with both transactional and behavioral data")
print(f" Columns: {list(analytics_df.columns)}")
# --- Write back to Redshift ---
print("\nWriting aggregated results to Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DB_DEV,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS public.customer_analytics_summary")
cursor.execute("""
CREATE TABLE public.customer_analytics_summary (
user_id INTEGER,
total_spent DECIMAL(10,2),
purchase_count INTEGER,
avg_ticket_price DECIMAL(10,2),
preferred_category INTEGER,
user_state VARCHAR(50),
page_views INTEGER,
cart_additions INTEGER,
email_engagement INTEGER,
total_sessions INTEGER,
conversion_rate DECIMAL(8,4),
engagement_score DECIMAL(10,2)
)
""")
# Write using cursor.write_dataframe
write_df = analytics_df[[
"user_id", "total_spent", "purchase_count", "avg_ticket_price",
"preferred_category", "user_state", "page_views", "cart_additions",
"email_engagement", "total_sessions", "conversion_rate", "engagement_score"
]].copy()
cursor.write_dataframe(write_df, "customer_analytics_summary")
conn.commit()
print(f" Written {len(write_df)} rows to customer_analytics_summary")
conn.close()Your ETL pipeline is complete. You now have a unified analytics dataset that combines who bought what (Redshift) with how they behaved online (MongoDB). Next up, you will run statistical tests on this data to find out whether engagement actually drives spending.
Run Statistical Analysis
Your ETL pipeline successfully merged transactional data from Redshift with behavioral data from MongoDB into a unified analytics DataFrame.
Now it is time to ask the data a real business question: does email engagement actually correlate with higher spending? You will apply two classical statistical tests to your merged dataset to find out.
In this step, get ready to:
- Segment customers by engagement level and run an independent samples t-test.
- Run a Chi-Squared test for independence between region and category preference.
- Print a formatted summary with effect sizes and significance conclusions.
Segment customers and run an independent samples t-test
A t-test compares the average of two groups and tells you whether the difference is statistically meaningful or just random noise. You will split your customers into high-engagement and low-engagement groups based on their median email_engagement value, then test whether the high group spends more.
- In analytics_pipeline.py, scroll to the very bottom of the file (after conn.close() in the write-back section).
- Add the t-test section by copying this code below the existing ETL code:
# ============================================================
# STEP 4: STATISTICAL ANALYSIS
# ============================================================
print("\n" + "=" * 60)
print("STEP 4: STATISTICAL ANALYSIS")
print("=" * 60)
# --- T-Test: High vs Low Email Engagement ---
print("\n--- Independent Samples T-Test ---")
print("H0: No difference in total_spent between high and low email engagement groups")
print("H1: There is a difference in total_spent between groups")
median_engagement = analytics_df["email_engagement"].median()
high_engagement = analytics_df[analytics_df["email_engagement"] > median_engagement]["total_spent"]
low_engagement = analytics_df[analytics_df["email_engagement"] <= median_engagement]["total_spent"]
t_stat, p_value_t = stats.ttest_ind(high_engagement, low_engagement)
# Cohen's d
pooled_std = np.sqrt(
((len(high_engagement) - 1) * high_engagement.std() ** 2
+ (len(low_engagement) - 1) * low_engagement.std() ** 2)
/ (len(high_engagement) + len(low_engagement) - 2)
)
cohens_d = (high_engagement.mean() - low_engagement.mean()) / pooled_std
print(f"\n High engagement group: n={len(high_engagement)}, mean=${high_engagement.mean():.2f}")
print(f" Low engagement group: n={len(low_engagement)}, mean=${low_engagement.mean():.2f}")
print(f" T-statistic: {t_stat:.4f}")
print(f" P-value: {p_value_t:.6f}")
print(f" Cohen's d: {cohens_d:.4f}")
print(f" Result: {'SIGNIFICANT' if p_value_t < 0.05 else 'NOT SIGNIFICANT'} at alpha=0.05")What does this code do?
The median split divides all users into two equally sized groups based on their combined email opens and clicks.
stats.ttest_ind() from SciPy runs an independent two-sample t-test. It returns a t-statistic (how far apart the group means are in standard-error units) and a p-value (the probability of seeing this difference by chance).
Cohen's d measures the practical size of the effect. A p-value tells you if the difference is real. Cohen's d tells you if it is big enough to matter.
What is a t-test?
An independent samples t-test answers the question: are these two groups genuinely different, or could the gap between them have appeared by luck alone?
If the p-value is below 0.05 (the standard significance threshold), you reject the null hypothesis and conclude the groups differ.
- Save analytics_pipeline.py.
You will not see output yet. The next two chunks complete the analysis section before you run the full script.
Run a Chi-Squared test for independence
The t-test checks if two groups have different averages. A Chi-Squared test checks whether two categorical variables are related at all. Here you will test whether a customer's geographic region predicts their preferred event category.
- In analytics_pipeline.py, add the Chi-Squared test section directly below the t-test code you just added:
# --- Chi-Squared Test: Region vs Category ---
print("\n--- Chi-Squared Test for Independence ---")
print("H0: Customer region and preferred category are independent")
print("H1: Customer region and preferred category are not independent")
contingency = pd.crosstab(analytics_df["user_state"], analytics_df["preferred_category"])
chi2, p_value_chi, dof, expected = stats.chi2_contingency(contingency)
# Cramer's V
n = contingency.sum().sum()
min_dim = min(contingency.shape[0], contingency.shape[1]) - 1
cramers_v = np.sqrt(chi2 / (n * min_dim)) if min_dim > 0 else 0
print(f"\n Contingency table shape: {contingency.shape}")
print(f" Chi2 statistic: {chi2:.4f}")
print(f" P-value: {p_value_chi:.6f}")
print(f" Degrees of freedom: {dof}")
print(f" Cramer's V: {cramers_v:.4f}")
print(f" Result: {'SIGNIFICANT' if p_value_chi < 0.05 else 'NOT SIGNIFICANT'} at alpha=0.05")What does this code do?
pd.crosstab() builds a contingency table counting how many users in each state prefer each event category.
stats.chi2_contingency() compares the observed counts against what you would expect if region and category were completely unrelated. A large chi2 value means the observed pattern deviates sharply from independence.
Cramer's V is the effect-size companion for Chi-Squared. It ranges from 0 (no association) to 1 (perfect association), making it easy to compare across different table sizes.
What is a Chi-Squared test?
A Chi-Squared test answers: are these two categorical variables independent, or does knowing one help you predict the other?
If region and category are independent, customers in every state buy the same mix of event types. If they are not independent, some regions strongly prefer certain categories.
- Save analytics_pipeline.py.
You will not see output yet. The final chunk adds a summary and then you will run the entire pipeline.
Summarize findings and run the pipeline
A clean summary section prints all your key metrics in one place. This makes it easy to scan the results without scrolling through individual test outputs.
- Add the analysis summary section at the bottom of analytics_pipeline.py, directly below the Chi-Squared code:
# --- Summary ---
print("\n" + "=" * 60)
print("ANALYSIS SUMMARY")
print("=" * 60)
print(f"\n Total users analyzed: {len(analytics_df)}")
print(f" Total sales records: {len(sales_df)}")
print(f" Total interactions: {len(interactions_df)}")
print(f"\n T-Test (engagement vs spending): p={p_value_t:.6f}, d={cohens_d:.4f}")
print(f" Chi-Squared (region vs category): p={p_value_chi:.6f}, V={cramers_v:.4f}")
print("\nPipeline complete!")What does this code do?
This summary block consolidates the total dataset sizes and both test results (p-values and effect sizes) into a quick-reference output. The final "Pipeline complete!" line confirms the entire script ran without error.
- Save analytics_pipeline.py.
- Run the full pipeline by executing this command in your terminal:
python analytics_pipeline.py- Wait for the script to complete. You should see the ETL extraction and transformation output, followed by the statistical analysis section showing both test results and the final ANALYSIS SUMMARY block.
Your console output ends with a summary showing the total users analyzed, p-values for both tests, Cohen's d, and Cramer's V. Each test prints SIGNIFICANT or NOT SIGNIFICANT at alpha=0.05.
How do I interpret my results?
If the t-test p-value is below 0.05, high-engagement customers spend significantly more than low-engagement customers. Cohen's d tells you the magnitude: below 0.2 is small, 0.5 is medium, 0.8+ is large.
If the Chi-Squared p-value is below 0.05, region and category preference are not independent. Cramer's V tells you how strongly they are related. Because your interaction data is randomly generated, your results will vary across runs.
Seeing an error when running the script?
- If you see a connection error, confirm your Redshift workgroup is still active and the credentials in the configuration section are correct.
- If you see NameError: name 'analytics_df' is not defined, make sure the statistical analysis code is below the ETL section that creates analytics_df. The script runs top to bottom.
- If you see a MongoDB timeout, check that your Atlas cluster is running and that your IP address is still on the allowlist.
Help me debug my analytics pipeline error.
✔️ Awesome, I've got everything!
Great. Double check you have saved analytics_pipeline.py and that your console shows the full ANALYSIS SUMMARY output.
ⓧ I'd like to double check the full code
Here is the complete analytics_pipeline.py file including both the ETL pipeline and the statistical analysis section:
import redshift_connector
import pymongo
import pandas as pd
import numpy as np
from scipy import stats
# --- Configuration ---
REDSHIFT_HOST = "your-workgroup.your-account.your-region.redshift-serverless.amazonaws.com"
REDSHIFT_DB_SAMPLE = "sample_data_dev"
REDSHIFT_DB_DEV = "dev"
REDSHIFT_USER = "admin"
REDSHIFT_PASSWORD = "your-password-here"
MONGO_URI = "mongodb+srv://your-username:your-password@your-cluster.mongodb.net/"
MONGO_DB = "customer_analytics"
MONGO_COLLECTION = "interactions"
# ============================================================
# STEP 3: ETL PIPELINE
# ============================================================
# --- Extract from Redshift ---
print("=" * 60)
print("STEP 3: ETL PIPELINE")
print("=" * 60)
print("\nExtracting sales data from Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DB_SAMPLE,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
sales_query = """
SELECT
s.buyerid AS user_id,
s.qtysold,
s.pricepaid,
s.commission,
s.saletime,
e.catid,
e.eventname,
u.city,
u.state
FROM tickit.sales s
JOIN tickit.event e ON s.eventid = e.eventid
JOIN tickit.users u ON s.buyerid = u.userid
"""
cursor.execute(sales_query)
sales_df = cursor.fetch_dataframe()
print(f" Extracted {len(sales_df)} sales records from Redshift")
conn.close()
# --- Extract from MongoDB ---
print("\nExtracting interaction data from MongoDB...")
mongo_client = pymongo.MongoClient(MONGO_URI)
db = mongo_client[MONGO_DB]
collection = db[MONGO_COLLECTION]
interactions = list(collection.find({}, {"_id": 0}))
interactions_df = pd.DataFrame(interactions)
print(f" Extracted {len(interactions_df)} interaction records from MongoDB")
mongo_client.close()
# --- Transform: Aggregate sales per user ---
print("\nTransforming data...")
sales_agg = sales_df.groupby("user_id").agg(
total_spent=("pricepaid", "sum"),
purchase_count=("pricepaid", "count"),
avg_ticket_price=("pricepaid", "mean"),
total_commission=("commission", "sum"),
).reset_index()
# Get most common category per user
top_category = sales_df.groupby("user_id")["catid"].agg(
lambda x: x.mode().iloc[0] if not x.mode().empty else None
).reset_index()
top_category.columns = ["user_id", "preferred_category"]
sales_agg = sales_agg.merge(top_category, on="user_id", how="left")
# Get user region from sales data
user_region = sales_df.groupby("user_id")["state"].first().reset_index()
user_region.columns = ["user_id", "user_state"]
sales_agg = sales_agg.merge(user_region, on="user_id", how="left")
# --- Transform: Aggregate interactions per user ---
interaction_agg = interactions_df.groupby("user_id").agg(
page_views=("event_type", lambda x: (x == "page_view").sum()),
cart_additions=("event_type", lambda x: (x == "add_to_cart").sum()),
email_opens=("event_type", lambda x: (x == "email_open").sum()),
email_clicks=("event_type", lambda x: (x == "email_click").sum()),
total_sessions=("session_id", "nunique"),
).reset_index()
interaction_agg["email_engagement"] = interaction_agg["email_opens"] + interaction_agg["email_clicks"]
# --- Merge ---
print("\nMerging datasets...")
analytics_df = sales_agg.merge(interaction_agg, on="user_id", how="inner")
# Derived features
analytics_df["conversion_rate"] = np.where(
analytics_df["page_views"] > 0,
analytics_df["purchase_count"] / analytics_df["page_views"],
0,
)
analytics_df["engagement_score"] = (
analytics_df["email_engagement"] * 0.3
+ analytics_df["page_views"] * 0.3
+ analytics_df["cart_additions"] * 0.4
)
print(f" Merged dataset: {len(analytics_df)} users with both transactional and behavioral data")
print(f" Columns: {list(analytics_df.columns)}")
# --- Write back to Redshift ---
print("\nWriting aggregated results to Redshift...")
conn = redshift_connector.connect(
host=REDSHIFT_HOST,
database=REDSHIFT_DB_DEV,
user=REDSHIFT_USER,
password=REDSHIFT_PASSWORD
)
cursor = conn.cursor()
cursor.execute("DROP TABLE IF EXISTS public.customer_analytics_summary")
cursor.execute("""
CREATE TABLE public.customer_analytics_summary (
user_id INTEGER,
total_spent DECIMAL(10,2),
purchase_count INTEGER,
avg_ticket_price DECIMAL(10,2),
preferred_category INTEGER,
user_state VARCHAR(50),
page_views INTEGER,
cart_additions INTEGER,
email_engagement INTEGER,
total_sessions INTEGER,
conversion_rate DECIMAL(8,4),
engagement_score DECIMAL(10,2)
)
""")
write_df = analytics_df[[
"user_id", "total_spent", "purchase_count", "avg_ticket_price",
"preferred_category", "user_state", "page_views", "cart_additions",
"email_engagement", "total_sessions", "conversion_rate", "engagement_score"
]].copy()
cursor.write_dataframe(write_df, "customer_analytics_summary")
conn.commit()
print(f" Written {len(write_df)} rows to customer_analytics_summary")
conn.close()
# ============================================================
# STEP 4: STATISTICAL ANALYSIS
# ============================================================
print("\n" + "=" * 60)
print("STEP 4: STATISTICAL ANALYSIS")
print("=" * 60)
# --- T-Test: High vs Low Email Engagement ---
print("\n--- Independent Samples T-Test ---")
print("H0: No difference in total_spent between high and low email engagement groups")
print("H1: There is a difference in total_spent between groups")
median_engagement = analytics_df["email_engagement"].median()
high_engagement = analytics_df[analytics_df["email_engagement"] > median_engagement]["total_spent"]
low_engagement = analytics_df[analytics_df["email_engagement"] <= median_engagement]["total_spent"]
t_stat, p_value_t = stats.ttest_ind(high_engagement, low_engagement)
# Cohen's d
pooled_std = np.sqrt(
((len(high_engagement) - 1) * high_engagement.std() ** 2
+ (len(low_engagement) - 1) * low_engagement.std() ** 2)
/ (len(high_engagement) + len(low_engagement) - 2)
)
cohens_d = (high_engagement.mean() - low_engagement.mean()) / pooled_std
print(f"\n High engagement group: n={len(high_engagement)}, mean=${high_engagement.mean():.2f}")
print(f" Low engagement group: n={len(low_engagement)}, mean=${low_engagement.mean():.2f}")
print(f" T-statistic: {t_stat:.4f}")
print(f" P-value: {p_value_t:.6f}")
print(f" Cohen's d: {cohens_d:.4f}")
print(f" Result: {'SIGNIFICANT' if p_value_t < 0.05 else 'NOT SIGNIFICANT'} at alpha=0.05")
# --- Chi-Squared Test: Region vs Category ---
print("\n--- Chi-Squared Test for Independence ---")
print("H0: Customer region and preferred category are independent")
print("H1: Customer region and preferred category are not independent")
contingency = pd.crosstab(analytics_df["user_state"], analytics_df["preferred_category"])
chi2, p_value_chi, dof, expected = stats.chi2_contingency(contingency)
# Cramer's V
n = contingency.sum().sum()
min_dim = min(contingency.shape[0], contingency.shape[1]) - 1
cramers_v = np.sqrt(chi2 / (n * min_dim)) if min_dim > 0 else 0
print(f"\n Contingency table shape: {contingency.shape}")
print(f" Chi2 statistic: {chi2:.4f}")
print(f" P-value: {p_value_chi:.6f}")
print(f" Degrees of freedom: {dof}")
print(f" Cramer's V: {cramers_v:.4f}")
print(f" Result: {'SIGNIFICANT' if p_value_chi < 0.05 else 'NOT SIGNIFICANT'} at alpha=0.05")
# --- Summary ---
print("\n" + "=" * 60)
print("ANALYSIS SUMMARY")
print("=" * 60)
print(f"\n Total users analyzed: {len(analytics_df)}")
print(f" Total sales records: {len(sales_df)}")
print(f" Total interactions: {len(interactions_df)}")
print(f"\n T-Test (engagement vs spending): p={p_value_t:.6f}, d={cohens_d:.4f}")
print(f" Chi-Squared (region vs category): p={p_value_chi:.6f}, V={cramers_v:.4f}")
print("\nPipeline complete!")Your statistical analysis is complete. You now have concrete evidence about whether email engagement drives higher spending and whether customer geography predicts category preferences. Next up, you will visualize these insights in an interactive QuickSight dashboard.
Build the QuickSight Dashboard
Your analytics data is sitting in Redshift, fully aggregated and ready for stakeholders. The missing piece is a visual layer that turns those numbers into insights people can actually explore.
In this step, you will connect Amazon QuickSight to your Redshift Serverless workgroup and build an interactive dashboard with KPIs, trend charts, and customer segment visualizations.
In this step, get ready to:
- Sign up for the QuickSight 30-day free trial and connect it to Redshift.
- Create KPI and sales trend visualizations from your analytics summary table.
- Add customer segment charts and publish your analysis as a dashboard.
Sign up for QuickSight and connect to Redshift
Amazon QuickSight is a cloud-native business intelligence service that lets you build interactive dashboards directly on top of data sources like Redshift. You will activate the 30-day free trial, then create a dataset that points to your customer_analytics_summary table.
- In the AWS Console, search for QuickSight in the top search bar and select it.
- You will be prompted to sign up. Enter a unique QuickSight account name (for example, your AWS username).
- Enter a notification email address for the account owner.
- Choose the same AWS Region you used for your Redshift Serverless workgroup as the initial SPICE data storage region.
- For authentication method, select the recommended option: Password-based.
- Review your choices and click Create account.
- Wait for the account to finish provisioning, then click Go to Quick to open the QuickSight console.
What is SPICE?
SPICE (Super-fast, Parallel, In-memory Calculation Engine) is QuickSight's in-memory data store. When you import data into SPICE, QuickSight caches it for fast querying without hitting your Redshift workgroup every time someone views the dashboard.
Before creating a dataset, you need to ensure your Redshift Serverless workgroup is publicly accessible so QuickSight can reach it.
- Open a new browser tab and navigate to the Amazon Redshift console.
- In the left navigation, select Serverless dashboard and click on your workgroup name.
- Under the workgroup settings, verify that Publicly accessible is turned on. If it is not, click Edit and enable it, then save.
Now return to the QuickSight console to create your dataset.
- In the QuickSight console, click Datasets in the left navigation.
- Click New dataset.
- Choose the Redshift Auto-discovered tile. If your workgroup does not appear, choose Redshift Manual connect instead.
- For Data source name, enter customer-analytics-redshift.
- For Database name, enter dev.
- For UserName, enter admin.
- For Password, enter the password you set when creating your Redshift Serverless namespace.
- Click Validate connection to confirm QuickSight can reach your workgroup.
- Once validated, click Create data source.
After the data source is created, QuickSight prompts you to choose a table.
- For Schema, choose public.
- Select the customer_analytics_summary table.
- Click Select.
- Choose Import to SPICE for quicker analytics, then click Visualize.
Why import to SPICE?
Importing to SPICE caches your data in QuickSight's fast in-memory engine. This means your dashboard loads instantly without spinning up Redshift compute (and consuming your free trial RPU-hours) every time you refresh a chart.
QuickSight now opens an analysis workspace with your dataset loaded.
Connection validation failed?
Make sure your Redshift Serverless workgroup has Publicly accessible enabled in its settings. Also check that the database name is dev (not sample_data_dev) since your analytics table lives in the dev database.
If you used a manual connection, enter your workgroup endpoint without the port (for example, workgroup-name.account-id.region.redshift-serverless.amazonaws.com) and set the port to 5439.
help me connect QuickSight to my Redshift Serverless workgroup
Build KPI and sales trend visualizations
Your analysis workspace starts with a blank canvas. You will add three visuals: a KPI card for total revenue, a line chart showing sales over time, and a bar chart breaking down revenue by event category.
- In the Visual types pane on the left, click the KPI icon.
- From the Fields list pane, drag total_spent to the Value field well.
- The KPI card now shows the sum of all customer spending as your total revenue metric.
What is a KPI visual?
A KPI (Key Performance Indicator) visual displays a single high-level metric prominently on your dashboard. It is the first thing stakeholders look at to understand overall business performance at a glance.
Next, add a line chart to show how sales change over time.
- Click Add on the application bar, then choose Add visual.
- In the Visual types pane, choose the Line chart icon.
- Drag purchase_count to the Value field well.
- Drag user_id to the X axis field well. QuickSight will aggregate by count.
Use the dataset fields creatively
The customer_analytics_summary table is pre-aggregated at the user level, so it does not have a date dimension. For a time-based line chart, you could create a calculated field based on purchase_count or connect a second dataset with the raw tickit.sales table that includes saletime. For this project, use user_state as the X axis dimension to show a distribution of sales metrics by state.
Now add a vertical bar chart for revenue by event category.
- Click Add on the application bar, then choose Add visual.
- In the Visual types pane, choose the Vertical bar chart icon.
- Drag preferred_category to the X axis field well.
- Drag total_spent to the Value field well.
- Your chart now shows total revenue grouped by event category (the category each user spent the most on).
You should now see three visuals on your canvas: a KPI card, a line or distribution chart, and a bar chart.
Don't see your fields in the Fields list?
Make sure the customer_analytics_summary dataset is selected in the Data pane at the top of the left panel. If you see no fields at all, your SPICE import may not have completed yet. Check the Datasets page for import status.
help me troubleshoot missing fields in my QuickSight analysis
Add customer segment visualizations and publish the dashboard
The first three visuals cover high-level performance. Now you will add two more charts that reveal customer segment insights, then publish everything as a shareable dashboard.
- Click Add on the application bar, then choose Add visual.
- In the Visual types pane, choose the Horizontal bar chart icon.
- Drag email_engagement to the Y axis field well. QuickSight will bin it as a dimension.
- Drag total_spent to the Value field well and change its aggregation to Average by clicking the field's dropdown arrow in the field well.
- This chart shows average spend broken down by engagement level, mirroring the t-test groups from your statistical analysis.
Finally, add a donut chart showing customer distribution by region.
- Click Add on the application bar, then choose Add visual.
- In the Visual types pane, choose the Donut chart icon.
- Drag user_state to the Group/Color field well.
- QuickSight automatically counts users per state and renders the donut chart segments.
Your analysis now has five visualizations covering KPIs, trends, category breakdowns, engagement segments, and regional distribution.
Now publish the analysis as a dashboard that others can view.
- In the top application bar, click Publish, then choose Publish dashboard.
- Choose Publish as a new dashboard.
- For Dashboard name, enter Customer Behavior Analytics.
- Under Select sheets, keep the default All sheets selected.
- Click Publish.
- On the Share dashboard page that appears, close it by clicking the X icon. You can share later if needed.
What is a dashboard vs. an analysis?
An analysis is your private editing workspace where you build and iterate on visuals. A dashboard is the published, read-only version that other people can view and interact with using filters and controls.
You should now see your published dashboard with all five visualizations rendered.
Dashboard shows no data?
Confirm that your SPICE import completed successfully. Navigate to Datasets in QuickSight, click your dataset, and check that the Last updated timestamp shows a recent successful import.
If the import failed, verify that the customer_analytics_summary table in Redshift actually has data by running a quick query in Query Editor v2.
help me debug my QuickSight dashboard showing no data
Your analytics pipeline is now complete from data extraction all the way to a self-service BI dashboard. Next up, take on the Secret Mission to add RFM customer segmentation and a heatmap to your dashboard.
Secret mission
RFM Customer Segmentation
Your analytics pipeline already identifies high-engagement vs low-engagement customers. But marketing teams need more granular segmentation. Build an RFM (Recency, Frequency, Monetary) scoring model that classifies every customer into actionable tiers like Champions, Loyal, At Risk, and Lost. Then visualize the distribution as a heatmap in QuickSight.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses Redshift Serverless (covered by the $300 free trial credit) and QuickSight (30-day free trial). Both will incur charges after their trial periods end.
Cost warning
Amazon Redshift Serverless charges $0.375 per RPU-hour (minimum 4 RPUs = $1.50/hour) when queries are running. Your $300 free trial credit covers 90 days of usage. After the credit is exhausted, you will be billed for active query time.
Amazon QuickSight charges $9-24/user/month after the 30-day free trial ends. If you do not unsubscribe before the trial expires, you will be charged automatically.
Resources you used:
- Amazon Redshift Serverless workgroup and namespace (with sample_data_dev and dev databases, including customer_analytics_summary and customer_rfm_segments tables).
- Amazon QuickSight subscription (datasets, analyses, and published dashboard).
- MongoDB Atlas customer_analytics database with interactions collection (50,000+ documents).
- Local customer-analytics project folder (generate_interactions.py, analytics_pipeline.py, rfm_segmentation.py).
Keep everything running
No action needed. Choose this if you are still actively exploring the dashboard or plan to extend the pipeline soon.
- Redshift Serverless only charges when queries are running. It automatically pauses when idle, so there are no compute charges while you are not querying.
- Monitor your remaining free trial credit in the Redshift Serverless console under the Free trial section of the Serverless dashboard.
- QuickSight continues counting down your 30-day trial regardless of usage. Note the date you signed up and unsubscribe before it expires if you do not want to be charged.
- MongoDB Atlas free tier has no ongoing charges.
- Your local customer-analytics folder remains on your Desktop.
Pause - I'll come back to this later
Shut down running processes to free up resources while keeping your files and data intact for later.
- Redshift Serverless automatically pauses when idle. No manual action needed. Your tables and data remain stored in the namespace.
- QuickSight trial keeps counting down even when paused. There is no way to pause the trial clock.
- MongoDB Atlas free tier data persists indefinitely with no charges.
- Your local files stay on disk until you manually delete them.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow each step below to fully clean up.
Delete the Redshift Serverless workgroup:
- In the AWS Console, search for and open Amazon Redshift.
- Select Amazon Redshift Serverless in the left menu.
- Click Workgroup configuration in the left menu.
- Select your workgroup, then click Delete Amazon Redshift Serverless instance.
- Type delete in the confirmation dialog and click Delete.
Delete the Redshift Serverless namespace:
- Click Namespace configuration in the left menu.
- Select your namespace from the list.
- Click Actions and select Delete namespace.
- Type delete in the confirmation dialog and click Delete to confirm.
Unsubscribe from Amazon QuickSight:
- Navigate to the QuickSight unsubscribe page:
- https://us-east-1.quicksight.aws.amazon.com/sn/console/unsubscribe
- Follow the on-screen prompts to confirm deletion. This removes all dashboards, analyses, datasets, and QuickSight users from your account.
Drop the MongoDB Atlas database:
- In your MongoDB Atlas console, navigate to your cluster.
- Select the Collections tab.
- Find the customer_analytics database, click the trash icon next to it, and confirm deletion.
Delete local project files:
- Remove the project folder by running this command:
cd ~/Desktop
rm -rf customer-analytics- On Windows, run this instead:
cd ~\Desktop
rmdir /s /q customer-analyticsUninstall Python packages (optional):
- Remove the project-specific packages by running:
pip uninstall redshift-connector pymongo pandas numpy scipy -yThat's a Wrap!
That's a Wrap!
Nice work! You just built an end-to-end customer behavior analytics platform that integrates data from two fundamentally different database technologies, applies statistical rigor to business questions, and delivers self-service dashboards to stakeholders.
You've learned how to:
- Build a Python ETL pipeline that extracts and merges data from Amazon Redshift and MongoDB Atlas into a unified analytics dataset with derived features like conversion rate and engagement score.
- Apply statistical hypothesis testing (independent samples t-test and Chi-Squared test) with effect sizes (Cohen's d and Cramer's V) to identify significant customer behavior patterns.
- Connect Amazon QuickSight to Redshift and build an interactive dashboard with KPIs, sales trends, and customer segment visualizations.
- Secret Mission: Implemented RFM segmentation scoring (Recency, Frequency, Monetary) and visualized customer tiers in a heatmap on your QuickSight dashboard.
Ready to quiz yourself?