Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
Add a UUID Surrogate Key to Each Row in a CSV with Python
Generate a unique UUID string for every row in a CSV file using the standard-library uuid and csv modules.
import uuid
import csv
def add_surrogate_key(filename):
with open(filename, newline='') as f_in:
reader = csv.DictReader(f_in)
rows = list(reader)
for row in rows:
row['surrogate_key'] = str(uuid.uuid4())
with open(filename, 'w', newline='') as f_out:
writer = csv.DictWri…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,…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 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 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 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 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…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)
…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.