Reference library

Data pipelines & processing

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

5 matches

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…

11 0 Open

Data pipelines & processing easy

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.

pipeline json aggregation

Python

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…

10 0 Open

Data pipelines & processing medium

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.

datetime grouping time-window

Python

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,
…

12 0 Open

Data pipelines & processing easy

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.

grouping defaultdict data-aggregation

Python

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…

14 0 Open

Data pipelines & processing easy

How to Reduce Aggregate Counts from Mapped Chunks in Python

Combine a list of mapped chunk dictionaries into a single aggregated count dictionary using functools.reduce.

reduce aggregation dictionary

Python

from functools import reduce
from collections import defaultdict

def aggregate_chunks(mapped_chunks):
    """Combine mapped chunk counts into a single aggregate dict."""
    return reduce(
        lambda acc, chunk: {
            **acc,
            **{k: acc.get(k, 0) + v for k, v in chunk.items()}
        },
       …

14 0 Open

Browse by section

Each section groups closely related Python snippets.