Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
Generate a Mock CDC Changelog in Python
Simulate a CDC changelog with INSERT, UPDATE, and DELETE operations, timestamps, and record snapshots for testing data pipelines.
import json
from datetime import datetime, timedelta
def generate_mock_changelog(records, operations=("INSERT", "UPDATE", "DELETE")):
"""Simulate a CDC changelog from a list of record snapshots."""
base_time = datetime(2025, 1, 1, 8, 0, 0)
changelog = []
for idx, record in enumerate(records):
…Group Python Events into Sessions with a Gap Timeout
Groups timestamped events into sessions, starting a new session when the time gap exceeds a specified timeout.
from itertools import groupby
from datetime import datetime, timedelta
def session_window_group(events, gap_seconds=300):
"""Group events into sessions where gap > gap_seconds starts a new session."""
if not events:
return []
events = sorted(events, key=lambda x: x[0])
sessions = []
c…How to Convert Data Types in a Python Data Pipeline
Demonstrates a simple Python data pipeline that converts string values to proper types (bool, int, float, datetime) and outputs structured JSON.
import json
from datetime import datetime
def convert_value(value):
"""Convert string values to appropriate Python types."""
if value.lower() == "true":
return True
if value.lower() == "false":
return False
if value.isdigit():
return int(value)
try:
return float(val…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 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 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 Partition Output Files by Date Key in Python
Group output files into a dictionary partitioned by a YYYYMMDD date key extracted from the filename prefix.
from pathlib import Path
from collections import defaultdict
def partition_files_by_date(directory: str) -> dict:
"""Partition output files by date key extracted from filename (YYYYMMDD prefix)."""
path = Path(directory)
partitions = defaultdict(list)
for file in path.iterdir():
if file.i…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 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": "…Idempotent Pipeline Dedupe by Record ID Set in Python
Filters records against a persistent set of seen IDs, returning only new ones and the updated set for idempotent pipeline processing.
def dedupe_records(records, seen_ids=None):
"""Return records whose id has not been seen before."""
if seen_ids is None:
seen_ids = set()
unique = []
for record in records:
record_id = record.get("id")
if record_id not in seen_ids:
seen_ids.add(record_id)
…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-%…Rollback dataset to previous snapshot pointer in Python
A SnapshotManager class stores timestamped data snapshots and rolls back to the most recent snapshot at or before a target time.
from datetime import datetime, timedelta
class SnapshotManager:
def __init__(self):
self.snapshots = {} # timestamp -> data
self.current_pointer = None
def create_snapshot(self, data):
timestamp = datetime.now()
self.snapshots[timestamp] = data
self.current_pointer =…Validate dict schema at pipeline boundary in Python
This code validates a dictionary against a TypedDict schema at a pipeline boundary, enforcing required fields and types with custom error messages.
from typing import Any, TypedDict
class Person(TypedDict):
name: str
age: int
email: str
def validate_person(data: dict[str, Any]) -> Person:
errors: list[str] = []
if not isinstance(data.get("name"), str) or not data["name"].strip():
errors.append("name must be a non-empty string")
…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.