Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
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 Group Data by Key in Python
Group a list of dictionaries by a specified key using a defaultdict and compute per-group averages.
from collections import defaultdict
def group_by_key(data, key):
grouped = defaultdict(list)
for item in data:
grouped[item[key]].append(item)
return dict(grouped)
if __name__ == "__main__":
records = [
{"name": "Alice", "dept": "Engineering", "score": 85},
{"name": "Bob", "de…How to Group Rows by Key into Nested Arrays in Python
This code groups rows in a list of dictionaries by a specified key and returns a dictionary with each key mapped to a list of values from another key.
from collections import defaultdict
def implode_rows(rows, key, value_key):
grouped = defaultdict(list)
for row in rows:
grouped[row[key]].append(row[value_key])
return dict(grouped)
if __name__ == "__main__":
data = [
{"category": "fruit", "item": "apple"},
{"category": "fr…How to Hash Email Addresses in a PII Masking Pipeline in Python
Replaces every email address in a text string with its SHA-256 hash to protect personally identifiable information (PII).
import hashlib
import re
def hash_email(email: str) -> str:
"""Mask an email address by hashing it with SHA-256."""
normalized = email.strip().lower()
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
def mask_pii_emails(text: str) -> str:
"""Replace all email addresses in text with their…How to Implement Incremental Load with Watermark by updated_at in Python
Load only new or changed rows into SQLite by comparing an updated_at timestamp against a stored watermark, returning counts and the new watermark.
import sqlite3
from datetime import datetime, timedelta
def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
"""Load only rows with updated_at greater than the last watermark."""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create table if it doesn't exist
…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 List Failed Records in a Dead Letter Queue Mock in Python
A mock Dead Letter Queue stores failed processing records with error details and timestamps, lists them, and exports to JSON.
import json
from datetime import datetime, timedelta
import random
class DeadLetterQueue:
def __init__(self):
self.failed_records = []
def add_failed_record(self, record_id, payload, error_message):
self.failed_records.append({
"record_id": record_id,
"payload": paylo…How to Merge Incremental Snapshot Upsert Dict in Python
Merge a snapshot dict into a base dict, recursively updating nested dictionaries while preferring snapshot values on conflicts.
def merge_upsert(base: dict, snapshot: dict) -> dict:
"""
Merge a snapshot dict into a base dict, preferring snapshot values
on key conflicts (upsert semantics). Nested dicts are merged recursively.
"""
result = dict(base)
for key, value in snapshot.items():
if key in result and i…How to Merge Multiple Data Sources in Python
A beginner-friendly helper that merges lists of dictionaries from multiple sources into one combined list using key filtering.
import json
def merge_pipeline_data(*data_sources, keys=()):
"""Merge multiple data sources (list of dicts) into a single list of merged dicts.
Args:
*data_sources: One or more lists of dictionaries.
keys: Tuple of keys to include from each source (empty means all keys).
Returns:
…How to Parse Data in Python: A Beginner's Helper
This helper parses a JSON payload, extracts user names, emails, and signup dates, then summarizes the results.
import json
from datetime import datetime
from typing import Dict, List
def parse_data(payload: str) -> Dict[str, List]:
"""Parse a JSON payload and extract useful fields."""
raw = json.loads(payload)
users = raw.get("users", [])
parsed = {
"names": [],
"emails": [],
"signup_…How to Process CSV Data in Python with a Data Helper
Build a beginner-friendly data helper in Python that loads a CSV file, filters rows by a condition, and summarizes numeric fields.
import csv
from pathlib import Path
DATA = [
{"name": "Alice", "score": 88, "passed": True},
{"name": "Bob", "score": 42, "passed": False},
{"name": "Carol", "score": 95, "passed": True},
]
def load_csv(file_path: Path) -> list[dict]:
with file_path.open(newline="", encoding="utf-8") as f:
r…How to Register a Dataset Schema as JSON in Python
Define a catalog of dataset schemas and serialize them to JSON with the standard library json module.
import json
catalog = {
"name": "sample_catalog",
"version": "1.0",
"datasets": [
{
"id": "users",
"type": "table",
"fields": [
{"name": "id", "type": "integer", "key": True},
{"name": "email", "type": "string", "nullable": False}…How to Safely Coerce Strings to Numbers in Python
A safe conversion function that turns strings into integers or floats, returning a fallback value when conversion fails.
import math
def to_number(value, fallback=None):
"""Safely coerce a string to int or float, returning fallback on failure."""
if isinstance(value, (int, float)):
return value
try:
# Try int first for clean whole numbers
return int(value)
except (ValueError, TypeError):
…How to Sort a List of Dictionaries by Key in Python
A reusable helper function that sorts a list of dictionaries by a specified key, with optional descending order support.
from typing import List
def sort_records(records: List[dict], key: str, descending: bool = False) -> List[dict]:
"""Sort a list of dictionaries by a specified key."""
return sorted(records, key=lambda record: record[key], reverse=descending)
def demonstrate_sorting() -> None:
users = [
{"name": …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 Unpivot Wide to Long with pandas melt in Python
This code demonstrates how to use pandas.melt to unpivot a wide DataFrame into a tidy long format, converting subject columns into rows.
import pandas as pd
# Sample wide-format data
df_wide = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'math': [90, 85, 95],
'science': [80, 92, 88]
})
print("Original wide DataFrame:")
print(df_wide)
# Melt: unpivot subject columns into rows
df_long = pd.melt(
df_wide,
…How to Validate Data in a Python Pipeline
A helper module to validate common record types — email, positive integer, and non-empty string list — before processing data in a pipeline.
from typing import Any, Iterable
def is_valid_email(email: str) -> bool:
"""Basic email check: one '@', no spaces, dot after '@'."""
if "@" not in email or " " in email:
return False
local, _, domain = email.partition("@")
return bool(local) and "." in domain
def is_positive_int(value: Any)…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 create a dated snapshot path for a dataset in Python
Generate a versioned directory path combining a base directory, dataset name, and today's date, ready for creating snapshots in data pipelines.
import datetime
import os
from pathlib import Path
def snapshot_path(base_dir: str, dataset_name: str) -> Path:
"""Return a dated snapshot path for a dataset under a base directory."""
today = datetime.date.today().isoformat()
return Path(base_dir) / dataset_name / today
if __name__ == "__main__":
…How to detect anomalies in a column using z-score in Python
Detect outliers in a list of numbers using z-score statistics, flagging values that deviate significantly from the mean.
import random
def z_score_anomaly_detection(data, threshold=2.0):
"""
Detect anomalies in a list of numbers using z-score.
"""
mean = sum(data) / len(data)
variance = sum((x - mean) ** 2 for x in data) / len(data)
std_dev = variance ** 0.5
if std_dev == 0:
return []
a…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": "…How to route late-arriving data to a side output in Python
Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.
from collections import defaultdict
def late_arriving_side_output(events, late_threshold_ts):
"""
Mock a streaming pipeline that separates late-arriving data events
into a side output list (e.g., for dead-letter analysis).
events: list of (timestamp, data) tuples, timestamps as ints.
late_thresho…How to shard output by primary key hash mod N in Python
This code computes a consistent shard index for any primary key string using an MD5 hash mod the number of shards, enabling stable key-based data distribution.
import hashlib
def shard_id(primary_key: str, num_shards: int) -> int:
"""Return the shard index for a primary key using MD5 hash mod N."""
digest = hashlib.md5(primary_key.encode("utf-8")).hexdigest()
hash_int = int(digest, 16)
return hash_int % num_shards
if __name__ == "__main__":
keys = ["use…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.