Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
Count Records Processed per Category in Python
Use a Counter dictionary to track how many records of each type (ok, error, retry) were processed in a data pipeline.
from collections import Counter
import random
processed_counter = Counter()
def process_records(records):
for record in records:
processed_counter[record] += 1
return len(records)
if __name__ == "__main__":
sample_records = [random.choice(["ok", "error", "retry"]) for _ in range(10)]
print(f…How to Build Data Processing Functions in Python
Create reusable helper functions to load, filter, transform, and aggregate CSV data in Python.
import csv
from pathlib import Path
def load_data(filepath):
"""Load CSV data into a list of dicts."""
with open(filepath, "r", newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
def filter_rows(rows, column, value):
"""Keep rows where column equals value."""
return [row for…How to Build a Simple Data Pipeline in Python
A beginner-friendly data pipeline that loads JSON, filters records by a field value, and aggregates counts per category.
import json
from pathlib import Path
def load_json(filepath: str | Path) -> list[dict]:
"""Load a JSON file containing a list of records."""
with Path(filepath).open("r", encoding="utf-8") as f:
return json.load(f)
def filter_records(records: list[dict], field: str, value) -> list[dict]:
"""Kee…How to Clean and Format Data in Python
This code loads JSON data, cleans records by removing empty fields and normalizing text, then summarizes the results with counts and unique keys.
import json
from pathlib import Path
def load_data(filepath: str) -> dict:
"""Load JSON data from a file."""
with Path(filepath).open("r", encoding="utf-8") as f:
return json.load(f)
def clean_records(records: list[dict]) -> list[dict]:
"""Remove empty fields and normalize text to lowercase."""…How to Compress Pipeline Output Gzip Per Partition in Python
Compress each partition of pipeline output into a separate gzip file and verify the compressed data by reading it back.
import gzip
import io
import random
from pathlib import Path
def compress_partition(partition_data: list[str], output_path: Path) -> int:
"""Compress a partition of data to a gzip file, returns bytes written."""
with gzip.open(output_path, 'wt', encoding='utf-8') as f:
f.writelines(partition_data)
…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 Count JSON Records in Python
Read a JSON file and count the number of top-level records, handling both list and dictionary structures.
import json
from pathlib import Path
def count_records(json_file):
"""Count top-level records in a JSON file."""
with open(json_file, "r") as f:
data = json.load(f)
# Handle both list of records and dict of records
if isinstance(data, list):
return len(data)
elif isinstance(da…How to Deduplicate Events with At-Least-Once Delivery in Python
Implements an exactly-once processing pattern for at-least-once event delivery by tracking seen event IDs in a set, skipping duplicates.
seen_ids = set()
def process_event(event_id: str, payload: dict) -> dict:
"""Process an event exactly once, ignoring duplicates."""
if event_id in seen_ids:
return {"status": "duplicate", "event_id": event_id}
seen_ids.add(event_id)
return {"status": "processed", "event_id": event_id, **payloa…How to Explode an Array Field into Multiple Rows in Python
This code flattens a list of dictionaries by exploding each array field value into its own row, duplicating the other fields as needed.
from collections import defaultdict
data = [
{"id": 1, "name": "Alice", "tags": ["python", "data", "ai"]},
{"id": 2, "name": "Bob", "tags": ["web", "devops"]},
{"id": 3, "name": "Carol", "tags": []},
]
def explode_array_field(records, array_field):
result = []
for record in records:
for v…How to Filter Data in Python
Filter a list of dictionaries by exact key-value matches or numerical ranges using concise list comprehensions.
from typing import List, Dict, Any
def filter_data(
data: List[Dict[str, Any]], key: str, value: Any
) -> List[Dict[str, Any]]:
"""Return records where data[key] equals value."""
return [record for record in data if record.get(key) == value]
def filter_by_range(
data: List[Dict[str, Any]], key: str…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 Implement a Sliding Window Average in Python
Compute the average of the most recent N values in a stream using a bounded deque, efficiently updating the total as new values arrive.
from collections import deque
class SlidingWindowAverage:
def __init__(self, window_size):
self.window_size = window_size
self.window = deque(maxlen=window_size)
self.total = 0
def add(self, value):
if len(self.window) == self.window_size:
self.total -= self.windo…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 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…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.