Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
How to Count Events by Minute with a Tumbling Window in Python
Group timestamps into fixed 60-second tumbling windows and count events per bucket using a dict.
from collections import defaultdict
from datetime import datetime, timedelta
def tumbling_window_count(events, window_seconds=60):
buckets = defaultdict(int)
for event in events:
ts = datetime.fromisoformat(event["timestamp"])
bucket_start = ts - timedelta(seconds=ts.second % window_seconds,
…How to Implement SCD Type 1 Overwrite in Python with SQLite
Implement SCD Type 1 dimension updates in Python using SQLite — overwrite existing rows with new data while preserving keys.
import sqlite3
# Simulate a dimension table with SCD Type 1 (overwrite)
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
# Create dimension table
cursor.execute("""
CREATE TABLE customer_dim (
customer_id INTEGER PRIMARY KEY,
customer_name TEXT,
city TEXT,
updated_at TEXT…How to Validate Fact Table Grain Row Counts in Python
Validate fact table grain by checking dimension key references, unique grain combinations, duplicate rows, and dimension cardinality from a CSV file.
import csv
import hashlib
from pathlib import Path
def validate_fact_grain(fact_file: Path, expected_dim_keys: dict[str, set[str]]) -> dict:
"""
Validate fact table grain by checking each row's dimension keys exist
in expected dimension tables and row count consistency.
"""
dim_references = {}
…How to perform a star schema join in Python
Denormalize mock fact and dimension tables by building lookup dicts and enriching each sales fact with customer, product, and date attributes.
from datetime import date
# Mock dimension tables
customers = [
{"customer_id": 1, "name": "Alice", "city": "New York"},
{"customer_id": 2, "name": "Bob", "city": "Los Angeles"},
{"customer_id": 3, "name": "Carol", "city": "Chicago"},
]
products = [
{"product_id": 101, "name": "Laptop", "category": "…Normalize Timestamps to UTC DateTime in Python
Convert timestamps in multiple formats to UTC-aware datetime objects using datetime.strptime and astimezone.
from datetime import datetime, timezone
raw_timestamps = [
"2024-01-15 14:30:00+02:00",
"17/05/2024 09:15:00 -0500",
"2024-03-01T22:45:00Z",
"2024-06-20 08:00:00+09:30"
]
def parse_and_convert(ts: str) -> datetime:
normalized_ts = ts.strip().replace("Z", "+00:00")
formats = [
"%Y-%m-%…Browse by section
Each section groups closely related Python snippets.
Data pipelines & processing — Python code examples
What you will find here
This page collects data pipelines & processing snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.
Samples vs tutorials and challenges
Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.