Reference library

Data pipelines & processing

ETL-style flows, batch transforms, validation, and moving data between formats.

12 matches
Data pipelines & processing easy

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.

counter metrics data-pipeline
Python
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…
15 0 Open
Data pipelines & processing easy

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.

etl csv json
Python
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…
13 0 Open
Data pipelines & processing easy

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.

etl csv json
Python
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…
11 0 Open
Data pipelines & processing easy

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.

data-pipeline type-conversion json
Python
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…
11 0 Open
Data pipelines & processing easy

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.

json counting file-reading
Python
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…
12 0 Open
Data pipelines & processing easy

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.

grouping defaultdict data-pipelines
Python
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…
15 0 Open
Data pipelines & processing easy

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.

dead-letter-queue json logging
Python
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…
13 0 Open
Data pipelines & processing easy

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.

dict merge upsert
Python
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…
13 0 Open
Data pipelines & processing easy

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.

sorting dictionaries data-pipelines
Python
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": …
12 0 Open
Data pipelines & processing medium

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.

streaming jsonl large-files
Python
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 = …
13 0 Open
Data pipelines & processing medium

Map Partition Over Chunks in Python with Multiprocessing and Mock

Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.

multiprocessing chunking parallel
Python
from multiprocessing import Pool
from unittest.mock import patch, Mock

def process_chunk(chunk):
    return [x * x for x in chunk]

def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with Pool() as pool:
     …
12 0 Open
Data pipelines & processing easy

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.

pytest fixtures data-pipelines
Python
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(…
16 0 Open

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.