Reference library

Data pipelines & processing

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

6 matches

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…

15 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 easy

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.

json data cleaning data pipelines

Python

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."""…

13 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 Implement Incremental Load with Watermark by updated_at in Python

Load only new or changed rows into SQLite by comparing an updated_at timestamp against a stored watermark, returning counts and the new watermark.

incremental-load watermark sqlite

Python

import sqlite3
from datetime import datetime, timedelta


def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
    """Load only rows with updated_at greater than the last watermark."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Create table if it doesn't exist
  …

12 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.