Member Data Migration Pipeline (ETL Practice Project)

A small, self-directed ETL pipeline built to practice the core mechanics of a real data migration: extracting data from multiple inconsistent source systems, cleaning and standardizing it, and loading it into a data warehouse (Snowflake, with a local SQLite fallback for easy testing).

Motivation

Real customer/data migrations rarely deal with clean, single-source data. They deal with legacy exports full of inconsistent formatting, duplicate records, missing fields, and a second (or third) system with a completely different schema that still needs to be reconciled. This project is a small, honest simulation of that exact problem - built to practice the specific skills involved (data cleaning, deduplication, schema mapping, validation, and loading into a cloud warehouse) rather than a toy "clean CSV in, clean CSV out" exercise.

Data

The "raw" data in data/raw/ is synthetic, hand-crafted data - not a real company's data. It's deliberately messy in ways designed to mirror real migration problems:

• legacy_members.csv - inconsistent column naming, mixed-case emails, phone numbers in three different formats, dates in two different formats plus an unparseable text value, a duplicate member record with conflicting data, a missing name, and a sentinel value (-999) used as a placeholder for "no handicap on file."
• api_activity_export.json - a second source system with a completely different schema and field names, including a member with a blank name and a null numeric field.

Pipeline stages

1. Extract (src/extract.py) - reads both raw sources with no cleaning, exactly as a real extract step would.
2. Transform (src/transform.py) - the core of the project:
   • Standardizes column names, casing, whitespace, phone formats, and date formats
   • Converts a known sentinel value to a proper null rather than treating it as real data
   • Deduplicates on member_id, keeping the more complete record
   • Merges both sources with an outer join, deliberately, so that records existing in only one system are surfaced (flagged via a source_status column) rather than silently dropped
3. Load - two versions:
   • src/load_sqlite.py - loads into a local SQLite file, so the pipeline is fully runnable with zero external accounts
   • src/load_snowflake.py - loads into a real Snowflake warehouse, using a free trial account (see Setup below)

src/pipeline.py orchestrates all three stages end-to-end.

How to run

pip install -r requirements.txt

# Run against local SQLite (no account needed)
python src/pipeline.py --target sqlite

# Run against Snowflake (requires setup - see below)
python src/pipeline.py --target snowflake

Snowflake setup (optional)

1. Create a free trial account at https://signup.snowflake.com/
2. In the Snowflake UI, create a database, warehouse, and schema (the default setup wizard handles this).
3. Install the connector: pip install snowflake-connector-python
4. Set the following environment variables:

   export SNOWFLAKE_ACCOUNT=your_account_identifier
   export SNOWFLAKE_USER=your_username
   export SNOWFLAKE_PASSWORD=your_password
   export SNOWFLAKE_WAREHOUSE=COMPUTE_WH
   export SNOWFLAKE_DATABASE=YOUR_DATABASE
   export SNOWFLAKE_SCHEMA=PUBLIC

5. Run python src/pipeline.py --target snowflake

Testing

pytest tests/test_transform.py -v

Six unit tests cover the specific cleaning rules individually (name dropping, phone normalization, deduplication logic, sentinel value handling, unparseable date handling, and the outer-join merge behavior).

Data quality checks

The pipeline prints a basic data quality report after transforming and before loading - row counts, missing email/phone counts, how many records matched across both sources versus only appearing in one, and a check for duplicate emails post-cleaning. A real production pipeline would take this further (e.g. failing the load if a data quality threshold is breached), which this project does not attempt.

Limitations

• Synthetic data, small scale. Nine sample rows is enough to exercise every cleaning rule but says nothing about how this logic would perform on a real dataset with millions of rows or messier, less predictable errors.
• Full-refresh load, not incremental. load_sqlite.py and load_snowflake.py both overwrite the target table completely on each run (if_exists="replace" / overwrite=True). A real production pipeline would more likely use an incremental/upsert (merge) strategy so that repeated runs don't reprocess the entire dataset - this is a deliberate simplification for a practice project, not a production pattern.
• No orchestration/scheduling. This runs as a single script end-to-end. A real pipeline would typically run on a schedule or trigger (e.g. Airflow, dbt, or a cloud-native scheduler), which is out of scope here.
• No CI/CD. The tests exist and pass locally, but there's no automated pipeline running them on every change yet.

Why this project

Built to practice the specific skills involved in data migration and pipeline work - schema mapping, deduplication, data quality validation, and loading into a modern cloud warehouse.