Museum CRM Data Pipeline with dbt
Build a dbt pipeline that cleans museum CRM data and catches quality issues.
Introduction
30 Second Summary
Museums track thousands of interactions every year. Visitors buy tickets, members renew, donors give, and campaigns run. Behind the scenes, all of that data lands in a CRM system that is messy, inconsistent, and full of gaps.
In this project, you will build a local analytics platform that generates realistic museum CRM data, loads it into PostgreSQL, and transforms it with dbt into clean, tested, business-ready models a museum executive could actually use.
What You'll Build
You will run a single command and watch dbt catch dirty data (null emails, duplicate records, invalid statuses) before it ever reaches your analytics layer, then query five mart models that answer real nonprofit business questions.
By the end of this project, you'll have:
- A Python ETL pipeline that generates a full year of museum operations data with intentional quality problems, so you can prove your dbt tests actually catch real issues.
- A complete dbt project with staging and mart layers where you run dbt build and see every data quality test pass after you fix the dirty data.
- Five analytics mart models you can query to find your museum's most valuable members, best-performing campaigns, and visitor-to-member conversion rates.
- Secret Mission: Add source freshness monitoring and a custom generic test that validates email format across all contact-related models.
Are there any prerequisites?
You should be comfortable with Python, SQL, and basic dbt concepts (models, tests, sources). You also need Docker Desktop installed and running on your machine before starting.
Before We Start
Before We Start...
Before we dive in, let's lock in what you're about to build and why it matters.
Set Up Your Environment
To build a museum CRM analytics platform, you need three things connected: a PostgreSQL database to hold the raw data, dbt to transform it, and a project structure that keeps everything organized.
Without these tools wired together, there is nothing to model or query. In this step, you will spin up PostgreSQL in Docker, install dbt and Faker into a virtual environment, and initialize a dbt project that can talk to your database.
In this step, get ready to:
- Start PostgreSQL in Docker and create the museum database.
- Install dbt and Faker in a Python virtual environment.
- Initialize the dbt project and confirm the database connection.
Start PostgreSQL in Docker
Your museum CRM data needs a database to live in. Docker lets you run PostgreSQL locally without installing it on your machine.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir museum-crm-pipeline
cd museum-crm-pipeline- Start a PostgreSQL 16 container with the museum database by running this command:
docker run -d --name museum_postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=museum_crm -p 5432:5432 postgres:16What does this command do?
- -d runs the container in the background so you keep your terminal.
- --name museum_postgres gives the container a name you can reference later.
- -e POSTGRES_DB=museum_crm tells PostgreSQL to create a database called museum_crm on first boot.
- -p 5432:5432 maps the container's port to your machine so dbt can connect via localhost.
- Confirm the container is running by running:
docker psYou should see a row with museum_postgres in the NAMES column and postgres:16 in the IMAGE column.
Don't see the container?
- Make sure Docker Desktop is running. Check for the Docker icon in your menu bar.
- If you see a port conflict error, another service is already using port 5432. Stop it or change the host port (e.g. -p 5433:5432).
- help me fix my Docker container not starting
Create a Python virtual environment and install dependencies
A virtual environment keeps your project's packages isolated from your system Python. This prevents version conflicts with other projects on your machine.
- Create and activate a virtual environment by running these commands:
macOS
python3 -m venv venv
source venv/bin/activateWindows
python -m venv venv
.\venv\Scripts\activateYour terminal prompt should now show (venv) at the beginning, confirming the virtual environment is active.
- Install dbt-postgres and Faker by running:
pip install dbt-postgres==1.11.0 Faker==40.31.0What is dbt-postgres installing?
The dbt-postgres package automatically installs dbt-core and psycopg2-binary (the PostgreSQL driver) as dependencies. You get three packages from a single install command.
Faker is a data generation library you will use in the next step to create realistic museum CRM records.
- Verify dbt installed correctly by running:
dbt --versionYou should see version information for dbt-core and the postgres plugin listed in the output.
- Verify Faker installed correctly by running:
python -c "import faker; print(faker.VERSION)"You should see 40.31.0 printed to your terminal.
Seeing 'command not found' for dbt?
- Make sure your virtual environment is active. Look for (venv) in your terminal prompt.
- If you closed your terminal, navigate back to ~/Desktop/museum-crm-pipeline and reactivate with source venv/bin/activate.
- help me troubleshoot my dbt installation
Initialize the dbt project and configure the connection
Now that dbt is installed and PostgreSQL is running, you need a dbt project that knows how to connect to your database. The project holds your models, tests, and documentation.
- Initialize a new dbt project by running:
dbt init museum_crm --skip-profile-setupWhat does --skip-profile-setup do?
By default, dbt init launches an interactive wizard to configure your database connection. The --skip-profile-setup flag skips those prompts so you can write the configuration file yourself with exact values.
- Confirm the project was created by running:
ls museum_crmYou should see folders like models, macros, seeds, and a dbt_project.yml file.
- Open the museum-crm-pipeline folder in VS Code.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- Click File in the top menu bar.
- Click Open Folder.
- Select the museum-crm-pipeline folder on your Desktop, then click Open.
- In the VS Code Explorer sidebar, expand the museum_crm folder and open dbt_project.yml.
- Replace the entire content of the file with the following:
name: museum_crm
version: "1.0.0"
config-version: 2
profile: museum_crm
model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["data-tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
clean-targets:
- target
- dbt_packages
models:
museum_crm:
staging:
+materialized: view
marts:
+materialized: tableWhat does this configuration do?
- test-paths: ["data-tests"] tells dbt to look for singular data tests in a folder called data-tests instead of the default tests.
- The models: section sets staging models to materialize as views (fast, always fresh) and marts as tables (precomputed for faster queries).
- profile: museum_crm tells dbt which connection profile to look up in profiles.yml.
- Save dbt_project.yml.
Now you need to create the connection profile that tells dbt how to reach your PostgreSQL container.
- Create the dbt profiles directory and file by running these commands in your terminal:
mkdir -p ~/.dbt- Create the profiles file by running:
cat > ~/.dbt/profiles.yml << 'EOF'
museum_crm:
target: dev
outputs:
dev:
type: postgres
host: localhost
port: 5432
user: postgres
password: postgres
dbname: museum_crm
schema: analytics
threads: 4
EOFWhat is profiles.yml?
This file lives at ~/.dbt/profiles.yml and stores database connection details separately from your project code. The schema: analytics setting means dbt will create all models in a schema called analytics inside your museum_crm database.
- Move into the dbt project directory by running:
cd museum_crmBefore you run the connection test, what do you think will happen? Do you expect dbt to connect successfully, or will something be misconfigured?
- Test the connection by running:
dbt debugYou should see a series of checks for your configuration, profiles, and database connection. The last line reads All checks passed! confirming dbt can reach your Dockerized PostgreSQL.
Seeing a connection error?
- Confirm your Docker container is still running with docker ps. If it stopped, restart it with docker start museum_postgres.
- Double-check your ~/.dbt/profiles.yml has host: localhost and port: 5432 with no extra spaces.
- Make sure you are inside the museum_crm folder when running dbt debug. The command needs to find dbt_project.yml in the current directory.
- help me fix my dbt debug connection failure
✔️ Awesome, I've got everything!
Great. Double check you have saved dbt_project.yml and move on to the next step.
ⓧ I'd like to double check the full code
Here is the complete dbt_project.yml you should have at museum_crm/dbt_project.yml:
name: museum_crm
version: "1.0.0"
config-version: 2
profile: museum_crm
model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["data-tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
clean-targets:
- target
- dbt_packages
models:
museum_crm:
staging:
+materialized: view
marts:
+materialized: tableAnd your ~/.dbt/profiles.yml should contain:
museum_crm:
target: dev
outputs:
dev:
type: postgres
host: localhost
port: 5432
user: postgres
password: postgres
dbname: museum_crm
schema: analytics
threads: 4Your database is running and dbt knows how to connect to it. Next up, you will generate realistic museum CRM data and load it into PostgreSQL so you have something to model.
Generate and Load Museum CRM Data
Your database is running but empty. Without data in those tables, there is nothing for dbt to model, test, or transform.
In this step you will write a Python script that simulates a full year of museum operations. It generates visitors, members, donors, ticket sales, campaigns, and visit logs. Crucially, it plants intentional data quality problems (null emails, duplicate IDs, invalid statuses) that will prove why automated testing matters in later steps.
In this step, get ready to:
- Write a Python data generator that creates 6 tables of realistic CRM data with seeded randomness.
- Run the generator and verify both row counts and the presence of dirty data.
Write the Python data generator script
The generator uses Faker (seeded with a fixed value so every run produces the same data) and psycopg2 to bulk-load records into a raw schema. You will build it piece by piece.
- Create a new file called generate_museum_data.py in the same directory that contains your museum_crm/ dbt project folder (one level above it).
- Start the file with the imports, Faker seed, database config, and constant lists by pasting this code:
"""Museum CRM Data Generator - Creates realistic museum operational data with intentional quality issues."""
import random
from datetime import datetime, timedelta
from faker import Faker
import psycopg2
from psycopg2.extras import execute_values
# Seed both generators for reproducible output
fake = Faker()
Faker.seed(42)
random.seed(42)
# Connection details matching the Docker container from Step 1
DB_CONFIG = {
"dbname": "museum_crm",
"user": "postgres",
"password": "postgres",
"host": "localhost",
"port": "5432",
}
# Valid values for categorical fields
MEMBERSHIP_STATUSES = ["active", "expired", "cancelled", "pending"]
MEMBERSHIP_TYPES = ["individual", "family", "student", "senior", "patron"]
TICKET_TYPES = ["general", "special_exhibit", "guided_tour", "family_pack", "school_group"]
CAMPAIGN_TYPES = ["email", "direct_mail", "social_media", "event", "phone"]
COMMUNICATION_PREFS = ["email", "mail", "phone", "sms", "none"]
COUPON_CODES = ["WELCOME10", "MEMBER20", "SUMMER15", "FAMILY25", "STUDENT10", None, None, None]
EXHIBITS = ["Impressionist Gallery", "Modern Art Wing", "Ancient Egypt", "Natural History", "Photography", "Sculpture Garden", "Children's Museum", "Rotating Exhibit"]What does this code do?
Faker.seed(42) and random.seed(42) lock the random number generators so every run produces identical data. This makes your pipeline reproducible and testable.
execute_values from psycopg2.extras batches INSERT statements into a single VALUES clause, which is significantly faster than looping individual inserts.
- Below the constants, add the schema creation function that defines the raw_contacts table:
def create_schema_and_tables(conn):
with conn.cursor() as cur:
cur.execute("DROP SCHEMA IF EXISTS raw CASCADE;")
cur.execute("CREATE SCHEMA raw;")
cur.execute("""
CREATE TABLE raw.raw_contacts (
contact_id VARCHAR(50),
first_name VARCHAR(100),
last_name VARCHAR(100),
email VARCHAR(200),
phone VARCHAR(50),
address_city VARCHAR(100),
address_state VARCHAR(50),
source VARCHAR(50),
email_consent BOOLEAN,
communication_preference VARCHAR(20),
privacy_opt_out BOOLEAN,
created_at TIMESTAMP,
updated_at TIMESTAMP
);
""")- Continue inside the same with block. Add the memberships and donations tables:
cur.execute("""
CREATE TABLE raw.raw_memberships (
membership_id VARCHAR(50),
contact_id VARCHAR(50),
membership_type VARCHAR(50),
status VARCHAR(50),
start_date DATE,
end_date DATE,
renewal_count INTEGER,
amount_paid NUMERIC(10, 2),
payment_method VARCHAR(50),
created_at TIMESTAMP
);
""")
cur.execute("""
CREATE TABLE raw.raw_donations (
donation_id VARCHAR(50),
contact_id VARCHAR(50),
campaign_id VARCHAR(50),
amount NUMERIC(10, 2),
donation_date DATE,
donation_type VARCHAR(50),
is_recurring BOOLEAN,
tax_receipt_sent BOOLEAN,
created_at TIMESTAMP
);
""")- Continue adding the tickets and campaigns tables:
cur.execute("""
CREATE TABLE raw.raw_tickets (
ticket_id VARCHAR(50),
contact_id VARCHAR(50),
ticket_type VARCHAR(50),
quantity INTEGER,
unit_price NUMERIC(10, 2),
total_amount NUMERIC(10, 2),
coupon_code VARCHAR(50),
discount_amount NUMERIC(10, 2),
purchase_date DATE,
visit_date DATE,
created_at TIMESTAMP
);
""")
cur.execute("""
CREATE TABLE raw.raw_campaigns (
campaign_id VARCHAR(50),
campaign_name VARCHAR(200),
campaign_type VARCHAR(50),
start_date DATE,
end_date DATE,
budget NUMERIC(10, 2),
target_audience VARCHAR(100),
status VARCHAR(50),
created_at TIMESTAMP
);
""")- Finish the function with the visits table and the commit:
cur.execute("""
CREATE TABLE raw.raw_visits (
visit_id VARCHAR(50),
contact_id VARCHAR(50),
visit_date DATE,
exhibit_visited VARCHAR(200),
duration_minutes INTEGER,
is_member_visit BOOLEAN,
party_size INTEGER,
created_at TIMESTAMP
);
""")
conn.commit()What does create_schema_and_tables do?
It drops and recreates the raw schema each time so the generator is idempotent. Running it twice gives you the same clean starting point.
The 6 tables mirror a real museum CRM: contacts (the people), memberships, donations, tickets, marketing campaigns, and visit logs.
- Save generate_museum_data.py and confirm it exists by running:
ls generate_museum_data.pyYou should see the filename printed back. The script is not runnable yet because it has no generator functions or main() block.
- Now add the contact generator function. This is where the first quality issue lives. Paste below the create_schema_and_tables function:
def generate_contacts(n=500):
contacts = []
for i in range(n):
contact_id = f"CTK-{i+1:05d}"
created_at = fake.date_time_between(start_date="-2y", end_date="-6M")
updated_at = fake.date_time_between(start_date=created_at, end_date="now")
email = fake.email()
email_consent = random.choice([True, True, True, False])
communication_preference = random.choice(COMMUNICATION_PREFS)
privacy_opt_out = random.choice([False, False, False, False, True])
# Intentional quality issue: ~25 null emails
if i >= 475:
email = None
contacts.append((
contact_id, fake.first_name(), fake.last_name(), email,
fake.phone_number() if random.random() > 0.1 else None,
fake.city(), fake.state_abbr(),
random.choice(["website", "front_desk", "event", "referral", "social_media"]),
email_consent, communication_preference, privacy_opt_out, created_at, updated_at,
))
return contactsWhere is the dirty data?
The line if i >= 475: email = None intentionally nulls out the last 25 contacts' email addresses. In Step 4, a dbt not_null test will catch these and force you to handle them in the staging model.
The script also includes email_consent, communication_preference, and privacy_opt_out fields. These model GDPR-style governance and will drive a contactability flag in the staging layer.
- Add the membership generator below. This injects both duplicate IDs and invalid statuses:
def generate_memberships(contacts, n=200):
memberships = []
member_contacts = random.sample(contacts, min(n, len(contacts)))
for i, contact in enumerate(member_contacts):
contact_id = contact[0]
membership_id = f"MEM-{i+1:05d}"
start_date = fake.date_between(start_date="-2y", end_date="-1M")
end_date = start_date + timedelta(days=365)
status = random.choice(MEMBERSHIP_STATUSES)
membership_type = random.choice(MEMBERSHIP_TYPES)
amount_map = {"individual": 75.00, "family": 150.00, "student": 40.00, "senior": 55.00, "patron": 500.00}
amount = amount_map.get(membership_type, 75.00)
# Intentional quality issue: ~10 invalid statuses
if i >= 190:
status = "pending_review"
memberships.append((
membership_id, contact_id, membership_type, status, start_date, end_date,
random.randint(0, 5), amount, random.choice(["credit_card", "check", "cash", "online"]),
fake.date_time_between(start_date=start_date, end_date=start_date + timedelta(days=1)),
))
# Intentional quality issue: ~15 duplicate membership_ids
for i in range(15):
original = memberships[i]
duplicate = (
original[0], original[1], original[2], original[3], original[4], original[5],
original[6] + 1, original[7], original[8],
fake.date_time_between(start_date="-1M", end_date="now"),
)
memberships.append(duplicate)
return membershipsTwo quality issues in one function
The last 10 memberships get status "pending_review" which is not in the valid list. An accepted_values test will flag these later.
The second loop copies 15 existing records with the same membership_id but a different created_at timestamp. A unique test will catch these duplicates.
- Add the donation generator:
def generate_donations(contacts, campaigns, n=300):
donations = []
donor_contacts = random.sample(contacts, min(n, len(contacts)))
for i, contact in enumerate(donor_contacts):
campaign = random.choice(campaigns) if random.random() > 0.3 else None
donation_date = fake.date_between(start_date="-2y", end_date="today")
donations.append((
f"DON-{i+1:05d}", contact[0], campaign[0] if campaign else None,
round(random.choice([25, 50, 100, 250, 500, 1000, 5000]) * random.uniform(0.8, 1.2), 2),
donation_date, random.choice(["one_time", "recurring", "pledge", "in_kind"]),
random.choice([True, False, False, False]), True,
fake.date_time_between(start_date=donation_date, end_date=donation_date + timedelta(days=1)),
))
return donations- Add the campaign generator:
def generate_campaigns(n=10):
campaigns = []
campaign_themes = [
"Spring Membership Drive", "Year-End Giving", "Summer Family Fun",
"New Exhibit Launch", "Donor Appreciation Gala", "Student Outreach",
"Holiday Gift Memberships", "Corporate Partnership Push",
"Anniversary Celebration", "Community Day Promotion",
]
for i in range(n):
start_date = fake.date_between(start_date="-18M", end_date="-1M")
end_date = start_date + timedelta(days=random.randint(14, 90))
campaigns.append((
f"CMP-{i+1:05d}", campaign_themes[i], random.choice(CAMPAIGN_TYPES),
start_date, end_date, round(random.uniform(500, 15000), 2),
random.choice(["all_contacts", "lapsed_members", "donors", "visitors", "prospects"]),
"completed" if end_date < datetime.now().date() else "active",
fake.date_time_between(start_date=start_date - timedelta(days=7), end_date=start_date),
))
return campaigns- Add the ticket generator. This contains the last quality issue (future-dated purchases):
def generate_tickets(contacts, n=800):
tickets = []
ticket_contacts = random.choices(contacts, k=n)
for i, contact in enumerate(ticket_contacts):
ticket_type = random.choice(TICKET_TYPES)
quantity = random.randint(1, 6)
price_map = {"general": 25.00, "special_exhibit": 35.00, "guided_tour": 45.00, "family_pack": 80.00, "school_group": 15.00}
unit_price = price_map.get(ticket_type, 25.00)
coupon = random.choice(COUPON_CODES)
discount = 0.0
if coupon:
discount_pct = int("".join(filter(str.isdigit, coupon))) / 100
discount = round(unit_price * quantity * discount_pct, 2)
total = round(unit_price * quantity - discount, 2)
purchase_date = fake.date_between(start_date="-1y", end_date="today")
visit_date = purchase_date + timedelta(days=random.randint(0, 30))
# Intentional quality issue: ~5 tickets with future purchase dates
if i >= 795:
purchase_date = fake.date_between(start_date="+1M", end_date="+6M")
visit_date = purchase_date + timedelta(days=random.randint(0, 14))
tickets.append((
f"TKT-{i+1:05d}", contact[0], ticket_type, quantity, unit_price, total,
coupon, discount, purchase_date, visit_date,
fake.date_time_between(start_date=purchase_date, end_date=purchase_date + timedelta(days=1)),
))
return tickets- Add the visit generator and the bulk-load helper function:
def generate_visits(contacts, n=1000):
visits = []
visit_contacts = random.choices(contacts, k=n)
for i, contact in enumerate(visit_contacts):
visit_date = fake.date_between(start_date="-1y", end_date="today")
visits.append((
f"VIS-{i+1:05d}", contact[0], visit_date, random.choice(EXHIBITS),
random.randint(30, 240), random.choice([True, False]), random.randint(1, 6),
fake.date_time_between(start_date=visit_date, end_date=visit_date + timedelta(hours=12)),
))
return visits
def load_data(conn, table, columns, data):
with conn.cursor() as cur:
sql = f"INSERT INTO raw.{table} ({', '.join(columns)}) VALUES %s"
execute_values(cur, sql, data, page_size=100)
conn.commit()
print(f" Loaded {len(data)} rows into raw.{table}")Why execute_values instead of executemany?
The psycopg2 docs state that executemany is not faster than looping execute() in a loop. execute_values batches rows into a single VALUES clause per page, reducing network round-trips dramatically.
- Finally, add the main() function that orchestrates generation and loading:
def main():
print("Connecting to museum_crm database...")
conn = psycopg2.connect(**DB_CONFIG)
print("Creating schema and tables...")
create_schema_and_tables(conn)
print("Generating data...")
contacts = generate_contacts(500)
campaigns = generate_campaigns(10)
memberships = generate_memberships(contacts, 200)
donations = generate_donations(contacts, campaigns, 300)
tickets = generate_tickets(contacts, 800)
visits = generate_visits(contacts, 1000)
print("Loading data into PostgreSQL...")
load_data(conn, "raw_contacts", ["contact_id", "first_name", "last_name", "email", "phone", "address_city", "address_state", "source", "email_consent", "communication_preference", "privacy_opt_out", "created_at", "updated_at"], contacts)
load_data(conn, "raw_campaigns", ["campaign_id", "campaign_name", "campaign_type", "start_date", "end_date", "budget", "target_audience", "status", "created_at"], campaigns)
load_data(conn, "raw_memberships", ["membership_id", "contact_id", "membership_type", "status", "start_date", "end_date", "renewal_count", "amount_paid", "payment_method", "created_at"], memberships)
load_data(conn, "raw_donations", ["donation_id", "contact_id", "campaign_id", "amount", "donation_date", "donation_type", "is_recurring", "tax_receipt_sent", "created_at"], donations)
load_data(conn, "raw_tickets", ["ticket_id", "contact_id", "ticket_type", "quantity", "unit_price", "total_amount", "coupon_code", "discount_amount", "purchase_date", "visit_date", "created_at"], tickets)
load_data(conn, "raw_visits", ["visit_id", "contact_id", "visit_date", "exhibit_visited", "duration_minutes", "is_member_visit", "party_size", "created_at"], visits)
conn.close()
print("\nDone! Museum CRM data loaded successfully.")
print("Quality issues injected: ~25 null emails, ~15 duplicate membership IDs,")
print("~10 invalid statuses, ~5 future-dated tickets.")
if __name__ == "__main__":
main()What does main do?
It generates data in dependency order: contacts first (since memberships and donations reference them), then campaigns (since donations reference those), then the rest. Each table is bulk-loaded with load_data which prints the row count as confirmation.
- Save generate_museum_data.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file, then move on to running it.
ⓧ I'd like to double check the full code
Scroll through your file and confirm these sections appear in order:
- Imports, Faker/random seeds, DB_CONFIG, and constant lists.
- create_schema_and_tables with 6 CREATE TABLE statements ending in conn.commit().
- generate_contacts (null emails for i >= 475).
- generate_memberships (invalid statuses for i >= 190, 15 duplicate appends).
- generate_donations, generate_campaigns, generate_tickets (future dates for i >= 795), generate_visits.
- load_data helper using execute_values.
- main() orchestrating everything, ending with if __name__ == "__main__": main().
Run the generator and verify the raw data
The script is complete. Before you run it, what do you expect to see printed in the terminal?
- Run the data generator by executing:
python generate_museum_data.pyYou should see output similar to this:
Connecting to museum_crm database...
Creating schema and tables...
Generating data...
Loading data into PostgreSQL...
Loaded 500 rows into raw.raw_contacts
Loaded 10 rows into raw.raw_campaigns
Loaded 215 rows into raw.raw_memberships
Loaded 300 rows into raw.raw_donations
Loaded 800 rows into raw.raw_tickets
Loaded 1000 rows into raw.raw_visits
Done! Museum CRM data loaded successfully.
Quality issues injected: ~25 null emails, ~15 duplicate membership IDs,
~10 invalid statuses, ~5 future-dated tickets.Why 215 memberships instead of 200?
The generator creates 200 base memberships plus 15 intentional duplicates with the same membership_id. That gives 215 total rows. The dbt unique test will catch these in Step 4.
- Verify the contacts row count by connecting to PostgreSQL and running a query:
docker exec -it museum_postgres psql -U postgres -d museum_crm -c "SELECT COUNT(*) FROM raw.raw_contacts;"You should see 500 in the output.
- Now confirm the dirty data is present. Run a spot check for null emails:
docker exec -it museum_postgres psql -U postgres -d museum_crm -c "SELECT contact_id, email FROM raw.raw_contacts WHERE email IS NULL LIMIT 3;"You should see 3 rows where the email column shows empty (null) values, with contact IDs in the CTK-004xx range.
Script failed to connect?
Make sure your Docker container is still running. Check with docker ps and look for museum_postgres. If it stopped, restart it with docker start museum_postgres.
If you see a "relation already exists" error, that means the script partially ran before failing. Run it again. The DROP SCHEMA IF EXISTS raw CASCADE at the start makes it safe to re-run.
Help me debug this connection error
Your raw data is loaded and you have confirmed the quality issues are present. Next up, you will build dbt staging models that transform this messy data into clean, typed, well-documented views.
Build dbt Staging Models with Tests
Your raw museum data is loaded and ready to query. But those raw tables have cryptic column names, inconsistent types, and no documentation for anyone else on your team to understand.
Staging models solve this by creating a clean, typed, well-documented interface between raw operational data and the analytics that will rely on it. In this step, you will declare your raw tables as dbt sources, then build six staging models that rename columns, cast types, and add computed fields.
In this step, get ready to:
- Declare all 6 raw tables as dbt sources with column-level documentation.
- Create staging SQL models that clean and standardize each raw table.
- Add data quality tests that enforce contracts on primary keys, statuses, and relationships.
Define sources and create staging models
Before dbt can reference your raw tables, it needs a source declaration that maps table names to a logical namespace. You will also configure materialization settings so staging models build as views.
- Open museum_crm/dbt_project.yml in VS Code and replace its contents with the following configuration:
name: museum_crm
version: "1.0.0"
config-version: 2
profile: museum_crm
model-paths: ["models"]
analysis-paths: ["analyses"]
test-paths: ["data-tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]
clean-targets:
- target
- dbt_packages
models:
museum_crm:
staging:
+materialized: view
marts:
+materialized: tableWhat does this configuration do?
The models: block at the bottom tells dbt to materialize all staging models as views (lightweight, always up-to-date) and all mart models as tables (precomputed for fast querying). The test-paths setting points to data-tests where you will store singular test SQL files later.
- Create the directory models/staging/ inside your museum_crm project by running:
mkdir -p models/staging- Create the source declaration file by running:
touch models/staging/__sources.yml- Open models/staging/__sources.yml and paste the first half of the source definitions (the header plus contacts and memberships tables):
sources:
- name: museum_raw
description: Raw operational data from the museum CRM system.
schema: raw
config:
loaded_at_field: created_at
tables:
- name: raw_contacts
description: Master contact records.
columns:
- name: contact_id
description: Unique identifier for each contact (format CTK-XXXXX)
- name: email
description: Contact email address. May be null for walk-in visitors.
- name: email_consent
description: Whether the contact has opted in to email communications.
- name: communication_preference
description: Preferred communication channel.
- name: privacy_opt_out
description: Whether the contact has requested to be excluded from marketing.
- name: raw_memberships
description: Membership purchase and renewal records.
columns:
- name: membership_id
description: Unique identifier for each membership record.
- name: contact_id
description: Foreign key to raw_contacts.
- name: status
description: Current membership status.- Immediately below the raw_memberships block, add the remaining four table declarations:
- name: raw_donations
description: Donation transaction records.
columns:
- name: donation_id
description: Unique identifier for each donation.
- name: contact_id
description: Foreign key to raw_contacts.
- name: campaign_id
description: Foreign key to raw_campaigns. Null if not campaign-attributed.
- name: raw_tickets
description: Ticket purchase records.
columns:
- name: ticket_id
description: Unique identifier for each ticket purchase.
- name: contact_id
description: Foreign key to raw_contacts.
- name: coupon_code
description: Applied coupon code. Null if no coupon was used.
- name: raw_campaigns
description: Marketing campaign definitions.
columns:
- name: campaign_id
description: Unique identifier for each campaign.
- name: raw_visits
description: Visitor attendance log.
columns:
- name: visit_id
description: Unique identifier for each visit record.
- name: contact_id
description: Foreign key to raw_contacts.- Save __sources.yml and dbt_project.yml.
What are dbt sources?
Sources tell dbt about tables that exist outside of your dbt project. By declaring them in YAML, you can reference raw tables with the {{ source() }} function instead of hardcoding schema and table names. This makes your models portable and gives dbt the metadata it needs to generate documentation and lineage graphs.
Now create the first staging model. This is the most complex one because it derives two computed flags from the raw contact fields.
- Create models/staging/stg_contacts.sql and paste the following:
with source as (
select * from {{ source('museum_raw', 'raw_contacts') }}
),
cleaned as (
select
contact_id,
initcap(trim(first_name)) as first_name,
initcap(trim(last_name)) as last_name,
lower(trim(email)) as email,
phone,
address_city,
address_state,
source as acquisition_source,
coalesce(email_consent, false) as email_consent,
coalesce(communication_preference, 'none') as communication_preference,
coalesce(privacy_opt_out, false) as privacy_opt_out,
case
when email is not null
and email_consent = true
and privacy_opt_out = false
then true
else false
end as is_email_contactable,
email is not null as has_valid_email,
created_at,
updated_at
from source
where contact_id is not null
)
select * from cleanedWhat does this model do?
- Filters out rows where contact_id is null (garbage records that cannot be linked to anything).
- Standardizes names with initcap(trim(...)) and lowercases emails for consistent matching.
- Renames source to acquisition_source so downstream queries read like plain English.
- Derives is_email_contactable (can we email this person?) and has_valid_email (is there an email on file at all?) from consent and privacy fields.
- Save stg_contacts.sql and verify it compiles by running:
dbt compile --select stg_contactsYou should see output starting with your dbt version, a line like "Found 1 model, 0 tests... 6 sources", and then the compiled SQL printed to your terminal.
Seeing a compilation error?
- Check that __sources.yml uses the exact source name museum_raw and that your model references {{ source('museum_raw', 'raw_contacts') }} with matching spelling.
- Confirm your YAML indentation uses spaces (not tabs). YAML is sensitive to whitespace.
- Make sure your Docker PostgreSQL container is still running. dbt compile needs a live database connection.
help me debug a dbt compilation error in my staging model
Next, create the memberships staging model. This one handles duplicate records and invalid status values.
- Create models/staging/stg_memberships.sql and paste:
with source as (
select * from {{ source('museum_raw', 'raw_memberships') }}
),
deduplicated as (
select
*,
row_number() over (
partition by membership_id
order by created_at desc
) as row_num
from source
),
cleaned as (
select
membership_id,
contact_id,
membership_type,
case
when status in ('active', 'expired', 'cancelled', 'pending') then status
else 'unknown'
end as status,
start_date,
end_date,
renewal_count,
amount_paid,
payment_method,
created_at
from deduplicated
where row_num = 1
)
select * from cleanedHow does deduplication work here?
The row_number() over (partition by membership_id order by created_at desc) assigns rank 1 to the most recent record for each membership. The where row_num = 1 filter keeps only the latest, discarding older duplicates.
The case statement maps any status not in the valid list (like the injected pending_review) to unknown so downstream queries never encounter unexpected values.
- Save stg_memberships.sql and create models/staging/stg_donations.sql with the following:
with source as (
select * from {{ source('museum_raw', 'raw_donations') }}
),
cleaned as (
select
donation_id,
contact_id,
campaign_id,
amount,
donation_date,
donation_type,
is_recurring,
tax_receipt_sent,
created_at
from source
where donation_id is not null
and amount > 0
)
select * from cleaned- Save stg_donations.sql and verify both new models compile by running:
dbt compile --select stg_memberships stg_donationsYou should see compiled SQL output for both models, confirming they parse correctly against your sources.
The remaining three staging models follow the same pattern: pull from source, filter invalid rows, and pass through clean columns.
- Create models/staging/stg_tickets.sql with this content:
with source as (
select * from {{ source('museum_raw', 'raw_tickets') }}
),
cleaned as (
select
ticket_id,
contact_id,
ticket_type,
quantity,
unit_price,
total_amount,
coupon_code,
discount_amount,
purchase_date,
visit_date,
coupon_code is not null as used_coupon,
created_at
from source
where ticket_id is not null
and purchase_date <= current_date
)
select * from cleaned- Create models/staging/stg_campaigns.sql:
with source as (
select * from {{ source('museum_raw', 'raw_campaigns') }}
),
cleaned as (
select
campaign_id,
campaign_name,
campaign_type,
start_date,
end_date,
budget,
target_audience,
status,
end_date - start_date as duration_days,
created_at
from source
where campaign_id is not null
)
select * from cleaned- Save both files, then verify they compile by running:
dbt compile --select stg_tickets stg_campaignsYou should see the compiled SQL for both models printed to your terminal.
- Create the final staging model models/staging/stg_visits.sql:
with source as (
select * from {{ source('museum_raw', 'raw_visits') }}
),
cleaned as (
select
visit_id,
contact_id,
visit_date,
exhibit_visited,
duration_minutes,
is_member_visit,
party_size,
created_at
from source
where visit_id is not null
and visit_date <= current_date
)
select * from cleaned- Save stg_visits.sql and confirm all six staging models compile together by running:
dbt compile --select stagingYou should see compiled SQL for all 6 models. The output confirms your Jinja references resolve correctly and the SQL is valid.
Add data quality tests to staging models
dbt's built-in generic tests let you declare data contracts in YAML. These tests run assertions against your models every time you build the project, catching bad data before it reaches analytics.
- Create models/staging/stg_contacts.yml and paste:
models:
- name: stg_contacts
description: Cleaned and standardized contact records. Filters out records with null contact_id. Adds contactability flags derived from consent and privacy fields.
columns:
- name: contact_id
description: Primary key for contacts.
data_tests:
- unique
- not_null
- name: email
description: Lowercased email address. Null for contacts without email on file.
data_tests:
- not_null
- name: communication_preference
description: Preferred communication channel.
data_tests:
- accepted_values:
arguments:
values: ["email", "mail", "phone", "sms", "none"]
- name: is_email_contactable
description: Derived flag indicating whether the contact can be emailed.
- name: has_valid_email
description: Whether the contact has an email address on file.What do these tests assert?
- unique ensures no duplicate contact_id values exist after cleaning.
- not_null on contact_id guarantees every row has a primary key.
- not_null on email asserts that all contacts in the cleaned model have an email address. This will catch the ~25 null-email records from your raw data.
- accepted_values restricts communication_preference to exactly the five allowed options.
- Create models/staging/stg_memberships.yml and paste:
models:
- name: stg_memberships
description: Deduplicated and validated membership records.
columns:
- name: membership_id
description: Primary key for memberships.
data_tests:
- unique
- not_null
- name: contact_id
description: Foreign key to stg_contacts.
data_tests:
- not_null
- relationships:
arguments:
to: ref('stg_contacts')
field: contact_id
- name: status
description: Membership status after validation.
data_tests:
- accepted_values:
arguments:
values: ["active", "expired", "cancelled", "pending", "unknown"]What does the relationships test do?
The relationships test checks that every contact_id in stg_memberships exists in stg_contacts. This is referential integrity enforcement. If a membership points to a contact that was filtered out (e.g., a null contact_id record), this test will flag it.
- Save both YAML files.
Run dbt run to materialize staging models
All six staging models are written and your test definitions are in place. Now materialize the models as views in your PostgreSQL database to confirm everything compiles and executes correctly.
- Before you run this, predict: will all six models create successfully, or will something fail?
- Materialize all staging models by running:
dbt run --select stagingYou should see 6 lines like "OK created sql view model analytics.stg_contacts" with each model completing in under a second. All 6 should show [CREATE VIEW] status.
Seeing an error on one of the models?
- Check the specific model file for typos in column names. Compare your SQL against the source table columns in __sources.yml.
- If you see "Relation not found", make sure you ran generate_museum_data.py in Step 2 and that the raw schema exists.
- Confirm your Docker container museum_postgres is still running with docker ps.
help me debug why my dbt staging model failed to create
- Now generate the documentation and view the data lineage by running:
dbt docs generate && dbt docs serveYour browser should open to the dbt documentation site. Click the lineage graph icon in the bottom-right corner to see how your 6 raw source tables flow into the 6 staging models.
Stop the docs server when done
Press Ctrl+C in your terminal to stop the docs server and get your terminal back for the next step.
Your staging layer is in place and documented. In the next step, you will run the full test suite and discover that some of your tests catch the intentional quality issues hiding in the raw data.
Catch and Fix Data Quality Failures
Your staging models compiled and materialized successfully in the last step. Compilation means the SQL is valid. It does not mean the data is clean.
Running dbt build adds test execution to the mix. This is where dbt proves its value: bad data that would silently corrupt downstream analytics gets caught before it reaches decision-makers.
In this step, get ready to:
- Run dbt build and observe test failures caused by dirty source data.
- Update staging models to handle null values, duplicates, and invalid statuses.
- Re-run the full test suite and confirm all tests pass.
Run dbt build and observe test failures
In the previous step you ran dbt run --select staging which only materializes models. It never checks whether the data inside those models is trustworthy.
Before you run the next command, think about this: your raw data has ~25 null emails, ~15 duplicate membership IDs, and ~10 invalid statuses. Do you think the tests you defined will catch these problems, or will they slip through?
- Run dbt build on the staging layer to materialize models AND execute tests:
dbt build --select stagingYou will see test failures in your terminal output. The build reports failures like these:
- not_null_stg_contacts_email reports ~25 failures (contacts without email addresses).
- unique_stg_memberships_membership_id reports ~15 failures (duplicate membership records).
- accepted_values_stg_memberships_status reports ~10 failures (the invalid pending_review status).
What do these failures mean?
Each failure is dbt telling you that data reached your staging layer in a state you explicitly said is unacceptable. The not_null test on email means "every contact must have an email." The unique test on membership_id means "no duplicates allowed."
These tests enforce a data quality contract. Without them, null emails and duplicate IDs would silently flow into your analytics and produce incorrect totals, broken joins, and misleading dashboards.
These failures are intentional
You planted these quality issues in the data generator. The test failures prove the testing layer works. Now you will update your staging models to handle the dirty data gracefully.
Update staging models to handle dirty data
Each staging model needs a targeted fix for the quality issues its tests caught. You will add filtering, deduplication, and value mapping so that downstream analytics only receive clean records.
- Open models/staging/stg_contacts.sql and replace its entire contents with the following:
with source as (
select * from {{ source('museum_raw', 'raw_contacts') }}
),
cleaned as (
select
contact_id,
initcap(trim(first_name)) as first_name,
initcap(trim(last_name)) as last_name,
lower(trim(email)) as email,
phone,
address_city,
address_state,
source as acquisition_source,
coalesce(email_consent, false) as email_consent,
coalesce(communication_preference, 'none') as communication_preference,
coalesce(privacy_opt_out, false) as privacy_opt_out,
case
when email is not null
and email_consent = true
and privacy_opt_out = false
then true
else false
end as is_email_contactable,
email is not null as has_valid_email,
created_at,
updated_at
from source
where contact_id is not null
)
select * from cleanedWhat does this fix?
The where contact_id is not null clause filters out true garbage rows (records with no identifier at all). The has_valid_email boolean flag tracks which contacts lack email so downstream models can filter or segment accordingly.
The is_email_contactable flag combines email presence, consent, and privacy opt-out into a single reusable indicator for marketing queries.
- Save stg_contacts.sql.
- Verify the model compiles by running:
dbt run --select stg_contactsYou should see the model materialize as a view with no compilation errors.
Seeing a compilation error?
Check that your source() call matches the source name museum_raw and table name raw_contacts exactly as defined in __sources.yml. Confirm there are no stray commas after the last column in your SELECT.
help me fix a dbt compilation error in my stg_contacts.sql model
Next, the memberships model needs deduplication and status validation. This fix has two logical parts: the deduplication window function and the status mapping.
- Open models/staging/stg_memberships.sql and replace the contents with the source and deduplication CTEs:
with source as (
select * from {{ source('museum_raw', 'raw_memberships') }}
),
deduplicated as (
select
*,
row_number() over (
partition by membership_id
order by created_at desc
) as row_num
from source
),What does row_number() do here?
The row_number() window function assigns a number to each row within groups that share the same membership_id. By ordering desc on created_at, row 1 is always the most recent record. Keeping only row_num = 1 discards older duplicates.
- Immediately below the deduplication CTE, add the cleaned CTE and final select:
cleaned as (
select
membership_id,
contact_id,
membership_type,
case
when status in ('active', 'expired', 'cancelled', 'pending') then status
else 'unknown'
end as status,
start_date,
end_date,
renewal_count,
amount_paid,
payment_method,
created_at
from deduplicated
where row_num = 1
)
select * from cleanedWhat does the CASE statement do?
The CASE expression maps any status value that is not in the allowed list (active, expired, cancelled, pending) to unknown. This ensures the accepted_values test passes while preserving the record for investigation.
- Save stg_memberships.sql.
- Verify both updated models compile by running:
dbt run --select stg_contacts stg_membershipsYou should see both models materialize as views. Two lines showing OK confirm they compiled and ran without error.
Getting a SQL syntax error?
Verify the comma after the closing ), of the deduplicated CTE is present. Each CTE must be separated by a comma. The final CTE (cleaned) has no trailing comma before the select * from cleaned statement.
help me debug a SQL syntax error in my stg_memberships.sql dbt model
Finally, the tickets model needs to exclude records with impossible future purchase dates.
- Open models/staging/stg_tickets.sql and replace its contents with the following:
with source as (
select * from {{ source('museum_raw', 'raw_tickets') }}
),
cleaned as (
select
ticket_id,
contact_id,
ticket_type,
quantity,
unit_price,
total_amount,
coupon_code,
discount_amount,
purchase_date,
visit_date,
coupon_code is not null as used_coupon,
created_at
from source
where ticket_id is not null
and purchase_date <= current_date
)
select * from cleanedWhat does this fix?
The purchase_date <= current_date filter removes the ~5 tickets with future purchase dates. These are clearly data entry errors. The used_coupon boolean provides a clean flag for downstream coupon analysis without repeating the null-check logic.
- Save stg_tickets.sql.
✔️ Awesome, I've got everything!
Great. Make sure all three files are saved before proceeding to the final test run.
ⓧ I'd like to double check the full code
Scroll up and confirm your three updated files contain:
- stg_contacts.sql: The cleaned CTE with where contact_id is not null, the is_email_contactable CASE expression, and the has_valid_email flag.
- stg_memberships.sql: Three CTEs (source, deduplicated, cleaned) with row_number() partitioned by membership_id, the status CASE mapping, and where row_num = 1.
- stg_tickets.sql: The cleaned CTE with where ticket_id is not null and purchase_date <= current_date and the used_coupon boolean.
Re-run the test suite and confirm all tests pass
Your models now filter garbage rows, deduplicate records, and map invalid values. The tests should reflect this.
Before you run the build, predict: will all tests pass now? Which test do you think was the trickiest to resolve?
- Run the full staging build with tests:
dbt build --select stagingYou should see all models materialize and all tests pass. The summary shows 0 failures.
What is a data quality contract?
Every time you run dbt build, these tests execute automatically. If someone changes the data generator, adds a new source feed, or the upstream CRM introduces a bug, your tests catch it immediately.
This is a data quality contract: a set of machine-enforced guarantees that downstream consumers can rely on. No analyst querying your mart models will encounter null primary keys, duplicate records, or unexpected status values.
Still seeing test failures?
Run dbt test --select stg_contacts to isolate which model is failing. Check that your WHERE clauses and CASE expressions match the code above exactly. A missing where row_num = 1 in memberships is the most common oversight.
help me debug dbt test failures in my staging models
Your staging layer now enforces a data quality contract on every build. Next up, you will assemble these clean building blocks into analytics mart models that answer real museum business questions.
Build Analytics Mart Models
Your staging models are clean, tested, and ready. But a museum executive does not query staging views. They need answers to business questions: who are our most valuable members, which campaigns generate the best ROI, and are we converting visitors into members?
Mart models assemble those answers by joining multiple staging models into business-ready analytics tables. In this step, you will build five mart models that answer five distinct questions a museum decision-maker would ask.
In this step, get ready to:
- Create five analytics mart models covering member lifetime value, visitor conversion, donor segmentation, campaign performance, and ticket/coupon analysis.
- Add a singular data test that detects duplicate contacts across the CRM.
- Generate dbt documentation and view the full data lineage from raw sources through marts.
Create the member lifetime value model
The first mart model answers the most important CRM question: how much is each member worth across all their interactions with the museum? This means joining memberships, donations, and ticket purchases per contact into a single lifetime value score.
- Create the marts directory by running this command:
mkdir -p models/marts- Create a new file models/marts/mart_member_lifetime_value.sql and add the CTEs that pull data from each staging model:
with members as (
select * from {{ ref('stg_memberships') }}
),
contacts as (
select * from {{ ref('stg_contacts') }}
),
donations as (
select
contact_id,
count(*) as donation_count,
sum(amount) as total_donations
from {{ ref('stg_donations') }}
group by contact_id
),
tickets as (
select
contact_id,
count(*) as ticket_count,
sum(total_amount) as total_ticket_spend
from {{ ref('stg_tickets') }}
group by contact_id
),What do these CTEs do?
Each CTE pulls a focused slice of data from a different staging model. The donations and tickets CTEs pre-aggregate per contact so you can join them cleanly downstream without duplicating rows.
- Below the code you just wrote, add the member_stats CTE that calculates membership-level metrics:
member_stats as (
select
contact_id,
count(*) as membership_count,
sum(amount_paid) as total_membership_spend,
max(renewal_count) as max_renewal_count,
min(start_date) as first_membership_date,
max(end_date) as latest_end_date,
max(case when status = 'active' then 1 else 0 end) as has_active_membership
from members
group by contact_id
)- Below the member_stats CTE, add the final SELECT that joins everything and calculates lifetime value:
select
c.contact_id,
c.first_name,
c.last_name,
c.email,
c.acquisition_source,
ms.membership_count,
ms.max_renewal_count,
ms.first_membership_date,
ms.latest_end_date,
ms.has_active_membership = 1 as is_active_member,
case
when ms.has_active_membership = 1 then 'active'
when ms.latest_end_date >= current_date - interval '90 days' then 'at_risk'
else 'lapsed'
end as renewal_status,
current_date - ms.first_membership_date as tenure_days,
coalesce(ms.total_membership_spend, 0) as total_membership_spend,
coalesce(d.total_donations, 0) as total_donations,
coalesce(d.donation_count, 0) as donation_count,
coalesce(t.total_ticket_spend, 0) as total_ticket_spend,
coalesce(t.ticket_count, 0) as ticket_count,
coalesce(ms.total_membership_spend, 0)
+ coalesce(d.total_donations, 0)
+ coalesce(t.total_ticket_spend, 0) as total_lifetime_value
from member_stats ms
inner join contacts c on ms.contact_id = c.contact_id
left join donations d on ms.contact_id = d.contact_id
left join tickets t on ms.contact_id = t.contact_idWhat does the final SELECT do?
The renewal_status CASE expression classifies members as active, at-risk (membership ended within 90 days), or lapsed. The total_lifetime_value column sums all spending across memberships, donations, and tickets into one number a museum director can sort by.
- Save models/marts/mart_member_lifetime_value.sql.
- Verify the model compiles by running:
dbt run --select mart_member_lifetime_valueYou should see a line showing the model was created as a table in the analytics schema.
Seeing a compilation error?
Check that the opening with keyword is on line 1 and each CTE ends with a comma except member_stats (the last CTE before the SELECT). A missing comma between CTEs or an extra comma after the last one causes a syntax error.
Help me debug this
Create the visitor conversion and donor segmentation models
The next two models track how visitors become members and how donors should be segmented for outreach. Each follows the same CTE-then-SELECT pattern you used above.
- Create models/marts/mart_visitor_conversion.sql and add the CTEs that gather visit and membership data:
with visits as (
select * from {{ ref('stg_visits') }}
),
members as (
select * from {{ ref('stg_memberships') }}
),
contacts as (
select * from {{ ref('stg_contacts') }}
),
first_visit as (
select
contact_id,
min(visit_date) as first_visit_date,
count(*) as total_visits
from visits
group by contact_id
),
first_membership as (
select
contact_id,
min(start_date) as first_membership_date
from members
group by contact_id
)- Below the CTEs, add the final SELECT that calculates conversion metrics:
select
c.contact_id,
c.first_name,
c.last_name,
c.acquisition_source,
fv.first_visit_date,
fv.total_visits,
fm.first_membership_date,
fm.first_membership_date is not null as converted_to_member,
case
when fm.first_membership_date is not null and fv.first_visit_date is not null
then fm.first_membership_date - fv.first_visit_date
else null
end as days_to_convert
from first_visit fv
inner join contacts c on fv.contact_id = c.contact_id
left join first_membership fm on fv.contact_id = fm.contact_idWhat does this model answer?
The converted_to_member boolean and days_to_convert integer tell the museum how effectively they turn casual visitors into paying members and how long that journey takes on average.
- Save mart_visitor_conversion.sql.
- Now create models/marts/mart_donor_segmentation.sql and add the donor statistics CTEs:
with donations as (
select * from {{ ref('stg_donations') }}
),
contacts as (
select * from {{ ref('stg_contacts') }}
),
donor_stats as (
select
contact_id,
count(*) as donation_count,
sum(amount) as total_donated,
avg(amount) as avg_donation,
max(donation_date) as last_donation_date,
min(donation_date) as first_donation_date,
sum(case when is_recurring then 1 else 0 end) as recurring_donation_count
from donations
group by contact_id
)- Below the CTEs, add the SELECT that classifies donors into tiers and recency segments:
select
c.contact_id,
c.first_name,
c.last_name,
c.email,
c.is_email_contactable,
ds.donation_count,
ds.total_donated,
ds.avg_donation,
ds.first_donation_date,
ds.last_donation_date,
ds.recurring_donation_count,
current_date - ds.last_donation_date as days_since_last_donation,
case
when ds.total_donated >= 5000 then 'major'
when ds.total_donated >= 1000 then 'mid_level'
when ds.total_donated >= 100 then 'small'
else 'micro'
end as donor_tier,
case
when current_date - ds.last_donation_date <= 90 then 'active'
when current_date - ds.last_donation_date <= 365 then 'lapsing'
else 'lapsed'
end as recency_segment,
ds.recurring_donation_count > 0 as is_recurring_donor
from donor_stats ds
inner join contacts c on ds.contact_id = c.contact_idWhat are donor tiers?
The donor_tier CASE expression segments donors by total giving: major ($5000+), mid-level ($1000+), small ($100+), and micro. The recency_segment separately tracks engagement freshness so the museum can target lapsing donors before they go cold.
- Save mart_donor_segmentation.sql.
- Run all three mart models to confirm they compile:
dbt run --select mart_member_lifetime_value mart_visitor_conversion mart_donor_segmentationYou should see three lines each showing a model created as a table in the analytics schema.
One of the models failed?
Check that each file starts with with and each CTE name matches exactly what the SELECT references. A common mistake is a typo in the CTE alias (e.g. donor_stat instead of donor_stats).
Help me find my SQL syntax error
Create the campaign performance and ticket coupon models
The final two models measure marketing ROI and ticket revenue. The campaign model uses a more complex join pattern to attribute memberships driven by each campaign.
- Create models/marts/mart_campaign_performance.sql and add the source CTEs and donation aggregation:
with campaigns as (
select * from {{ ref('stg_campaigns') }}
),
donations as (
select * from {{ ref('stg_donations') }}
),
memberships as (
select * from {{ ref('stg_memberships') }}
),
campaign_donations as (
select
campaign_id,
count(*) as donation_count,
sum(amount) as total_donation_revenue,
count(distinct contact_id) as unique_donors
from donations
where campaign_id is not null
group by campaign_id
),- Below that code, add the membership attribution CTE and the start of the final SELECT:
campaign_memberships as (
select
d.campaign_id,
count(distinct m.contact_id) as memberships_driven
from donations d
inner join memberships m on d.contact_id = m.contact_id
and m.start_date between d.donation_date - interval '30 days'
and d.donation_date + interval '30 days'
where d.campaign_id is not null
group by d.campaign_id
)
select
c.campaign_id,
c.campaign_name,
c.campaign_type,
c.start_date,
c.end_date,
c.duration_days,
c.budget,
c.target_audience,
c.status,
coalesce(cd.donation_count, 0) as donation_count,
coalesce(cd.total_donation_revenue, 0) as total_donation_revenue,
coalesce(cd.unique_donors, 0) as unique_donors,
coalesce(cm.memberships_driven, 0) as memberships_driven,- Add the final columns and FROM clause at the bottom of the same file:
case
when c.budget > 0
then round(coalesce(cd.total_donation_revenue, 0) / c.budget, 2)
else 0
end as roi_ratio,
case
when c.budget > 0
then round(c.budget / nullif(coalesce(cd.unique_donors, 0), 0), 2)
else null
end as cost_per_donor
from campaigns c
left join campaign_donations cd on c.campaign_id = cd.campaign_id
left join campaign_memberships cm on c.campaign_id = cm.campaign_idHow does campaign membership attribution work?
The campaign_memberships CTE looks for members who started a membership within 30 days of donating through that campaign. This is a common attribution window in nonprofit analytics.
The roi_ratio divides total donations by campaign budget to show return on investment.
- Save mart_campaign_performance.sql.
- Create models/marts/mart_ticket_coupon_performance.sql and add the ticket summary CTEs:
with tickets as (
select * from {{ ref('stg_tickets') }}
),
summary_by_type as (
select
ticket_type,
count(*) as total_purchases,
sum(quantity) as total_tickets_sold,
sum(total_amount) as total_revenue,
avg(unit_price) as avg_unit_price,
sum(case when used_coupon then 1 else 0 end) as coupon_purchases,
sum(discount_amount) as total_discounts_given
from tickets
group by ticket_type
),
coupon_stats as (
select
coupon_code,
count(*) as times_used,
sum(discount_amount) as total_discount_value,
sum(total_amount) as revenue_with_coupon,
avg(quantity) as avg_quantity_per_purchase
from tickets
where used_coupon
group by coupon_code
)- Below the CTEs, add the final SELECT with discount impact calculations:
select
st.ticket_type,
st.total_purchases,
st.total_tickets_sold,
st.total_revenue,
st.avg_unit_price,
st.coupon_purchases,
st.total_discounts_given,
case
when st.total_purchases > 0
then round(st.coupon_purchases::numeric / st.total_purchases * 100, 1)
else 0
end as coupon_usage_rate_pct,
case
when st.total_revenue + st.total_discounts_given > 0
then round(st.total_discounts_given / (st.total_revenue + st.total_discounts_given) * 100, 1)
else 0
end as discount_impact_pct
from summary_by_type st
order by st.total_revenue descWhat does discount impact measure?
The discount_impact_pct shows what percentage of potential revenue was given away as coupon discounts. A high number suggests coupons are eating into margins.
The coupon_usage_rate_pct shows how many purchases used a coupon at all.
- Save mart_ticket_coupon_performance.sql.
- Run all five mart models together to confirm they compile:
dbt run --select martsYou should see five lines, each showing a model created as a table. All five should show OK status.
Seeing an error on the campaign model?
The most common issue is a missing comma between the coalesce(cm.memberships_driven, 0) as memberships_driven, line and the CASE expression that follows it. The comma at the end of that line is required because more columns follow.
Help me fix my campaign performance SQL
Add a singular data test and YAML definitions
Beyond built-in generic tests, dbt supports singular data tests. These are standalone SQL files that return rows which should NOT exist. If the query returns any rows, the test fails. You will use this to detect potential duplicate contacts in the CRM.
- Create the data tests directory by running:
mkdir -p data-tests- Create data-tests/assert_no_duplicate_contacts.sql with the following deduplication logic:
-- Singular data test: Detect potential duplicate contacts
with potential_duplicates as (
select
first_name,
last_name,
phone,
count(distinct contact_id) as contact_count
from {{ ref('stg_contacts') }}
where phone is not null
and first_name is not null
and last_name is not null
group by first_name, last_name, phone
having count(distinct contact_id) > 1
)
select * from potential_duplicatesHow does this singular test work?
The query finds contacts that share the same first name, last name, AND phone number but have different contact_id values. These are likely duplicate CRM entries that should be merged. If the query returns zero rows, the test passes.
This file lives in the data-tests/ directory (configured in dbt_project.yml as test-paths) rather than alongside models. dbt automatically discovers and runs it.
- Save data-tests/assert_no_duplicate_contacts.sql.
- Now add the YAML test definitions for all mart models. Create models/marts/marts.yml with the following content:
models:
- name: mart_member_lifetime_value
description: Member lifetime value analysis combining membership spend, donations, and ticket purchases.
columns:
- name: contact_id
description: Primary key - the member's contact ID.
data_tests:
- unique
- not_null
- name: mart_visitor_conversion
description: Visitor-to-member conversion analysis.
columns:
- name: contact_id
description: Primary key - the visitor's contact ID.
data_tests:
- unique
- not_null
- name: mart_donor_segmentation
description: Donor segmentation model with tier classification and recency scoring.
columns:
- name: contact_id
description: Primary key - the donor's contact ID.
data_tests:
- unique
- not_null- The file continues with the remaining two models. Add this below the donor segmentation entry:
- name: mart_campaign_performance
description: Campaign effectiveness analysis with ROI calculation.
columns:
- name: campaign_id
description: Primary key - the campaign identifier.
data_tests:
- unique
- not_null
- name: mart_ticket_coupon_performance
description: Ticket sales and coupon performance analysis by ticket type.
columns:
- name: ticket_type
description: Primary key - the ticket category.
data_tests:
- unique
- not_nullWhy unique and not_null on every mart primary key?
Mart models join multiple staging tables, which can accidentally introduce duplicate or null keys through bad join logic. Testing unique and not_null on every primary key catches these issues before they corrupt downstream reporting.
- Save models/marts/marts.yml.
- Run the full project build including all models and tests:
dbt buildBefore you run this, do you think all tests will pass? Consider whether the singular test will find actual duplicates in your generated data.
You should see output ending with a summary showing all resources passed with zero errors. The singular test passes because the Faker-generated data uses random names and phones, making exact duplicates across all three fields unlikely.
Seeing test failures on mart primary keys?
A unique failure on a mart means your JOIN logic produced duplicate rows. Check that aggregation CTEs (like donor_stats or member_stats) use group by contact_id to collapse rows before joining.
Help me debug my mart test failures
Generate documentation and view the data lineage
One of dbt's most powerful features is automatic documentation generation. Every source description, model description, and column description you wrote in YAML files becomes a searchable, browsable documentation site with a visual lineage graph.
- Generate the documentation and start the docs server by running:
dbt docs generate && dbt docs serveWhat does dbt docs serve do?
The dbt docs generate command creates a static site from your project metadata. The dbt docs serve command starts a local web server so you can browse it.
Press Ctrl+C in your terminal when you are done exploring.
- Open the URL shown in your terminal output in your browser.
- Click the lineage graph icon (bottom-right corner) to see the full data flow from raw sources through staging views to mart tables.
You should see raw source tables on the left flowing into staging models in the middle, which then feed into the five mart tables on the right. This is your complete data lineage.
- Press Ctrl+C in your terminal to stop the docs server.
- Query your top members to confirm the analytics are meaningful by running:
docker exec -it museum_postgres psql -U postgres -d museum_crm -c "SELECT first_name, last_name, total_lifetime_value, renewal_status FROM analytics.mart_member_lifetime_value ORDER BY total_lifetime_value DESC LIMIT 5;"You should see the museum's top 5 members ranked by combined spending, with their renewal status showing whether they are active, at-risk, or lapsed.
Query returns no rows?
Make sure you ran dbt build successfully in the previous substep. The dbt build command materializes the mart tables. Also confirm the table name is analytics.mart_member_lifetime_value (the schema comes from your profiles.yml configuration).
Help me troubleshoot my query
✔️ Awesome, I've got everything!
Great. Double check you have saved all five mart SQL files, the marts.yml, and the singular data test.
ⓧ I'd like to double check the full code
Confirm each of these files is present and complete by scrolling up to the code blocks where you created them:
- models/marts/mart_member_lifetime_value.sql (3 code blocks: source CTEs, member_stats CTE, final SELECT).
- models/marts/mart_visitor_conversion.sql (2 code blocks: CTEs, then SELECT).
- models/marts/mart_donor_segmentation.sql (2 code blocks: CTEs, then SELECT).
- models/marts/mart_campaign_performance.sql (3 code blocks: source CTEs + donations, memberships CTE + SELECT columns, final CASE + FROM).
- models/marts/mart_ticket_coupon_performance.sql (2 code blocks: CTEs, then SELECT).
- models/marts/marts.yml (2 code blocks covering all 5 model definitions).
- data-tests/assert_no_duplicate_contacts.sql (1 code block).
If any file is missing or incomplete, add the missing code from the relevant code block above, save, and re-run dbt build to confirm everything passes.
Your analytics layer is complete. You now have five mart models answering real museum business questions, a singular data test enforcing CRM governance, and full documentation with lineage. Next, you will take on a Secret Mission to add source freshness monitoring and a custom email validation test.
Secret mission
Source Freshness and Custom Email Validation
Production CRM pipelines need to detect when source data stops arriving and need reusable validation logic beyond built-in tests. Add source freshness monitoring and a custom generic test that validates email format across your contact models.
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 runs entirely locally via Docker, so there are no ongoing costs.
Resources you used:
- Docker container museum_postgres running PostgreSQL 16 (no cost, local only).
- Python virtual environment with dbt and Faker packages (no cost).
- Local project files including the museum_crm/ dbt project and generate_museum_data.py (no cost).
Keep everything running
No action needed. Choose this if you are still actively building or plan to continue working on the project soon.
- The museum_postgres Docker container continues running in the background. It uses minimal resources.
- Your Python virtual environment and all project files remain on disk, ready to use.
- Run dbt build anytime to re-execute models and tests.
Pause - I'll come back to this later
Shut down the running Docker container to free up memory. Your data persists inside the container volume, so you can restart it later without losing anything.
- Stop the PostgreSQL container by running:
docker stop museum_postgres- When you are ready to resume, restart it by running:
docker start museum_postgres- Deactivate your Python virtual environment by running:
deactivate- Reactivate it later with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).
Delete - I don't want to use this again
Remove all project resources and start fresh. This deletes everything permanently.
- Remove the Docker container and its data by running:
docker rm -f museum_postgres- Deactivate your virtual environment (if active) by running:
deactivate- Delete the dbt project directory by running:
rm -rf museum_crm/- Delete the data generator script by running:
rm generate_museum_data.py- Delete the Python virtual environment folder by running:
rm -rf venv/- Optionally remove the PostgreSQL Docker image to free disk space by running:
docker rmi postgres:16Nice Work!
Nice Work!
Nice work! You just built a full museum CRM analytics platform from scratch, complete with realistic dirty data, automated quality testing, and business-ready mart models.
You've learned how to:
- Build a Python ETL pipeline that generates realistic CRM data across 6 operational tables with intentional quality issues, then bulk-loads it into PostgreSQL using psycopg2.
- Design a dbt project with staging and mart layers, implementing data quality tests that catch null values, duplicates, and invalid statuses before they reach your analytics.
- Create 5 analytics mart models answering real business questions: member lifetime value, visitor conversion, donor segmentation, campaign ROI, and ticket/coupon performance.
- Secret Mission: Added source freshness monitoring and a reusable custom generic test for email validation across the project.
Ready to quiz yourself?