Reference library

Data pipelines & processing

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

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

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

43 0 Open

Data pipelines & processing medium

How to Implement Slowly Changing Dimension Type 2 History in Python

Build a type-2 slowly changing dimension pipeline that closes old records and opens new ones when customer data changes.

scd dimension history

Python

from datetime import datetime, timedelta

def apply_scd_type2(records, current_date):
    """Returns active records after inserting new records with type-2 history."""
    history = []
    active = {}

    for record in records:
        key = record["customer_id"]
        if key in active:
            active[key]["end…

13 0 Open

Data pipelines & processing medium

Map Partition Over Chunks in Python with Multiprocessing and Mock

Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.

multiprocessing chunking parallel

Python

from multiprocessing import Pool
from unittest.mock import patch, Mock

def process_chunk(chunk):
    return [x * x for x in chunk]

def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with Pool() as pool:
     …

12 0 Open

Browse by section

Each section groups closely related Python snippets.