Reference library

Data pipelines & processing

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

6 matches

Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets

A Python utility that uses pandas to find overlapping records across different Excel sheets based on specified key columns.

pandas excel data cleaning

Python

import pandas as pd
from pathlib import Path

def find_duplicate_records_across_sheets(file_path: str, key_columns: list, sheet_names: list) -> dict:
    """
    Detect duplicate records across multiple Excel sheets based on specified key columns.
    
    Args:
        file_path: Path to the Excel file
        key_co…

46 0 Open

Data pipelines & processing medium

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 easy

Generate a Deterministic Hash for Deduplication in Python

Create a stable SHA-256 fingerprint from nested data and file contents to deduplicate records in a data pipeline.

hashing deduplication sha256

Python

import hashlib
import json
from pathlib import Path

def natural_key_hash(data, salt=""):
    """
    Generate a deterministic fingerprint from raw data (dict/list/str).
    Uses JSON canonical-ish serialization with sorted keys and SHA-256.
    """
    canonical = json.dumps(data, sort_keys=True, separators=(",", ":"…

15 0 Open

Data pipelines & processing easy

How to Deduplicate Events with At-Least-Once Delivery in Python

Implements an exactly-once processing pattern for at-least-once event delivery by tracking seen event IDs in a set, skipping duplicates.

deduplication idempotent event-processing

Python

seen_ids = set()

def process_event(event_id: str, payload: dict) -> dict:
    """Process an event exactly once, ignoring duplicates."""
    if event_id in seen_ids:
        return {"status": "duplicate", "event_id": event_id}
    seen_ids.add(event_id)
    return {"status": "processed", "event_id": event_id, **payloa…

13 0 Open

Data pipelines & processing medium

How to Validate Fact Table Grain Row Counts in Python

Validate fact table grain by checking dimension key references, unique grain combinations, duplicate rows, and dimension cardinality from a CSV file.

csv data validation etl

Python

import csv
import hashlib
from pathlib import Path


def validate_fact_grain(fact_file: Path, expected_dim_keys: dict[str, set[str]]) -> dict:
    """
    Validate fact table grain by checking each row's dimension keys exist
    in expected dimension tables and row count consistency.
    """
    dim_references = {}
  …

13 0 Open

Data pipelines & processing easy

Implement Exactly-Once Transaction Log in Python

A mock transaction log that deduplicates transaction IDs so each is recorded only once, with a dataclass for records and simple in-memory storage.

transactions deduplication dataclass

Python

from dataclasses import dataclass
from typing import Dict, Optional


@dataclass
class TxnRecord:
    txn_id: str
    status: str


class ExactlyOnceTxnLog:
    def __init__(self) -> None:
        self._log: Dict[str, TxnRecord] = {}
        self._processed_ids: set = set()

    def record(self, txn_id: str, status: s…

14 0 Open

Browse by section

Each section groups closely related Python snippets.