How to Implement Slowly Changing Dimension Type 2 History in Python
Build a type-2 slowly changing dimension pipeline that closes old records and opens new ones when customer data changes.
Python code
28 linesfrom datetime import datetime, timedelta
def apply_scd_type2(records, current_date):
"""Returns active records after inserting new records with type-2 history."""
history = []
active = {}
for record in records:
key = record["customer_id"]
if key in active:
active[key]["end_date"] = current_date - timedelta(days=1)
history.append(active[key])
record["start_date"] = current_date
record["end_date"] = None
active[key] = record
return history + list(active.values())
if __name__ == "__main__":
base_date = datetime(2024, 1, 1)
records = [
{"customer_id": 1, "address": "123 A St", "start_date": None, "end_date": None},
{"customer_id": 2, "address": "456 B Ave", "start_date": None, "end_date": None},
{"customer_id": 1, "address": "789 C Blvd", "start_date": None, "end_date": None},
]
for i, relation in enumerate(apply_scd_type2(records, base_date + timedelta(days=i)), 1):
print(f"customer {relation['customer_id']}: {relation['address']} | start={relation['start_date'].date()} | end={relation['end_date'].date() if relation['end_date'] else 'current'}")
Output
customer 1: 123 A St | start=2024-01-01 | end=2024-01-01
customer 2: 456 B Ave | start=2024-01-01 | end=current
customer 1: 789 C Blvd | start=2024-01-02 | end=current
How it works
The apply_scd_type2 function processes records in order, keeping an active dictionary keyed by customer_id. When a customer appears again, the previous active record gets its end_date set to the day before the new current_date, and that closed version is appended to history. The new record then becomes active with start_date set to current_date and end_date as None, representing current status. Finally, the function returns closed historical records plus the current active ones, giving a complete snapshot of the dimension history.
Common mistakes
- Mutating input records in place which can affect other parts of the pipeline
- Forgetting to close the old active record before inserting the new one
- Using the same `current_date` for both close and open dates instead of closing one day prior
Variations
- Use a `defaultdict` and `last_seen` map to handle out-of-order event streams
- Store history in a SQL table with a `valid_from`/`valid_to` range and `is_current` flag
Real-world use cases
- Tracking customer address changes in a CRM system while preserving historical versions for reporting.
- Maintaining product pricing history where each price change creates a new dimension row with validity dates.
- Auditing employee job-title changes for compliance while keeping current and past records queryable.
Sponsored
More from Data pipelines & processing
- Add a UUID Surrogate Key to Each Row in a CSV with Python easy
- Attach Source File Metadata to Records in Python easy
- Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets medium
- Check Null Rate Threshold in PySpark DataFrame medium
- Count Records Processed per Category in Python easy
- Create Data Helper Functions in Python for Beginners easy
Keep learning
Related tutorials and quizzes for this topic.