Reference library

Data pipelines & processing

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

5 matches

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.

json pipeline helpers

Python

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

14 0 Open

Data pipelines & processing easy

Generate a Mock CDC Changelog in Python

Simulate a CDC changelog with INSERT, UPDATE, and DELETE operations, timestamps, and record snapshots for testing data pipelines.

cdc changelog mock-data

Python

import json
from datetime import datetime, timedelta


def generate_mock_changelog(records, operations=("INSERT", "UPDATE", "DELETE")):
    """Simulate a CDC changelog from a list of record snapshots."""
    base_time = datetime(2025, 1, 1, 8, 0, 0)
    changelog = []
    for idx, record in enumerate(records):
       …

15 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 create a dated snapshot path for a dataset in Python

Generate a versioned directory path combining a base directory, dataset name, and today's date, ready for creating snapshots in data pipelines.

date pathlib datasets

Python

import datetime
import os
from pathlib import Path


def snapshot_path(base_dir: str, dataset_name: str) -> Path:
    """Return a dated snapshot path for a dataset under a base directory."""
    today = datetime.date.today().isoformat()
    return Path(base_dir) / dataset_name / today


if __name__ == "__main__":
    …

15 0 Open

Data pipelines & processing easy

How to route late-arriving data to a side output in Python

Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.

data pipelines streaming dead-letter

Python

from collections import defaultdict

def late_arriving_side_output(events, late_threshold_ts):
    """
    Mock a streaming pipeline that separates late-arriving data events
    into a side output list (e.g., for dead-letter analysis).

    events: list of (timestamp, data) tuples, timestamps as ints.
    late_thresho…

12 0 Open

Browse by section

Each section groups closely related Python snippets.