When a lakehouse ingests data from an upstream REST API that keeps adding fields, renaming columns, or mutating existing values, you quickly run into the “what if I need the old title of a post?” dilemma. If you model the data with a classic Star Schema you’ll either lose history or explode the fact table; if you go for a Data Vault you can capture every change, but you pay with extra joins and storage. This post walks you through a practical side‑by‑side comparison using the real JSONPlaceholder Posts API, showing how to simulate schema evolution, build both models, and query them for current and historical insights. By the end you’ll know which pattern fits your next production pipeline that must stay agile to API churn.
Key Takeaways
- Data Vault’s satellites record every attribute change, giving you true time‑travel and auditability.
- Star Schemas excel at fast, current‑state reporting when the source is stable or changes are handled upstream.
- Simulating CDC on a static API lets you stress‑test model resilience to schema evolution.
- Choosing a pattern involves trade‑offs in storage, ETL complexity, and downstream query performance.
- In many lakehouses, a Data Vault serves as the raw integration layer, with Star Schemas built atop it for consumption.
The Problem
Modern lakehouses often pull data from external APIs that change on the fly: new fields appear, existing values mutate, or entire records are updated. For analytics, you need to know not just the latest post content but also how it evolved, so that machine‑learning features can be engineered on historical trends or compliance teams can audit changes. A Star Schema struggles with these changes unless you redesign dimensions or accept data loss. Data Vault, with its explicit historical satellites, seems a natural fit, but its complexity can be overkill if the data never changes. The question is: how do you decide which model to adopt for a given evolving API source?
Data and Sources
Data fetched from the JSONPlaceholder Posts API (static sample data, but we will simulate changes).
- https://jsonplaceholder.typicode.com/posts – source data
- Requests – HTTP client used in the script
- Pandas – data manipulation
- Data Vault 2.0 methodology – design principles
- Kimball Group Star Schema methodology – dimensional modeling
- Beyond Atomic: Advanced Iceberg Lakehouse Patterns for Production Pipelines – context on underlying storage
Data accessed on 2026-10-30.
Loading the Data
We fetch the raw JSON and convert it into a Pandas DataFrame. The script handles network failures and JSON parsing errors gracefully.
import requests
import pandas as pd
def load_snapshot(url):
try:
resp = requests.get(url, timeout=10)
resp.raise_for_status()
data = resp.json()
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Network error while fetching {url}: {e}") from e
except ValueError as e:
raise RuntimeError(f"Failed to parse JSON from {url}: {e}") from e
return pd.DataFrame(data)
The Core Logic
We now walk through the four key stages: simulating schema evolution, building a Star Schema, constructing a Data Vault, and querying both models.
Step 1 – Simulate Dynamic Data for Stress Testing
After the first snapshot, we create a second one that mutates titles, bodies, adds a new field, and inserts a new record to mimic an evolving API.
def simulate_snapshot_v2(df_v1):
df_v2 = df_v1.copy()
# Mutate title of post id 1
df_v2.loc[df_v2['id'] == 1, 'title'] = "Updated title for post 1"
# Mutate body of post id 2
df_v2.loc[df_v2['id'] == 2, 'body'] = "Updated body for post 2"
# Add a new category field to post id 3
df_v2.loc[df_v2['id'] == 3, 'category'] = "Tech"
# Insert a brand new post
new_post = {
"userId": 5,
"id": 101,
"title": "New post from API",
"body": "Content of the new post",
"category": "News"
}
df_v2 = df_v2.append(new_post, ignore_index=True)
return df_v2
Step 2 – Star Schema for Current State Analytics
The Star Schema is straightforward: a fact table for posts and a dimension for users. It captures the snapshot as it is, but any later changes would overwrite or duplicate data.
def build_star_schema(df):
# Fact table
fact_posts = df[['id', 'userId', 'title', 'body']].copy()
fact_posts.rename(columns={'id': 'post_id'}, inplace=True)
# Dimension table
dim_users = df[['userId']].drop_duplicates().copy()
dim_users.rename(columns={'userId': 'user_id'}, inplace=True)
return fact_posts, dim_users
Step 3 – Embracing Historical Fidelity: Designing a Data Vault
We split business keys (hubs), descriptive attributes (satellites), and relationships (links). Each load updates satellites with new timestamps and closes previous records.
import hashlib
from datetime import datetime
def md5_hash(row, columns):
concat = '|'.join(str(row[col]) for col in columns)
return hashlib.md5(concat.encode()).hexdigest()
def build_hub(df, key_col, hub_name):
hub = df[[key_col]].drop_duplicates().copy()
hub['hash_key'] = hub.apply(lambda r: md5_hash(r, [key_col]), axis=1)
hub['business_key'] = hub[key_col]
return hub[['hash_key', 'business_key']].rename(columns={'business_key': hub_name})
def build_satellite(df, hub_hash_col, hub_name, attr_cols, sat_name, load_ts):
sat = df[attr_cols + [hub_hash_col]].copy()
sat['load_date_timestamp'] = load_ts
sat['load_end_timestamp'] = None
sat['hash_key'] = sat.apply(lambda r: md5_hash(r, [hub_hash_col] + attr_cols), axis=1)
sat.rename(columns={hub_hash_col: f'{hub_name}_hash_key'}, inplace=True)
return sat[[f'{hub_name}_hash_key', 'hash_key', 'load_date_timestamp',
'load_end_timestamp'] + attr_cols]
def build_link(df, hub1, hub2, link_name, load_ts):
link = df[['userId', 'id']].copy()
link['user_hash_key'] = link['userId'].apply(lambda x: md5_hash({'userId': x}, ['userId']))
link['post_hash_key'] = link['id'].apply(lambda x: md5_hash({'id': x}, ['id']))
link['load_date_timestamp'] = load_ts
link['load_end_timestamp'] = None
link['hash_key'] = link.apply(lambda r: md5_hash(r, ['user_hash_key', 'post_hash_key']), axis=1)
return link[['hash_key', 'user_hash_key', 'post_hash_key',
'load_date_timestamp', 'load_end_timestamp']].rename(columns={
'user_hash_key': f'user_hash_key',
'post_hash_key': f'post_hash_key'
})
Step 4 – Querying the Models
We compare current‑state queries and deep historical queries for both models.
def query_star(fact, dim):
return fact.merge(dim, left_on='userId', right_on='user_id')
def query_dv_current(hub_post, sat_post, link_user_post, dim_users):
# Current post details
current_sat = sat_post[sat_post['load_end_timestamp'].isna()]
merged = current_sat.merge(hub_post, left_on='post_hash_key', right_on='hash_key')
merged = merged.merge(link_user_post, left_on='post_hash_key', right_on='post_hash_key')
merged = merged.merge(dim_users, left_on='user_hash_key', right_on='hash_key')
return merged[['post_id', 'title', 'body', 'category', 'userId']]
def query_dv_history(sat_post, post_id):
post_hash = sat_post[sat_post['post_id'] == post_id]['hash_key'].iloc[0]
history = sat_post[sat_post['hash_key'] == post_hash]
return history.sort_values('load_date_timestamp')
Putting It Together
The main routine orchestrates the flow: load data, simulate evolution, build both schemas, run sample queries, and print results. It also demonstrates error handling for network failures and file write errors.
if __name__ == "__main__":
try:
df_v1 = load_snapshot("https://jsonplaceholder.typicode.com/posts")
except RuntimeError as e:
print(e)
sys.exit(1)
df_v2 = simulate_snapshot_v2(df_v1)
# Build Star Schema from v1
fact_star, dim_star = build_star_schema(df_v1)
# Build Data Vault from v1
load_ts_v1 = datetime.utcnow()
hub_post_v1 = build_hub(df_v1, 'id', 'post')
hub_user_v1 = build_hub(df_v1, 'userId', 'user')
sat_post_v1 = build_satellite(df_v1, 'post_hash_key', 'post',
['title', 'body', 'category'], 'post_details', load_ts_v1)
link_user_post_v1 = build_link(df_v1, hub_user_v1, hub_post_v1, 'user_post', load_ts_v1)
# Build Data Vault from v2 (merge with existing)
load_ts_v2 = datetime.utcnow()
hub_post_v2 = pd.concat([hub_post_v1, build_hub(df_v2, 'id', 'post')]).drop_duplicates(subset='hash_key')
hub_user_v2 = pd.concat([hub_user_v1, build_hub(df_v2, 'userId', 'user')]).drop_duplicates(subset='hash_key')
sat_post_v2 = build_satellite(df_v2, 'post_hash_key', 'post',
['title', 'body', 'category'], 'post_details', load_ts_v2)
sat_post_full = pd.concat([sat_post_v1, sat_post_v2]).reset_index(drop=True)
link_user_post_v2 = pd.concat([link_user_post_v1, build_link(df_v2, hub_user_v2, hub_post_v2, 'user_post', load_ts_v2)])
# Close previous satellite records for changed posts
for pid in df_v2['id'].unique():
current = sat_post_full[(sat_post_full['post_id'] == pid) &
sat_post_full['load_end_timestamp'].isna()].sort_values('load_date_timestamp')
if len(current) > 1:
# Close all but the latest
sat_post_full.loc[current.index[:-1], 'load_end_timestamp'] = current.iloc[-2]['load_date_timestamp']
# Run queries
print("Star Schema current state:")
print(query_star(fact_star, dim_star).head())
print("\nData Vault current state:")
print(query_dv_current(hub_post_v2, sat_post_full, link_user_post_v2, hub_user_v2).head())
print("\nData Vault historical for post_id=1:")
hist = query_dv_history(sat_post_full, 1)
print(hist[['load_date_timestamp', 'title', 'body', 'category']])
# Save outputs to CSV for verification
try:
query_star(fact_star, dim_star).to_csv("star_current.csv", index=False)
query_dv_current(hub_post_v2, sat_post_full, link_user_post_v2, hub_user_v2).to_csv("dv_current.csv", index=False)
hist.to_csv("dv_history_post1.csv", index=False)
except IOError as e:
print(f"Failed to write output files: {e}")
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Architecting for Change: Data Vault vs. Star Schema for Evolving API Data in a Lakehouse
Author: Mausam Adhikari
"""
import sys
import requests
import pandas as pd
import hashlib
from datetime import datetime
import os
# ------------------------------------------------------------------ #
# 1. Data Loading
# ------------------------------------------------------------------ #
def load_snapshot(url):
try:
resp = requests.get(url, timeout=10)
resp.raise_for_status()
data = resp.json()
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Network error while fetching {url}: {e}") from e
except ValueError as e:
raise RuntimeError(f"Failed to parse JSON from {url}: {e}") from e
return pd.DataFrame(data)
# ------------------------------------------------------------------ #
# 2. Simulate API evolution
# ------------------------------------------------------------------ #
def simulate_snapshot_v2(df_v1):
df_v2 = df_v1.copy()
df_v2.loc[df_v2['id'] == 1, 'title'] = "Updated title for post 1"
df_v2.loc[df_v2['id'] == 2, 'body'] = "Updated body for post 2"
df_v2.loc[df_v2['id'] == 3, 'category'] = "Tech"
new_post = {
"userId": 5,
"id": 101,
"title": "New post from API",
"body": "Content of the new post",
"category": "News"
}
df_v2 = df_v2.append(new_post, ignore_index=True)
return df_v2
# ------------------------------------------------------------------ #
# 3. Star Schema
# ------------------------------------------------------------------ #
def build_star_schema(df):
fact_posts = df[['id', 'userId', 'title', 'body']].copy()
fact_posts.rename(columns={'id': 'post_id'}, inplace=True)
dim_users = df[['userId']].drop_duplicates().copy()
dim_users.rename(columns={'userId': 'user_id'}, inplace=True)
return fact_posts, dim_users
# ------------------------------------------------------------------ #
# 4. Data Vault helpers
# ------------------------------------------------------------------ #
def md5_hash(row, columns):
concat = '|'.join(str(row[col]) for col in columns)
return hashlib.md5(concat.encode()).hexdigest()
def build_hub(df, key_col, hub_name):
hub = df[[key_col]].drop_duplicates().copy()
hub['hash_key'] = hub.apply(lambda r: md5_hash(r, [key_col]), axis=1)
hub['business_key'] = hub[key_col]
return hub[['hash_key', 'business_key']].rename(columns={'business_key': hub_name})
def build_satellite(df, hub_hash_col, hub_name, attr_cols, sat_name, load_ts):
sat = df[attr_cols + [hub_hash_col]].copy()
sat['load_date_timestamp'] = load_ts
sat['load_end_timestamp'] = None
sat['hash_key'] = sat.apply(lambda r: md5_hash(r, [hub_hash_col] + attr_cols), axis=1)
sat.rename(columns={hub_hash_col: f'{hub_name}_hash_key'}, inplace=True)
return sat[[f'{hub_name}_hash_key', 'hash_key', 'load_date_timestamp',
'load_end_timestamp'] + attr_cols]
def build_link(df, hub_user, hub_post, link_name, load_ts):
link = df[['userId', 'id']].copy()
link['user_hash_key'] = link['userId'].apply(lambda x: md5_hash({'userId': x}, ['userId']))
link['post_hash_key'] = link['id'].apply(lambda x: md5_hash({'id': x}, ['id']))
link['load_date_timestamp'] = load_ts
link['load_end_timestamp'] = None
link['hash_key'] = link.apply(lambda r: md5_hash(r, ['user_hash_key', 'post_hash_key']), axis=1)
return link[['hash_key', 'user_hash_key', 'post_hash_key',
'load_date_timestamp', 'load_end_timestamp']]
# ------------------------------------------------------------------ #
# 5. Queries
# ------------------------------------------------------------------ #
def query_star(fact, dim):
return fact.merge(dim, left_on='userId', right_on='user_id')
def query_dv_current(hub_post, sat_post, link_user_post, hub_user):
current_sat = sat_post[sat_post['load_end_timestamp'].isna()]
merged = current_sat.merge(hub_post, left_on='post_hash_key', right_on='hash_key')
merged = merged.merge(link_user_post, left_on='post_hash_key', right_on='post_hash_key')
merged = merged.merge(hub_user, left_on='user_hash_key', right_on='hash_key')
return merged[['post_id', 'title', 'body', 'category', 'userId']]
def query_dv_history(sat_post, post_id):
post_hash = sat_post[sat_post['post_id'] == post_id]['hash_key'].iloc[0]
history = sat_post[sat_post['hash_key'] == post_hash]
return history.sort_values('load_date_timestamp')
# ------------------------------------------------------------------ #
# 6. Main
# ------------------------------------------------------------------ #
def main():
# Load snapshot v1
try:
df_v1 = load_snapshot("https://jsonplaceholder.typicode.com/posts")
except RuntimeError as e:
print(e)
sys.exit(1)
# Simulate snapshot v2
df_v2 = simulate_snapshot_v2(df_v1)
# Build Star Schema from v1
fact_star, dim_star = build_star_schema(df_v1)
# Build Data Vault from v1
load_ts_v1 = datetime.utcnow()
hub_post_v1 = build_hub(df_v1, 'id', 'post')
hub_user_v1 = build_hub(df_v1, 'userId', 'user')
sat_post_v1 = build_satellite(df_v1, 'post_hash_key', 'post',
['title', 'body', 'category'], 'post_details', load_ts_v1)
link_user_post_v1 = build_link(df_v1, hub_user_v1, hub_post_v1, 'user_post', load_ts_v1)
# Build Data Vault from v2 (merge with existing)
load_ts_v2 = datetime.utcnow()
hub_post_v2 = pd.concat([hub_post_v1, build_hub(df_v2, 'id', 'post')]).drop_duplicates(subset='hash_key')
hub_user_v2 = pd.concat([hub_user_v1, build_hub(df_v2, 'userId', 'user')]).drop_duplicates(subset='hash_key')
sat_post_v2 = build_satellite(df_v2, 'post_hash_key', 'post',
['title', 'body', 'category'], 'post_details', load_ts_v2)
sat_post_full = pd.concat([sat_post_v1, sat_post_v2]).reset_index(drop=True)
link_user_post_v2 = pd.concat([link_user_post_v1, build_link(df_v2, hub_user_v2, hub_post_v2, 'user_post', load_ts_v2)])
# Close previous satellite records for changed posts
for pid in df_v2['id'].unique():
current = sat_post_full[(sat_post_full['post_id'] == pid) &
sat_post_full['load_end_timestamp'].isna()].sort_values('load_date_timestamp')
if len(current) > 1:
sat_post_full.loc[current.index[:-1], 'load_end_timestamp'] = current.iloc[-2]['load_date_timestamp']
# Run queries
print("Star Schema current state:")
print(query_star(fact_star, dim_star).head())
print("\nData Vault current state:")
print(query_dv_current(hub_post_v2, sat_post_full, link_user_post_v2, hub_user_v2).head())
print("\nData Vault historical for post_id=1:")
hist = query_dv_history(sat_post_full, 1)
print(hist[['load_date_timestamp', 'title', 'body', 'category']])
# Save outputs to CSV for verification
try:
query_star(fact_star, dim_star).to_csv("star_current.csv", index=False)
query_dv_current(hub_post_v2, sat_post_full, link_user_post_v2, hub_user_v2).to_csv("dv_current.csv", index=False)
hist.to_csv("dv_history_post1.csv", index=False)
except IOError as e:
print(f"Failed to write output files: {e}")
if __name__ == "__main__":
main()
Expected Output
Running the script prints three tables. The first shows the fact/dimension join from the Star Schema, the second shows the current post details from the Data Vault, and the third lists all historical snapshots for post_id = 1, ordered by load time. The CSV files `star_current.csv`, `dv_current.csv`, and `dv_history_post1.csv` are created for further inspection.
Limitations and Tradeoffs
Data Vault’s historical tracking inflates storage and requires careful hash key design; if you have millions of posts, the satellite tables can grow quickly. Query performance suffers because retrieving the current view requires filtering on `load_end_timestamp` and joining multiple tables. Conversely, a Star Schema keeps the data model lean and queries are fast, but any schema change forces a redesign of the fact or dimension tables and you lose the ability to audit changes. In a lakehouse, you might mitigate these tradeoffs by keeping the Data Vault as the raw layer and materializing Star Schemas on top for specific analytical workloads.
Frequently Asked Questions
When is a Star Schema definitively better than a Data Vault?
If the source data is truly static or changes are handled upstream, and you only need current‑state reporting with low latency, a Star Schema is simpler and more performant.
How do you manage hash key collisions in a Data Vault?
Include source identifiers or surrogate keys in the hash input. In this example we hash the business key alone; for production, combine `source_system` + `business_key` to keep collisions astronomically low.
Can I build Star Schemas from a Data Vault?
Yes. The Data Vault serves as a canonical integration layer; downstream marts can materialize Star Schemas that join hubs and satellites into a denormalized view optimized for BI.
What's the role of `load_end_timestamp` in Data Vault satellites, and how does it compare to a `is_current` flag?
`load_end_timestamp` gives you true time‑travel; you can query any historical point by timestamp. A boolean `is_current` flag is simpler but only lets you see the latest state; it cannot answer “what was the title on 2023‑07‑01?”.
How does Data Vault support GDPR or CCPA compliance for data retention and deletion?
Because each change is stored in its own satellite row, you can purge or redact specific rows or entire satellites without affecting other parts of the model, enabling fine‑grained retention policies.
What I'd Change
In production I would replace the in‑memory hash generation with a deterministic, columnar hash function that can run in the lakehouse engine (e.g., using Iceberg’s `hash()` UDF). I’d also move the CDC simulation into a streaming pipeline that ingests real API deltas, and materialize a “current view” materialized view or incremental table to keep analytical queries fast. Finally, I’d add a governance layer that tags each satellite with its source system and retention policy, so compliance teams can audit and scrub data with minimal operational overhead. These changes turn the proof‑of‑concept into a robust, scalable foundation for any evolving API ingestion in a lakehouse.