Data pipelines & processing
ETL-style flows, batch transforms, validation, and moving data between formats.
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 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 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 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 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 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 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 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…Test a Python Pipeline with Fixture Sample Rows
Test pipeline functions with sample rows provided by a pytest fixture, verifying required keys and value constraints.
import pytest
def get_value(data: dict, key: str):
return data.get(key)
def sample_rows():
return [
{"name": "Alice", "age": 30, "city": "London"},
{"name": "Bob", "age": 25, "city": "Paris"},
{"name": "Charlie", "age": 35, "city": "Berlin"},
]
@pytest.fixture
def sample_data(…Union Multiple DataFrames with Aligned Columns in Python
Concatenate DataFrames with different columns, aligning them and filling missing values with NaN using pandas concat.
import pandas as pd
from io import StringIO
# Sample dataframes with different columns
df1 = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'age': [25, 30, 35]
})
df2 = pd.DataFrame({
'id': [4, 5],
'name': ['Diana', 'Eve'],
'city': ['NYC', 'LA']
})
df3 = pd.DataFrame({
…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.