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 Find Missing Values in Large Datasets in Python
Analyze missing values across multiple large pandas DataFrames with counts and percentages.
import pandas as pd
import numpy as np
def find_missing_values_summary(datasets):
"""Analyze missing values across multiple datasets (dict of name: DataFrame)."""
summary = {}
for name, df in datasets.items():
missing_count = df.isnull().sum()
total_rows = len(df)
missing_pct = (mi…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 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.
from 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…How to Stream a Large JSONL File Line by Line in Python
Process a large JSON-lines file incrementally using streaming techniques to avoid loading the entire file into memory.
import json
def process_large_file(filepath, chunk_size=8192):
"""
Stream a large JSON-lines file line by line, processing each record
without loading the entire file into memory.
"""
total_count = 0
total_sum = 0
with open(filepath, 'r') as f:
while True:
chunk = …How to Topologically Sort a DAG in Python
Compute a valid execution order for tasks with dependencies using Kahn's algorithm in Python.
from collections import defaultdict, deque
def topological_order(dependencies):
graph = defaultdict(list)
in_degree = defaultdict(int)
tasks = set(dependencies.keys())
for task, depends_on in dependencies.items():
for d in depends_on:
graph[d].append(task)
in_degree[t…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": "…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.