Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
Attach Source File Metadata to Records in Python
Add a source filename field to each record in a list by merging a new key into every dictionary using a dict unpacking comprehension.
from pathlib import Path
import json
def attach_source_metadata(records, source_file):
"""Attach source filename metadata to each record."""
return [
{**record, "source": Path(source_file).name}
for record in records
]
if __name__ == "__main__":
source = "/data/raw/customers.csv"
…Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets
A Python utility that uses pandas to find overlapping records across different Excel sheets based on specified key columns.
import pandas as pd
from pathlib import Path
def find_duplicate_records_across_sheets(file_path: str, key_columns: list, sheet_names: list) -> dict:
"""
Detect duplicate records across multiple Excel sheets based on specified key columns.
Args:
file_path: Path to the Excel file
key_co…Check Null Rate Threshold in PySpark DataFrame
This PySpark code checks the null rate of specified DataFrame columns against a threshold and returns violations.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count
def check_null_rate(df, threshold=0.2, columns=None):
"""
Check null rate for specified columns (or all) against a threshold.
Returns columns that exceed the threshold.
"""
cols = columns or df.columns
total…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…Create Data Helper Functions in Python for Beginners
Build reusable Python helper functions to load, filter, sort, summarize, and save JSON data — a beginner-friendly starting point for small data pipelines.
import json
from pathlib import Path
from typing import Any, Dict, List
def load_json_file(filepath: str) -> Dict[str, Any]:
"""Load JSON data from a file."""
with Path(filepath).open("r", encoding="utf-8") as file:
return json.load(file)
def filter_by_key(
data: List[Dict[str, Any]], key: str,…Deduplicate events by ID within a window in Python
Deduplicate event streams by ID within sliding time windows, keeping the newest occurrence per window using heaps and sets.
import heapq
from collections import defaultdict
def deduplicate_events(events, window_size):
"""Return events deduplicated by id, keeping newest within each sliding window."""
# Index events by (timestamp, id) for deterministic ordering
events_by_id = defaultdict(list)
for ts, eid, *payload in events…ETL in Python: Extract CSV, Transform Dict, Load JSON
Build a simple ETL pipeline in Python that reads a CSV file, transforms each row (stripping whitespace and converting numeric fields), and writes the result to JSON.
import csv
import json
from pathlib import Path
def extract_csv(file_path):
"""Read CSV file and return list of row dictionaries."""
with Path(file_path).open('r', newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
return list(reader)
def transform_dicts(rows):
"""Transform ro…ETL in Python: Extract CSV, Transform Dicts, Load JSON
Build a simple ETL pipeline that reads a CSV, normalizes keys and converts price to float, then writes structured JSON.
import csv
import json
from pathlib import Path
def etl_csv_to_json(csv_path: str, json_path: str) -> None:
"""Extract CSV, transform rows to dicts, load to JSON."""
with open(csv_path, mode='r', newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
records = list(reader)
# Trans…Enrich Events with Geo IP Data in Python
Returns a copy of each event dictionary, enriched with a geo-location dict from a mock IP-to-geo lookup table, with a fallback for unknown IPs.
import ipaddress
GEO_IP_DB = {
"192.168.1.10": {"country": "US", "city": "New York", "lat": 40.7128, "lon": -74.0060},
"10.0.0.5": {"country": "DE", "city": "Berlin", "lat": 52.5200, "lon": 13.4050},
"172.16.0.8": {"country": "JP", "city": "Tokyo", "lat": 35.6762, "lon": 139.6503},
}
EVENTS = [
{"id…Enrich a stream with reference data by key lookup in Python
Uses streamz to join each incoming record to a reference dictionary by name, adding department and level fields or defaults.
from streamz import Stream
reference = {"alice": {"dept": "eng", "level": 3}, "bob": {"dept": "sales", "level": 5}}
def enrich(record):
name = record.get("name")
ref = reference.get(name)
joined = dict(record)
if ref:
joined.update(ref)
else:
joined["dept"] = "unknown"
joi…Extract Schema.org Structured Data from Any Website in Python
A Python tool that fetches a webpage and extracts all JSON-LD structured data (Schema.org) embedded in <script> tags with type="application/ld+json".
import requests
from bs4 import BeautifulSoup
import json
def extract_schema_org(url):
"""Extract structured data (Schema.org) from a website."""
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
except requests.exceptions.RequestException as e:
return {"err…Fan Out Records to Multiple Sinks in Python
Distribute the same records across multiple target sinks (database, API, queue, etc.) using a defaultdict-based fan-out pattern.
import json
from collections import defaultdict
SINKS = ["database", "api", "message_queue", "data_lake", "monitoring"]
def fan_out(records, *sinks):
dist = defaultdict(list)
for record in records:
for sink in sinks:
dist[sink].append(record)
return dict(dist)
if __name__ == "__main_…Filter Records by Required Fields in Python
Filter a list of dictionaries, keeping only records where every required field is present and not None.
def filter_records(records, required_fields):
"""Return only records that have all required fields non-null."""
return [
record for record in records
if all(record.get(field) is not None for field in required_fields)
]
if __name__ == "__main__":
sample_records = [
{"name": "Al…Generate a Deterministic Hash for Deduplication in Python
Create a stable SHA-256 fingerprint from nested data and file contents to deduplicate records in a data pipeline.
import hashlib
import json
from pathlib import Path
def natural_key_hash(data, salt=""):
"""
Generate a deterministic fingerprint from raw data (dict/list/str).
Uses JSON canonical-ish serialization with sorted keys and SHA-256.
"""
canonical = json.dumps(data, sort_keys=True, separators=(",", ":"…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):
…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 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 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…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.