Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
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 Reduce Aggregate Counts from Mapped Chunks in Python
Combine a list of mapped chunk dictionaries into a single aggregated count dictionary using functools.reduce.
from functools import reduce
from collections import defaultdict
def aggregate_chunks(mapped_chunks):
"""Combine mapped chunk counts into a single aggregate dict."""
return reduce(
lambda acc, chunk: {
**acc,
**{k: acc.get(k, 0) + v for k, v in chunk.items()}
},
…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 Run a Mock Cron Pipeline Scheduler in Python
This code schedules a mock pipeline job to run every 2 seconds and hourly at :30 using the schedule library, then runs pending tasks for 10 seconds.
import time
import schedule
from datetime import datetime
def run_pipeline():
print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Pipeline executed")
schedule.every(2).seconds.do(run_pipeline)
schedule.every().hour.at(":30").do(run_pipeline)
print("Scheduler started. Press Ctrl+C to stop.")
end_time = ti…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 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 Track Checkpoint Offset After Batch Commit in Python
A batch processor that tracks the last successfully committed offset after processing records in batches, advancing the checkpoint only when each batch commits successfully.
import json
from typing import Any
class BatchProcessor:
"""Tracks checkpoint offset after committing batches."""
def __init__(self, batch_size: int = 3):
self.batch_size = batch_size
self.offset = 0 # last successfully committed offset (exclusive)
self.total_committed = 0
def …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.