DEV Community: WanjohiChristopher

Building an AI-Powered Customer Churn Prediction Pipeline on AWS (Step-by-Step)

WanjohiChristopher — Thu, 01 Jan 2026 00:56:43 +0000

Hey folks! 👋

I recently built a customer churn prediction system that not only predicts who will leave — but also explains why in plain English using Amazon Bedrock.

In this tutorial, I'll walk you through building the entire pipeline from scratch.

What we achieved:

✅ 84.2% AUC on validation data
✅ Real-time predictions via SageMaker endpoint
✅ Natural language explanations powered by Claude (Bedrock)

Let's dive in!

🎯 What We're Building

An end-to-end ML pipeline that:

Ingests customer data into S3
Trains a churn prediction model with SageMaker XGBoost
Deploys a real-time inference endpoint
Explains predictions using Amazon Bedrock (Claude)
Exposes everything via API Gateway + Lambda

Prerequisites: AWS account, basic Python knowledge

🏗️ Architecture Overview

The pipeline consists of 5 tiers:

Tier	Services	Purpose
Data Ingestion	S3	Store raw customer data
ML Training	SageMaker Training	Train XGBoost model
Model Storage	S3	Store model artifacts
Inference & AI	SageMaker Endpoint, Bedrock	Real-time predictions + NL explanations
API Layer	API Gateway, Lambda	Expose REST API

Step 1: Set Up S3 and Upload Data

First, create an S3 bucket and upload the dataset.

# Set bucket name with your account ID
export BUCKET_NAME=churn-prediction-$(aws sts get-caller-identity --query Account --output text)

# Create bucket
aws s3 mb s3://$BUCKET_NAME

# Upload your data
aws s3 cp WA_Fn-UseC_-Telco-Customer-Churn.csv s3://$BUCKET_NAME/raw/

📥 Dataset: Download the Telco Customer Churn dataset from Kaggle.

Step 2: Create SageMaker IAM Role

In AWS Console:

Go to IAM → Roles → Create role
Select SageMaker - Execution
Add policies: AmazonSageMakerFullAccess + AmazonS3FullAccess
Name it: SageMakerChurnRole

Step 3: Train the Model

Create train_churn.py:

import boto3
import sagemaker
import pandas as pd
import os
from sklearn.model_selection import train_test_split
from sagemaker.inputs import TrainingInput

# Config
BUCKET = os.environ['BUCKET_NAME']
ROLE = os.environ['ROLE_ARN']
PREFIX = 'churn-prediction'

session = sagemaker.Session()
region = session.boto_region_name

# Load and prepare data
df = pd.read_csv('WA_Fn-UseC_-Telco-Customer-Churn.csv')
df['TotalCharges'] = pd.to_numeric(df['TotalCharges'], errors='coerce').fillna(0)
df['Churn'] = (df['Churn'] == 'Yes').astype(int)

# Encode categorical columns
cat_cols = ['gender', 'Partner', 'Dependents', 'PhoneService', 'MultipleLines', 
            'InternetService', 'OnlineSecurity', 'OnlineBackup', 'DeviceProtection', 
            'TechSupport', 'StreamingTV', 'StreamingMovies', 'Contract', 
            'PaperlessBilling', 'PaymentMethod']

for col in cat_cols:
    df[col] = df[col].astype('category').cat.codes

# Features
feature_cols = ['SeniorCitizen', 'tenure', 'MonthlyCharges', 'TotalCharges'] + cat_cols
X = df[feature_cols]
y = df['Churn']

# Split and save
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

train_df = pd.concat([y_train.reset_index(drop=True), X_train.reset_index(drop=True)], axis=1)
test_df = pd.concat([y_test.reset_index(drop=True), X_test.reset_index(drop=True)], axis=1)
train_df.to_csv('train.csv', index=False, header=False)
test_df.to_csv('test.csv', index=False, header=False)

# Upload to S3
s3 = boto3.client('s3')
s3.upload_file('train.csv', BUCKET, f'{PREFIX}/train/train.csv')
s3.upload_file('test.csv', BUCKET, f'{PREFIX}/test/test.csv')

# Train XGBoost
container = sagemaker.image_uris.retrieve('xgboost', region, '1.7-1')

xgb = sagemaker.estimator.Estimator(
    image_uri=container,
    role=ROLE,
    instance_count=1,
    instance_type='ml.m5.xlarge',
    output_path=f's3://{BUCKET}/{PREFIX}/output',
    sagemaker_session=session
)

xgb.set_hyperparameters(
    objective='binary:logistic',
    num_round=100,
    max_depth=5,
    eta=0.2,
    eval_metric='auc'
)

xgb.fit({
    'train': TrainingInput(f's3://{BUCKET}/{PREFIX}/train', content_type='csv'),
    'validation': TrainingInput(f's3://{BUCKET}/{PREFIX}/test', content_type='csv')
})

# Deploy endpoint
predictor = xgb.deploy(
    initial_instance_count=1,
    instance_type='ml.t2.medium',
    endpoint_name='churn-prediction-endpoint',
    serializer=sagemaker.serializers.CSVSerializer()
)

Run it:

export BUCKET_NAME=churn-prediction-YOUR_ACCOUNT_ID
export ROLE_ARN=arn:aws:iam::YOUR_ACCOUNT_ID:role/SageMakerChurnRole
python3 train_churn.py

Training output:

2026-01-01 00:24:27 Uploading - Uploading generated training model
2026-01-01 00:24:27 Completed - Training job completed
Training seconds: 103
Billable seconds: 103

✅ Training complete!
Model artifact: s3://churn-prediction-905418352184/churn-prediction/output/sagemaker-xgboost-2026-01-01-00-22-03-339/output/model.tar.gz

Deploying endpoint (3-5 min)...
INFO:sagemaker:Creating model with name: sagemaker-xgboost-2026-01-01-00-24-53-959
INFO:sagemaker:Creating endpoint-config with name churn-prediction-endpoint
INFO:sagemaker:Creating endpoint with name churn-prediction-endpoint
---------------!
✅ Endpoint deployed: churn-prediction-endpoint
Test prediction: 0.4% churn probability

Step 4: Create Lambda with Bedrock Integration

Create a Lambda function ChurnPredictionAPI with this code:

import json
import boto3
import os

sagemaker_runtime = boto3.client('sagemaker-runtime')
bedrock = boto3.client('bedrock-runtime')

ENDPOINT_NAME = os.environ.get('SAGEMAKER_ENDPOINT', 'churn-prediction-endpoint')

def lambda_handler(event, context):
    body = json.loads(event['body']) if isinstance(event.get('body'), str) else event

    # Get prediction from SageMaker
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName=ENDPOINT_NAME,
        ContentType='text/csv',
        Body=body['features']
    )

    churn_prob = float(response['Body'].read().decode())

    # Generate explanation with Bedrock Claude
    prompt = f"""A customer has {churn_prob:.1%} churn probability.
Customer: Tenure {body.get('tenure', 'N/A')} months, ${body.get('monthly_charges', 'N/A')}/month, {body.get('contract', 'N/A')} contract.
In 2 sentences, explain the risk and suggest one retention action."""

    bedrock_response = bedrock.invoke_model(
        modelId='anthropic.claude-3-haiku-20240307-v1:0',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 100,
            "messages": [{"role": "user", "content": prompt}]
        })
    )

    explanation = json.loads(bedrock_response['body'].read())['content'][0]['text']
    risk = "High" if churn_prob > 0.7 else "Medium" if churn_prob > 0.4 else "Low"

    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({
            'churn_probability': f"{churn_prob:.1%}",
            'risk_level': risk,
            'explanation': explanation
        })
    }

Lambda configuration:

Runtime: Python 3.11
Timeout: 30 seconds
Role: LambdaChurnRole (with SageMaker + Bedrock permissions)
Environment variable: SAGEMAKER_ENDPOINT=churn-prediction-endpoint

Step 5: Create API Gateway

Create an HTTP API in API Gateway
Add Lambda integration → ChurnPredictionAPI
Create POST route: /predict
Deploy and get your invoke URL

🧪 Test the API

curl -X POST "https://YOUR_API_URL/predict" \
  -H "Content-Type: application/json" \
  -d '{
    "features": "0,24,65.5,1500.0,1,0,1,2,0,0,1,1,0,0,1,0,2,1,1",
    "tenure": 24,
    "monthly_charges": 65.5,
    "contract": "Month-to-month"
  }'

Response:


(.venv) server@DLG lambda_package % curl -X POST "https://jxairjovmi.execute-api.us-east-1.amazonaws.com/predict" \
  -H "Content-Type: application/json" \
  -d '{
    "features": "0,24,65.5,1500.0,1,0,1,2,0,0,1,1,0,0,1,0,2,1,1",
    "tenure": 24,
    "monthly_charges": 65.5,
    "contract": "Month-to-month"
  }'

{"churn_probability": "0.6%", "risk_level": "Low", "explanation": "The customer's high churn probability of 0.6% and the month-to-month contract indicate a significant risk of losing the customer. To mitigate this risk, a retention action could be to offer the customer a longer-term contract with a discounted monthly rate or additional benefits, which may help increase their loyalty and reduce the likelihood of churn."}%

🧹 Cleanup

Don't forget to delete resources to avoid charges:

# Delete SageMaker endpoint (most expensive!)
aws sagemaker delete-endpoint --endpoint-name churn-prediction-endpoint
aws sagemaker delete-endpoint-config --endpoint-config-name churn-prediction-endpoint

# Delete Lambda
aws lambda delete-function --function-name ChurnPredictionAPI

# Delete S3 bucket
aws s3 rb s3://$BUCKET_NAME --force

💡 Key Lessons Learned

SageMaker XGBoost is production-ready — achieved 84% AUC with minimal tuning.
Bedrock adds real business value — converting predictions to actionable insights makes ML accessible to non-technical stakeholders.
IAM permissions are tricky — create roles via Console if CLI gives explicit deny errors.
Cost awareness matters — always delete endpoints when not in use (~$0.05/hour adds up!)

Resources

Thanks for reading! If this helped you, follow me for more AWS + Data Engineering content.

Questions? Leave a comment below!

𝗩𝗼𝗶𝗰𝗲 𝗔𝗜: 𝗧𝗧𝗦 - 𝗚𝗶𝘃𝗶𝗻𝗴 𝗬𝗼𝘂𝗿 𝗔𝗜 𝗮 𝗩𝗼𝗶𝗰𝗲

WanjohiChristopher — Tue, 23 Dec 2025 13:45:00 +0000

We've covered how Voice AI listens (ASR), understands (NLU), decides (Dialog Management), remembers (Context), and writes (NLG).

Now for the final piece: 🔊 Making it speak.

That's TTS - Text-to-Speech.

𝗧𝗵𝗲 𝗧𝗿𝗮𝗻𝘀𝗳𝗼𝗿𝗺𝗮𝘁𝗶𝗼𝗻:
Input: "Great news! Your flight to Paris is confirmed."
Output: 〰️〰️〰️ (audio waveform).

𝗧𝗵𝗲 𝗧𝗧𝗦 𝗣𝗶𝗽𝗲𝗹𝗶𝗻𝗲:
1️⃣ 𝗧𝗲𝘅𝘁 𝗔𝗻𝗮𝗹𝘆𝘀𝗶𝘀
• "How to pronounce this?"
• Normalization ($50 → "fifty dollars")
• Grapheme-to-phoneme conversion
• Homograph resolution (read vs read)
2️⃣ 𝗣𝗿𝗼𝘀𝗼𝗱𝘆 𝗣𝗿𝗲𝗱𝗶𝗰𝘁𝗶𝗼𝗻
• How should it sound?
• Pitch contour (intonation)
• Duration (speed)
• Stress & emphasis
• Pauses
3️⃣ 𝗔𝗰𝗼𝘂𝘀𝘁𝗶𝗰 𝗠𝗼𝗱𝗲𝗹
• Generate mel spectrogram.
• Tacotron 2, FastSpeech 2, VITS.
• Maps phonemes → audio features.
4️⃣ 𝗩𝗼𝗰𝗼𝗱𝗲𝗿
• Convert to audio waveform.
• HiFi-GAN, WaveGlow, WaveNet.
• Spectrogram → actual audio.

🎯 And that closes the loop:
Listen → Think → Speak

That’s the full Voice AI pipeline.

Thanks for following along - next, I'll likely recap the full system and share a few real-world failure modes that make or break Voice AI in production. More coming soon. Keep building!!

Cheers!!

𝗩𝗼𝗶𝗰𝗲 𝗔𝗜: 𝗡𝗟𝗚 - 𝗧𝘂𝗿𝗻𝗶𝗻𝗴 𝗗𝗲𝗰𝗶𝘀𝗶𝗼𝗻𝘀 𝗜𝗻𝘁𝗼 𝗪𝗼𝗿𝗱𝘀

WanjohiChristopher — Mon, 22 Dec 2025 13:43:00 +0000

Voice AI listens (ASR), understands (NLU), and decides (Dialog Management).

But decisions aren't responses.
The system knows:
▶️ Action: inform
▶️ Flight: booked
▶️ Destination: Paris
▶️ Date: Dec 20
▶️ Confirmation: AB123

That's not what we say to a user.

This is where 𝗡𝗟𝗚 (Natural Language Generation) comes in.

It transforms structured data into natural speech:
Example:
🤖 "Great news! Your flight to Paris on December 20th is confirmed. Your confirmation number is AB123. Have a wonderful trip!"

𝗧𝗵𝗲 𝗡𝗟𝗚 𝗣𝗶𝗽𝗲𝗹𝗶𝗻𝗲:
1️⃣ 𝗖𝗼𝗻𝘁𝗲𝗻𝘁 𝗣𝗹𝗮𝗻𝗻𝗶𝗻𝗴
🔹"What information to convey?"
🔹Select facts, order them, prioritize.
2️⃣ 𝗦𝗲𝗻𝘁𝗲𝗻𝗰𝗲 𝗣𝗹𝗮𝗻𝗻𝗶𝗻𝗴
🔹"How to structure it?"
🔹One sentence or multiple?
🔹Combine facts?
3️⃣ 𝗦𝘂𝗿𝗳𝗮𝗰𝗲 𝗥𝗲𝗮𝗹𝗶𝘇𝗮𝘁𝗶𝗼𝗻
🔹"What exact words to use?" .
🔹Grammar, vocabulary, tone, fluency.

𝗧𝗵𝗲 𝗲𝘃𝗼𝗹𝘂𝘁𝗶𝗼𝗻:
🔹Templates → slot-filling.
🔹Statistical → n-grams, HMMs.
🔹Neural → Seq2Seq, Transformers.
🔹LLMs → GPT, Claude (SOTA) .
Below are 𝗿𝗲𝗰𝗼𝗺𝗺𝗲𝗻𝗱𝗮𝘁𝗶𝗼𝗻s based on use case:
🔹Need predictability → Templates.
🔹Need natural variety → LLM.
🔹Need both → Hybrid (LLM + guardrails).

The difference between a robotic assistant and a delightful one? NLG.

𝗩𝗼𝗶𝗰𝗲 𝗔𝗜: 𝗖𝗼𝗻𝘁𝗲𝘅𝘁 & 𝗠𝗲𝗺𝗼𝗿𝘆 - 𝗪𝗵𝘆 𝗖𝗼𝗻𝘃𝗲𝗿𝘀𝗮𝘁𝗶𝗼𝗻𝘀 𝗗𝗼𝗻'𝘁 𝗥𝗲𝘀𝗲𝘁

WanjohiChristopher — Sun, 21 Dec 2025 14:40:00 +0000

Dialog Management means = deciding what to do next.

But something else makes Voice AI feel human instead of robotic:

🧠 Context and memory.

𝗪𝗵𝘆 𝗰𝗼𝗻𝘁𝗲𝘅𝘁 𝗺𝗮𝘁𝘁𝗲𝗿𝘀
Consider this exchange:
🗣️ "Book me a flight to Paris."
🗣️ "Make it business class."
That second sentence only makes sense if the system remembers the first.
That's context.
𝗪𝗵𝗮𝘁 𝗰𝗼𝗻𝘁𝗲𝘅𝘁 & 𝗺𝗲𝗺𝗼𝗿𝘆 𝗮𝗰𝘁𝘂𝗮𝗹𝗹𝘆 𝗶𝗻𝗰𝗹𝘂𝗱𝗲:
→ 𝗦𝗵𝗼𝗿𝘁-𝘁𝗲𝗿𝗺 𝗰𝗼𝗻𝘁𝗲𝘅𝘁 (session memory)
🔹Recent turns.
🔹Slot values.
🔹Corrections.
🔹Current dialog state.
→ 𝗟𝗼𝗻𝗴-𝘁𝗲𝗿𝗺 𝗺𝗲𝗺𝗼𝗿𝘆
🔹User preferences.
🔹Past interactions.
🔹Frequent locations.
🔹Knowledge (RAG documents).
This information feeds directly into Dialog Management so the system can make better decisions.

Without memory, every interaction would feel like the first one.

LLMs can reason - but the architecture decides what to remember, when to retrieve it, and when to forget.

That balance is what makes Voice AI feel natural and safe.

𝗩𝗼𝗶𝗰𝗲 𝗔𝗜: 𝗗𝗶𝗮𝗹𝗼𝗴 𝗠𝗮𝗻𝗮𝗴𝗲𝗺𝗲𝗻𝘁 - 𝗧𝗵𝗲 𝗢𝗿𝗰𝗵𝗲𝘀𝘁𝗿𝗮𝘁𝗼𝗿

WanjohiChristopher — Sat, 20 Dec 2025 16:07:00 +0000

We've talked about how Voice AI listens (ASR) and understands (NLU).

But once the system understands the user, there's a harder question:
👉 What should happen next?

This is where Dialog Management comes in.

It's not about generating responses - it's about orchestrating decisions across multiple turns.

E𝘅𝗮𝗺𝗽𝗹𝗲:
👤 "Book a flight to Paris"
🤖 [dest: Paris, origin: ❓] → "Where from?"
👤 "New York"
🤖 [all slots filled ] → "NYC → Paris. Confirm?"

That decision flow? That's Dialog Management.

𝗨𝗻𝗱𝗲𝗿 𝘁𝗵𝗲 𝗵𝗼𝗼𝗱, 𝗶𝘁 𝗵𝗮𝗻𝗱𝗹𝗲𝘀:
→ Tracking conversation state across turns.
→ Knowing what's been said vs what's missing.
→ Deciding when to ask vs when to act.
→ Handling corrections and errors.
→ Executing actions and tools safely.

This is what turns one-shot commands (from the user) into real conversations.

Modern Voice AI agents may use LLMs here - but structure is still essential for reliability and safety.

Without dialog management, even the best models feel unpredictable.

➡️ Next up: How Voice AI remembers - context & memory management.

𝗩𝗼𝗶𝗰𝗲 𝗔𝗜: 𝗡𝗟𝗨 (𝗡𝗮𝘁𝘂𝗿𝗮𝗹 𝗟𝗮𝗻𝗴𝘂𝗮𝗴𝗲 𝗨𝗻𝗱𝗲𝗿𝘀𝘁𝗮𝗻𝗱𝗶𝗻𝗴) - 𝗨𝗻𝗱𝗲𝗿𝘀𝘁𝗮𝗻𝗱𝗶𝗻𝗴 𝗪𝗵𝗮𝘁 𝗬𝗼𝘂 𝗔𝗰𝘁𝘂𝗮𝗹𝗹𝘆 𝗠𝗲𝗮𝗻𝘁

WanjohiChristopher — Fri, 19 Dec 2025 13:33:00 +0000

What happens after the text arrives from ASR.

🗣️ Say you tell a voice assistant:
"Book me a flight to Paris next Friday"

ASR does its job and converts that into text.

But at this point, the system still doesn’t really understand anything.
It doesn’t know:
🔹what you’re trying to do.
🔹which parts of the sentence matter.
🔹or what information is missing.

That’s where NLU (Natural Language Understanding) comes in.

Here’s what NLU figures out behind the scenes:

1️⃣ - 𝗜𝗻𝘁𝗲𝗻𝘁 𝗖𝗹𝗮𝘀𝘀𝗶𝗳𝗶𝗰𝗮𝘁𝗶𝗼𝗻
What are you trying to do?
→ You want to book a flight.

2️⃣ - 𝗘𝗻𝘁𝗶𝘁𝘆 𝗘𝘅𝘁𝗿𝗮𝗰𝘁𝗶𝗼𝗻- details (entities)
→ destination: Paris
→ date: next Friday

3️⃣ And finally - 𝗦𝗹𝗼𝘁 𝗙𝗶𝗹𝗹𝗶𝗻𝗴 - what’s missing
→ where are you flying from?

So the system knows it needs to ask a follow-up.

That's the moment where the conversation starts to feel natural instead of scripted.

With models like GPT-4 or Claude, etc, a lot of this NLU work can now happen in one step without training separate intent classifiers or entity models. The model reasons about intent, details, and gaps together.

That’s a big reason modern Voice AI agents feel more flexible than the older "say it exactly this way" systems.

ASR (Automatic Speech Recognition)

WanjohiChristopher — Thu, 18 Dec 2025 22:30:00 +0000

Yesterday I shared the full Voice AI pipeline.
Today we're diving deep into Stage 1: ASR (Automatic Speech Recognition).

You speak → It becomes text.

Simple, right? Here's what actually happens:

𝟭. 𝗙𝗲𝗮𝘁𝘂𝗿𝗲 𝗘𝘅𝘁𝗿𝗮𝗰𝘁𝗶𝗼𝗻
Raw audio → Digital representation

MFCCs (Mel-Frequency Cepstral Coefficients)
Spectrograms
Filter Banks

𝟮. 𝗔𝗰𝗼𝘂𝘀𝘁𝗶𝗰 𝗠𝗼𝗱𝗲𝗹𝗶𝗻𝗴
Maps audio features to phonemes

Traditional: HMM-GMM, DNN-HMM
Modern: Transformers, Conformers

𝟯. 𝗗𝗲𝗰𝗼𝗱𝗶𝗻𝗴 & 𝗟𝗮𝗻𝗴𝘂𝗮𝗴𝗲 𝗠𝗼𝗱𝗲𝗹𝗶𝗻𝗴
Phonemes → Words using probabilities

Beam Search, CTC, Attention mechanisms

𝟰. 𝗣𝗼𝘀𝘁-𝗣𝗿𝗼𝗰𝗲𝘀𝘀𝗶𝗻𝗴
Clean up the output

Spell checking, punctuation, capitalization

The evolution has been wild:

𝗧𝗿𝗮𝗱𝗶𝘁𝗶𝗼𝗻𝗮𝗹 (1980s-2010s):
→ HMM + GMM
→ Required phonetic alignment
→ Separate components stitched together

𝗦𝗧𝗔𝗧𝗘-𝗢𝗙-𝗧𝗛𝗘-𝗔𝗥𝗧 (Now):
→ Whisper: 680K hours of training, 50+ languages
→ Wav2Vec 2.0: Self-supervised, works with limited data

Get ASR wrong and your entire voice pipeline fails. It's the foundation.

I've attached a diagram breaking down the full ASR architecture.

What ASR model are you using? Any surprises with accuracy or latency?

VOICE AI SYSTEM ARCHITECTURE

WanjohiChristopher — Thu, 18 Dec 2025 04:22:41 +0000

🎙️I’ve been diving deep into Voice AI Agents and decided to map out how they actually work.

You know when you ask Alexa or ChatGPT Voice a question and it just… responds intelligently?

There’s a lot happening in that split second.

How do voice agents work?

At a high level, every voice agent needs to handle three tasks:

👉Listen - capture audio and transcribe it
👉Think - interpret intent, reason, plan
👉Speak - generate audio and stream it back to the user

A Voice AI Agent typically goes through five core stages:
🔹Speech is converted to text (ASR).
🔹The system understands intent and entities (NLU).
🔹It reasons about what action to take (Dialog Manager / Agent Logic).
🔹It generates a response (NLG).
🔹Speaks it back naturally (TTS).

This same agent-style architecture powers Alexa, Siri, Google Assistant, and modern LLM-based voice agents like ChatGPT Voice.

I put together a diagram to visualize the full end-to-end pipeline behind Voice AI Agents - from speech input to intelligent action and response.

I’m planning to break down each component and share more on how agent-based voice systems are built.

Which Voice AI agent do you interact with the most?

Building a Production-Ready Data Pipeline on AWS: A Hands-On Guide for Data Engineers

WanjohiChristopher — Mon, 01 Dec 2025 04:39:37 +0000

Introduction

Modern data engineering requires scalable, fault-tolerant, and secure architectures. In this article, I walk through a fully operational AWS data pipeline using S3, Kinesis, Glue, Athena, Redshift, and QuickSight.
Everything here is hands-on — every step can be reproduced in your own AWS console, and I will include the exact screenshots from my implementation.

This article helps anyone learn:

How to build a real AWS ETL pipeline end-to-end
How to combine batch + streaming data
How to orchestrate jobs with Glue + Lambda
How to query data with Athena and Redshift
How to build dashboards with QuickSight

Architecture Overview
We will build this architecture:

Architecture Components

Amazon S3 – Data lake (Raw → Clean → Analytics Zones)
Amazon Kinesis Data Streams – Real-time data ingestion
AWS Glue – ETL jobs & data catalog
AWS Lambda – Event-driven transformations
Amazon Athena – Serverless SQL analytics
Amazon Redshift – Data warehouse
Amazon QuickSight – Dashboards
AWS Lake Formation – Governance
Amazon CloudWatch – Monitoring

Step 1: Create the Data Lake on Amazon S3

Create an S3 bucket like - your-datalake-name

/raw/
/clean/
/analytics/

Step 2: Ingest Real-Time Data using Kinesis

We simulate a real IoT or clickstream pipeline.

Step:

Go to Kinesis → Data Streams
Click Create Stream
Name it:

datalyte_stream

4.Set number of shards = 1

✔ Monitoring graph of incoming data

Step 3: Create AWS Glue Crawler (Schema Discovery)

We want Glue to:

Crawl S3 raw zone

Detect schema

Create database + tables in Glue Catalog

Steps:

Go to AWS Glue → Crawlers
Click Create Crawler
Name:

datalyte_raw_crawler

Choose S3 input: s3://ddatalyte-data-lake-cmp/raw/
Create a new Glue database:

datalyte_db

Glue Crawler setup

Running Crawler

Step 4: Build ETL Transformations with AWS Glue (Spark)

We transform data from raw → clean.

Steps:

Go to Glue → Jobs
Click Create Job
Choose Spark / Python
Script:

import pyspark.sql.functions as F

df = glueContext.create_dynamic_frame.from_catalog(
    database="datalyte_db",
    table_name="raw_data"
).toDF()

clean_df = df.withColumn("processed_at", F.current_timestamp())

clean_dyf = DynamicFrame.fromDF(clean_df, glueContext, "clean_dyf")

glueContext.write_dynamic_frame.from_options(
    frame=clean_dyf,
    connection_type="s3",
    connection_options={"path": "s3://datalyte-data-lake/clean/"},
    format="parquet"
)

Step 5: Query Clean Data using Amazon Athena

Go to Athena
Select database: datalyte_db
Run this query:

SELECT *
FROM clean_data
LIMIT 20;

Step 6: Load Analytics Data into Redshift

Now we push curated data into Redshift for BI.

Steps:
1.Create a Redshift Serverless Workspace
2.Create a schema:

CREATE SCHEMA datalyte;

3.Load S3 data into Redshift:

COPY datalyte.analytics_table
FROM 's3://datalyte-data-lake/analytics/'
IAM_ROLE '<your_redshift_role>'
FORMAT AS PARQUET;

Step 7: Build a Dashboard with QuickSight

Connect QuickSight to Redshift
Import analytics table
Create:

Line charts
Bar charts
KPIs (metrics)

Conclusion

This hands-on project demonstrates how AWS services like S3, Glue, Kinesis, Athena, and Redshift come together to form a complete, production-ready data pipeline. With this foundation in place, you can now extend the architecture to include BI, machine learning, or more advanced data engineering workflows on AWS.

Building a Real-Time Data Lake on AWS: S3, Glue, and Athena in Production

WanjohiChristopher — Mon, 24 Nov 2025 04:57:35 +0000

The 3 AM Wake-Up Call
It was a Tuesday morning at AWS Reinvent 2024 Las Vegas, when my phone lit up with a PagerDuty alert. Our analytics dashboard had timed out, and the culprit was an Athena query scanning 47 TB of S3 data-costing us $235 in a single failed attempt. The query was looking for a week's worth of user events, but due to poor partitioning, Athena had to scan three years of historical data.

That incident taught me an expensive lesson: a well-architected data lake isn't just about storing data cheaply in S3. It's about making that data queryable, maintainable, and cost-effective at scale.
Over four years of building data platforms on AWS, I've learned that the difference between a $500/month data lake and a $15,000/month one often comes down to a handful of architectural decisions. In this article, I'll share the battle-tested patterns that have helped me and the teams I've worked with build production-grade data lakes that are both performant and economical.

Table of Contents

Architecture foundations: The multi-zone approach that separates concerns
Partitioning strategies: The single biggest lever for query performance and cost
Schema evolution: How to change schemas without breaking downstream systems
Query optimization: Techniques that reduced our query times by 85%
Cost optimization: Real tactics that saved thousands per month.

Architecture Overview: The Three-Zone Data Lake

The foundation of a production data lake is separation of concerns. I've found the three-zone architecture to be the most practical approach:

Raw Zone (Bronze):

Stores data exactly as received from source systems
No transformations, no schema enforcement
Serves as the "source of truth" for reprocessing Example path: s3://datalake-raw/source_system/table_name/year=2024/month=11/day=23/

Processed Zone (Silver):

Cleaned, deduplicated, and validated data
Type conversions applied (strings to proper data types)
Bad records filtered or quarantined Example path: s3://datalake-processed/domain/table_name/year=2024/month=11/day=23/

Curated Zone (Gold):

Business-level aggregations and joins
Optimized for specific analytics use cases
Often denormalized for query performance Example path: s3://datalake-curated/analytics/user_activity_summary/year=2024/month=11/

The Glue Catalog:

AWS Glue Catalog sits at the center, acting as a centralized metadata repository:

Stores table schemas, partition information, and statistics
Shared across Athena, Glue ETL jobs, Redshift Spectrum, and EMR
Enables schema-on-read: define structure at query time, not ingestion time

Typical Data Flow
Ingestion: Data lands in Raw Zone (JSON, CSV, or streaming via Kinesis)
Processing: Glue ETL job reads from Raw, transforms, writes Parquet to Processed Zone
*Curation: * Another Glue job aggregates and joins data, writes to Curated Zone.

Query: Athena queries the Curated Zone for fast, cost-effective analytics.

Partitioning Strategies: Biggest Cost Lever

Please note proper partitioning is the difference between a $5 query and a $500 query.
Here's what I've learned works.
Time-Based Partitioning (The Foundation)
For time-series data (logs, events, transactions), partition by date:

CREATE EXTERNAL TABLE user_events (
    user_id STRING,
    event_type STRING,
    timestamp BIGINT,
    properties STRING
)
PARTITIONED BY (
    year STRING,
    month STRING,
    day STRING
)
STORED AS PARQUET
LOCATION 's3://datalake-processed/events/user_events/';

Why year/month/day separately? Flexibility.
You can query:

A single day: WHERE year='2024' AND month='11' AND day='23'
An entire month: WHERE year='2024' AND month='11'
Multiple months: WHERE year='2024' AND month IN ('10','11')

Partition Projection: The Game Changer
Without partition projection, you need to run MSCK REPAIR TABLE or ALTER TABLE ADD PARTITION every time new data arrives. With projection, Athena generates partition values automatically:

CREATE EXTERNAL TABLE user_events (
    user_id STRING,
    event_type STRING,
    properties STRING
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET
LOCATION 's3://datalake-processed/events/user_events/'
TBLPROPERTIES (
    'projection.enabled' = 'true',
    'projection.dt.type' = 'date',
    'projection.dt.range' = '2023-01-01,NOW',
    'projection.dt.format' = 'yyyy-MM-dd',
    'storage.location.template' = 's3://datalake-processed/events/user_events/dt=${dt}'
);

Real impact: This eliminated our nightly partition maintenance job and reduced query planning time from 10 seconds to sub-second.

Schema Evolution: Adding Columns Without Breaking Everything

The cardinal rule: append columns to the end, never remove or reorder.

Adding a new column (Safe):

ALTER TABLE user_events 
ADD COLUMNS (device_type STRING);

Old Parquet files without this column? No problem. Athena returns NULL for missing columns.

Changing data types (Don't Do this):

-- DON'T DO THIS:
ALTER TABLE user_events 
CHANGE COLUMN user_id user_id BIGINT;  -- Was STRING

If old files have non-numeric user IDs, queries will fail. Instead: add a new column (user_id_v2) and backfill gradually.

The Golden Rules

Add, don't modify - New columns are safe, changing existing ones is dangerous.
Make new columns nullable - Old data won't have values.
Use separate tables for breaking changes - user_events_v2 is better than breaking user_events.
Document schema versions - Add version info in table properties or a separate registry.

File Formats & Query Optimization: Speed and Cost in Harmony
Why Parquet Wins
Example:

100 GB of JSON logs -> 12 GB as Parquet with Snappy compression
Athena query cost: $5.00 (JSON) vs $0.60 (Parquet)

Parquet's columnar storage means you only read the columns you need, not entire rows.
File Size Matters
Sweet spot: 128 MB - 512 MB per file
Too many small files (< 10 MB)? Athena spends more time listing files than reading them. Compact them with a simple Glue job:

df = spark.read.parquet("s3://path/to/small/files/")
df.repartition(10).write.mode("overwrite").parquet("s3://path/to/compacted/")

Three Query Wins
Select only what you need

-- Scans 10 columns
SELECT * FROM user_events WHERE year='2024';

-- Scans 3 columns (83% cheaper)
SELECT user_id, event_type, timestamp FROM user_events WHERE year='2024';

2. Use CTAS for repeated aggregations
Running the same complex query daily? Materialize it once:

CREATE TABLE daily_summary
WITH (format='PARQUET', partitioned_by=ARRAY['dt']) AS
SELECT dt, user_id, COUNT(*) as events
FROM user_events GROUP BY dt, user_id;

3. Enable query result caching in your Athena Workgroup - Identical queries are free for 24 hours.

Top 5 Cost Optimization Wins
Here are the tactics that saved the most money in my projects:

Partition Projection → Eliminated 80% of "full table scans"

Saved: ~$2,000/month on a moderate data lake

Convert to Parquet → 10x reduction in data scanned

Saved: ~$3,500/month (migrated 500 GB daily ingestion from JSON)

S3 Lifecycle Policies → Move old data to cheaper storage

{
  "Rules": [{
    "Id": "Archive old partitions",
    "Status": "Enabled",
    "Transitions": [{
      "Days": 90,
      "StorageClass": "GLACIER_IR"
    }]
  }]
}

Saved: ~$800/month on storage costs

4. Athena Workgroup Data Scan Limits → Prevent runaway queries

Set per-query limit: 1 TB
Set per-workgroup monthly limit: 50 TB

Glue Job Bookmarks → Process only new data

job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Only processes new files since last run
datasource = glueContext.create_dynamic_frame.from_catalog(
    database="my_database",
    table_name="raw_events",
    transformation_ctx="datasource"
)

*Conclusion: *

Partition by time, always. Use partition projection.
Parquet isn't optional—it's essential for cost control.
Schema evolution requires discipline: add columns, don't modify them.
Monitor data scanned per query. It's your cost meter.

Questions from your own data lake implementations? I'd love to hear them.

RDS MySQL Zero ETL Integration with Redshift

WanjohiChristopher — Tue, 17 Dec 2024 17:30:11 +0000

Introduction:

Data integration is one of the most critical aspects of any data-driven organization.ETL, which stands for Extraction, Transformation, and Load, involves the extraction of data from various sources, transforming it to meet specific requirements, and then loading it into a target destination.Traditionally, integrating databases like Amazon RDS MySQL with Redshift for analytical workloads required manual ETL (Extract, Transform, Load) processes. However AWS has now this integration eliminates the manual complexity and time-consuming Extract, Transform, and Load (ETL) processes, offering a streamlined and near real-time solution for moving data from RDS MySQL to Redshift.

In this article, we will explore what RDS MySQL Zero ETL integration with Redshift is, how it works, and its benefits for developers and data engineers.

Zero ETL integrations make it simpler to analyze data from Amazon RDS to Redshift by removing the need for anyone to manage complex data pipelines.
AWS handles the data replication, transformation, and loading seamlessly, ensuring data in Redshift is continuously updated as changes occur in the source RDS database.

Benefits of Zero ETL Integration

Near real-Time Analytics.
Reduced Operational Overhead.
Scalability.
Cost-efficiency
Improved Data Consistency.

Lets Dive into process, we will do a simple example:

Step 1: Setup the RDS MySQL database with data.

Step 2: Setup the datawarehouse -Redshift in this case.

Step 3: Create a Zero ETL Integration.

In conclusion:
Amazon RDS MySQL Zero ETL integration with Amazon Redshift is a game-changer for organizations looking to simplify their data pipelines and enable real-time analytics. By automating the data movement process, AWS eliminates the complexities of traditional ETL workflows, reducing operational overhead, improving data consistency, and enabling timely insights.

AWS Database Migration-Heterogenous Approach using DMS.

WanjohiChristopher — Thu, 15 Feb 2024 02:12:56 +0000

Introduction

Ever wondered how you can migrate your database (data) from OnPrem to AWS Cloud, AWS RDS to AWS RDS or to S3 Bucket?
Then this article is for you.
We will delve into heterogeneous migration.

Define whats data migration?
Data Migration strategies.
Define heterogeneous migration.
We will use Mysql and Postgres as our databases for the use case.

Data Migration

Data migration is the movement of data from source systems to target systems (destination systems), depending on the environment in which the data is sitting.
For example, a database, say Postgres, can be running in an EC2 instance, and the client wants to migrate the database to the AWS Relational Database System(Aurora Postgres).In this case client will need help in knowing how they want to migrate to their preferred database. Thats where these services comein.

Data Migration Strategies

These are approaches used to migrate workloads to the AWS Cloud.
I will just mention all the approaches here only.

Retire
Retain
Rehost
Relocate
Repurchase
Replatform
Refactor or re-architect

Heterogenous Migration

When we talk about heterogenous migration, we talk about migrating data from totally different databases.
Meaning that Source database has completely different schema compared to the target database.
For example, from MySQL to Postgres or vice versa or from SQL Server to Postgres or from Posgres to Oracle.

We will first step into steps of migration:

Review Data Sources by checking the environment,schema,tables etc.

2.Determine the destination
We access where the organization wants its data to sit.3.

3.Data Migration strategy
At this point, we focus on creating a migration plan so that we understand clearly the requirements for the migration.

We will be able to:

assess security requirements.
Cost and time to be taken.
systems and data migration tools.
Implement.
Test the connections.
Run the migrations.

Our case, we will illustrate the process of migrating a AWS MySQL database to AWS Postgres (RDS).We will cover the demo in our next article.

Process

Create the source and target databases and insert data in the source schema(MYSQL).
We will be using AWS Data Migration Service and Schema Conversion Tool (AWS SCT) which we will download. 3.The first step here will be Performing Schema conversion.We convert the schema of mysql database to a PostgreSQL schema using AWS SCT.

In order to convert we need to connect our database endpoints in AWS SCT Tool locally.
Please see the below preview.

4.Navigate now to AWS Data Migration Service to create endpoints source and target,create replication instance and data migration tasks. At this point we wont focus on using serverless and also Data Migration projects we save them for later.
After creating the migration task we perform full load as shown below.

At this point we can confirm in our data whether it was migrated and successful.We can see we are successful;

Conclusion

In conclusion, in this article, we covered data migration, strategies, a plan, and heterogenous migration. You realise we didn't go too technical, as we plan to have the next article contain full technical details explaining each process from connection using endpoints and IP addresses to creating migration tasks one by one. I hope you enjoyed.
Cheers! and Happy Learning.