Reference library

Data pipelines & processing

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

4 matches

Check Null Rate Threshold in PySpark DataFrame

This PySpark code checks the null rate of specified DataFrame columns against a threshold and returns violations.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count

def check_null_rate(df, threshold=0.2, columns=None):
    """
    Check null rate for specified columns (or all) against a threshold.
    Returns columns that exceed the threshold.
    """
    cols = columns or df.columns
    total…

14 0 Open

Data pipelines & processing easy

How to Validate Data in a Python Pipeline

A helper module to validate common record types — email, positive integer, and non-empty string list — before processing data in a pipeline.

data-validation pipelines type-checking

Python

from typing import Any, Iterable


def is_valid_email(email: str) -> bool:
    """Basic email check: one '@', no spaces, dot after '@'."""
    if "@" not in email or " " in email:
        return False
    local, _, domain = email.partition("@")
    return bool(local) and "." in domain


def is_positive_int(value: Any)…

12 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

Validate dict schema at pipeline boundary in Python

This code validates a dictionary against a TypedDict schema at a pipeline boundary, enforcing required fields and types with custom error messages.

validation dict typeddict

Python

from typing import Any, TypedDict


class Person(TypedDict):
    name: str
    age: int
    email: str


def validate_person(data: dict[str, Any]) -> Person:
    errors: list[str] = []

    if not isinstance(data.get("name"), str) or not data["name"].strip():
        errors.append("name must be a non-empty string")
  …

13 0 Open

Browse by section

Each section groups closely related Python snippets.