Case Study: Elena's Multi-Source Data Pipeline
The Scenario
Elena Vasquez has been automating reports at her nonprofit for months now. Her Python scripts can read files, process data, and generate summaries. But her boss just dropped a new challenge on her desk.
"Elena, the board wants a quarterly impact report. We need to combine data from three different systems: our donor database exports to CSV, our program tracking tool produces JSON reports, and we want to add geographic data so we can show donors on a map. Can you pull it all together?"
Three data sources. Three different formats. One unified report. This is Elena's first real data pipeline.
The Data Sources
Source 1: Donor CSV (from the CRM)
The nonprofit's donor management system exports a CSV file every quarter:
donor_id,name,email,zip,amount,date,campaign
D001,"Martinez, Sofia",sofia.m@email.com,97201,250.00,2025-01-15,annual
D002,Chen Wei,wei.chen@email.com,98101,500.00,2025-01-22,gala
D003,"O'Brien, James",jobrien@email.com,97202,150.00,2025-02-03,annual
D004,Aisha Johnson,aisha.j@email.com,94102,1000.00,2025-02-14,gala
D005,Tom Williams,twill@email.com,97201,75.00,2025-02-28,annual
D006,Lisa Park,lpark@email.com,98101,350.00,2025-03-10,spring
D007,"Davis, Marcus",mdavis@email.com,94102,200.00,2025-03-15,spring
D008,Priya Sharma,psharma@email.com,97202,425.00,2025-03-22,gala
Notice the commas in some names ("Martinez, Sofia", "O'Brien, James") — these are properly quoted in the CSV. Elena's code needs to handle this correctly.
Source 2: Program Metrics JSON (from the dashboard)
The program tracking tool produces a JSON file with metrics for each program:
{
"quarter": "Q1-2025",
"programs": [
{
"name": "Youth Mentoring",
"participants": 145,
"sessions_held": 48,
"satisfaction_score": 4.7,
"budget_used": 42000
},
{
"name": "Job Training",
"participants": 89,
"sessions_held": 36,
"satisfaction_score": 4.5,
"budget_used": 38000
},
{
"name": "Community Garden",
"participants": 210,
"sessions_held": 24,
"satisfaction_score": 4.8,
"budget_used": 15000
}
],
"total_volunteers": 67,
"total_volunteer_hours": 2840
}
Source 3: Geographic Enrichment (simulated API)
Elena wants to add city and state information to each donation based on the donor's zip code. In production, she'd use a geocoding API. For this report, she builds a lookup dictionary:
zip_lookup = {
"97201": {"city": "Portland", "state": "OR", "region": "Pacific NW"},
"97202": {"city": "Portland", "state": "OR", "region": "Pacific NW"},
"98101": {"city": "Seattle", "state": "WA", "region": "Pacific NW"},
"94102": {"city": "San Francisco", "state": "CA", "region": "West Coast"},
}
Elena's Pipeline
Elena breaks the problem into the five stages of a data pipeline:
Stage 1: Load
import csv
import json
from pathlib import Path
def load_donations(csv_path: str) -> list[dict]:
"""Load donation records from the CRM export."""
donations = []
with open(csv_path, "r", newline="") as f:
reader = csv.DictReader(f)
for row in reader:
# Convert amount from string to float immediately
row["amount"] = float(row["amount"])
donations.append(row)
print(f" Loaded {len(donations)} donations from CSV")
return donations
def load_metrics(json_path: str) -> dict:
"""Load program metrics from the dashboard export."""
with open(json_path, "r") as f:
metrics = json.load(f)
program_count = len(metrics.get("programs", []))
print(f" Loaded metrics for {program_count} programs from JSON")
return metrics
Stage 2: Clean
def clean_donations(donations: list[dict]) -> list[dict]:
"""Validate and clean donation records."""
cleaned = []
skipped = 0
for d in donations:
# Skip records with invalid amounts
if d["amount"] <= 0:
skipped += 1
continue
# Normalize campaign names to lowercase
d["campaign"] = d["campaign"].strip().lower()
# Ensure email is present
if not d.get("email"):
d["email"] = "unknown@nonprofit.org"
cleaned.append(d)
if skipped:
print(f" Warning: skipped {skipped} records with invalid amounts")
return cleaned
Stage 3: Enrich
ZIP_LOOKUP = {
"97201": {"city": "Portland", "state": "OR", "region": "Pacific NW"},
"97202": {"city": "Portland", "state": "OR", "region": "Pacific NW"},
"98101": {"city": "Seattle", "state": "WA", "region": "Pacific NW"},
"94102": {"city": "San Francisco", "state": "CA", "region": "West Coast"},
}
def enrich_with_location(donations: list[dict]) -> list[dict]:
"""Add city/state/region based on zip code."""
enriched = 0
for d in donations:
location = ZIP_LOOKUP.get(d.get("zip", ""), {})
d["city"] = location.get("city", "Unknown")
d["state"] = location.get("state", "Unknown")
d["region"] = location.get("region", "Unknown")
if d["city"] != "Unknown":
enriched += 1
print(f" Enriched {enriched}/{len(donations)} records with location data")
return donations
Stage 4: Aggregate
from collections import Counter
def build_aggregates(donations: list[dict], metrics: dict) -> dict:
"""Compute summary statistics from combined data."""
total_raised = sum(d["amount"] for d in donations)
donor_count = len(donations)
avg_donation = total_raised / donor_count if donor_count else 0
# Donations by city
by_city: dict[str, float] = {}
for d in donations:
by_city[d["city"]] = by_city.get(d["city"], 0) + d["amount"]
# Donations by campaign
by_campaign: dict[str, float] = {}
for d in donations:
by_campaign[d["campaign"]] = by_campaign.get(d["campaign"], 0) + d["amount"]
# Program stats from JSON
total_participants = sum(
p["participants"] for p in metrics.get("programs", [])
)
avg_satisfaction = (
sum(p["satisfaction_score"] for p in metrics["programs"])
/ len(metrics["programs"])
) if metrics.get("programs") else 0
return {
"quarter": metrics.get("quarter", "Unknown"),
"total_raised": total_raised,
"donor_count": donor_count,
"average_donation": avg_donation,
"donations_by_city": dict(sorted(by_city.items(), key=lambda x: -x[1])),
"donations_by_campaign": dict(sorted(by_campaign.items(), key=lambda x: -x[1])),
"total_participants": total_participants,
"average_satisfaction": avg_satisfaction,
"volunteer_hours": metrics.get("total_volunteer_hours", 0),
}
Stage 5: Output
def generate_text_report(agg: dict) -> str:
"""Generate a formatted text report from aggregated data."""
lines = [
f"{'=' * 50}",
f" QUARTERLY IMPACT REPORT — {agg['quarter']}",
f"{'=' * 50}",
"",
f" Total raised: ${agg['total_raised']:>12,.2f}",
f" Number of donors: {agg['donor_count']:>12}",
f" Average donation: ${agg['average_donation']:>12,.2f}",
"",
" DONATIONS BY CITY:",
]
for city, amount in agg["donations_by_city"].items():
lines.append(f" {city:<20} ${amount:>10,.2f}")
lines.extend([
"",
" DONATIONS BY CAMPAIGN:",
])
for campaign, amount in agg["donations_by_campaign"].items():
lines.append(f" {campaign:<20} ${amount:>10,.2f}")
lines.extend([
"",
f" Total participants: {agg['total_participants']:>12}",
f" Avg satisfaction: {agg['average_satisfaction']:>12.1f}/5.0",
f" Volunteer hours: {agg['volunteer_hours']:>12,}",
f"{'=' * 50}",
])
return "\n".join(lines)
def export_report(agg: dict, donations: list[dict], output_dir: str) -> None:
"""Export the report as text, CSV summary, and full JSON."""
output = Path(output_dir)
# Text report
report = generate_text_report(agg)
with open(output / "quarterly_report.txt", "w") as f:
f.write(report)
# CSV summary by city
with open(output / "donations_by_city.csv", "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["city", "total_amount"])
writer.writeheader()
for city, amount in agg["donations_by_city"].items():
writer.writerow({"city": city, "total_amount": f"{amount:.2f}"})
# Full JSON export
with open(output / "quarterly_data.json", "w") as f:
json.dump(agg, f, indent=2)
print(f" Exported report to {output_dir}/")
Putting It All Together
def run_pipeline():
"""Execute the complete data pipeline."""
print("=== Quarterly Impact Report Pipeline ===\n")
# Stage 1: Load
donations = load_donations("donors_q1.csv")
metrics = load_metrics("program_metrics.json")
# Stage 2: Clean
donations = clean_donations(donations)
# Stage 3: Enrich
donations = enrich_with_location(donations)
# Stage 4: Aggregate
aggregates = build_aggregates(donations, metrics)
# Stage 5: Output
report = generate_text_report(aggregates)
print()
print(report)
export_report(aggregates, donations, ".")
print("\n=== Pipeline complete ===")
run_pipeline()
Expected output:
=== Quarterly Impact Report Pipeline ===
Loaded 8 donations from CSV
Loaded metrics for 3 programs from JSON
Enriched 8/8 records with location data
==================================================
QUARTERLY IMPACT REPORT — Q1-2025
==================================================
Total raised: $ 2,950.00
Number of donors: 8
Average donation: $ 368.75
DONATIONS BY CITY:
San Francisco $ 1,200.00
Portland $ 900.00
Seattle $ 850.00
DONATIONS BY CAMPAIGN:
gala $ 1,925.00
annual $ 475.00
spring $ 550.00
Total participants: 444
Avg satisfaction: 4.7/5.0
Volunteer hours: 2,840
==================================================
Exported report to ./
=== Pipeline complete ===
Discussion Questions
-
Elena's pipeline currently loads all data into memory. If the nonprofit grew to 100,000 donors, what problems might arise? What strategies could you use to process the data without loading it all at once? (Hint: think about generators from Chapter 6, or processing in batches.)
-
The geographic enrichment step uses a hardcoded dictionary. In a real system, Elena would call a geocoding API for each zip code. What challenges does this introduce? Think about: rate limiting, network failures, cost, and caching.
-
The pipeline processes data in a strict linear sequence: Load → Clean → Enrich → Aggregate → Output. Can you think of a scenario where you'd want to reorder these steps? What about adding feedback loops (e.g., the output of one pipeline becomes the input of another)?
-
Elena's boss asks: "Can you make this pipeline run automatically every Monday morning?" What additional infrastructure would Elena need? What could go wrong with an unattended pipeline, and how would you design it to handle failures gracefully?
Mini-Project
Extend Elena's pipeline to include a fourth data source: a CSV file of volunteer hours by volunteer name and program. Merge the volunteer data with the program metrics, and add a "Volunteers" section to the report showing: - Total volunteer hours per program - Top 3 volunteers by total hours - Average hours per volunteer
Write the code, create sample data files, and produce the extended report.