Reference library

Data pipelines & processing

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

4 matches

Deduplicate events by ID within a window in Python

Deduplicate event streams by ID within sliding time windows, keeping the newest occurrence per window using heaps and sets.

deduplication events heapq

Python

import heapq
from collections import defaultdict

def deduplicate_events(events, window_size):
    """Return events deduplicated by id, keeping newest within each sliding window."""
    # Index events by (timestamp, id) for deterministic ordering
    events_by_id = defaultdict(list)
    for ts, eid, *payload in events…

14 0 Open

Data pipelines & processing medium

How to Find Missing Values in Large Datasets in Python

Analyze missing values across multiple large pandas DataFrames with counts and percentages.

pandas missing-data data-cleaning

Python

import pandas as pd
import numpy as np

def find_missing_values_summary(datasets):
    """Analyze missing values across multiple datasets (dict of name: DataFrame)."""
    summary = {}
    for name, df in datasets.items():
        missing_count = df.isnull().sum()
        total_rows = len(df)
        missing_pct = (mi…

41 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

Idempotent Pipeline Dedupe by Record ID Set in Python

Filters records against a persistent set of seen IDs, returning only new ones and the updated set for idempotent pipeline processing.

deduplication idempotency pipelines

Python

def dedupe_records(records, seen_ids=None):
    """Return records whose id has not been seen before."""
    if seen_ids is None:
        seen_ids = set()
    unique = []
    for record in records:
        record_id = record.get("id")
        if record_id not in seen_ids:
            seen_ids.add(record_id)
           …

12 0 Open

Browse by section

Each section groups closely related Python snippets.