Cleaning

Bronze-to-silver DataFrame transformation functions.

All functions are pure: (DataFrame, **kwargs) -> DataFrame. No Spark actions are triggered — lazy evaluation is preserved throughout.

scamander.transformations.cleaning.add_audit_columns(df: DataFrame, source_file: str | None = None, batch_id: str | None = None) → DataFrame

Append standard audit columns to track data lineage.

Added columns:

• _ingestion_timestamp (TIMESTAMP): Time the transformation ran.

• _source_file (STRING): Optional path or name of the source file.

• _batch_id (STRING): Optional identifier for the pipeline run.

Parameters:

• df – Input DataFrame.

• source_file – Source file path or identifier.

• batch_id – Batch/run identifier (e.g. a UUID from the platform session).

Returns:

DataFrame with audit columns appended.

Example:

add_audit_columns(df, source_file="customers.csv", batch_id=batch_id)

scamander.transformations.cleaning.cast_to_schema(df: DataFrame, metadata: list[dict]) → DataFrame

Cast DataFrame columns to the types declared in a metadata column list.

Columns present in metadata but absent from df are silently skipped; extra df columns are preserved unchanged.

Parameters:

• df – Input DataFrame.

• metadata – List of column dicts as used by create_empty_table_from_metadata. Each dict must have at minimum ‘target’ and ‘type’ keys.

Returns:

DataFrame with columns cast to their declared types.

Example:

cast_to_schema(df, metadata["columns"])

scamander.transformations.cleaning.drop_columns(df: DataFrame, columns: list[str]) → DataFrame

Drop the specified columns when they are present.

Parameters:

• df – Input DataFrame.

• columns – Column names to remove. Names absent from df are ignored.

Returns:

DataFrame without the requested columns.

Example:

drop_columns(df, ["internal_notes", "raw_payload"])

scamander.transformations.cleaning.drop_duplicates(df: DataFrame, subset: list[str] | None = None) → DataFrame

Remove duplicate rows, optionally considering only a subset of columns.

Parameters:

• df – Input DataFrame.

• subset – Column names to consider for deduplication. When None, all columns are used.

Returns:

DataFrame without duplicates.

Example:

drop_duplicates(df, subset=["customer_id"])

scamander.transformations.cleaning.filter_nulls(df: DataFrame, columns: list[str]) → DataFrame

Remove rows where any of the specified columns contains a null.

Parameters:

• df – Input DataFrame.

• columns – Columns that must be non-null for a row to be retained. Columns absent from df are ignored.

Returns:

DataFrame with null rows removed.

Example:

filter_nulls(df, ["customer_id", "email"])

scamander.transformations.cleaning.flatten_struct(d: dict, parent_key: str = '', sep: str = '__') → dict

Recursively flatten a nested dict to a single level.

Lists are JSON-serialised to strings so no data is lost. All scalar values are cast to str; None stays None.

This is the Python-level companion to PySpark’s from_json / struct flattening — use it before creating a Spark DataFrame when the source data arrives as Python dicts (e.g. from reading raw JSON files).

Parameters:

• d – Input dict, potentially containing nested dicts and lists.

• parent_key – Key prefix accumulated during recursion (empty at top level).

• sep – Separator inserted between parent and child key segments. Default '__'.

Returns:

Flat dict where every value is a str or None.

Example:

flatten_struct({"a": {"b": 1}, "c": [1, 2], "d": None})
# → {"a__b": "1", "c": "[1, 2]", "d": None}

scamander.transformations.cleaning.normalize_locale_decimal_columns(df: DataFrame, columns: list[str] | None = None, decimal_type: str = 'DECIMAL(10,4)', decimal_separator: str = ',', null_markers: tuple[str, ...] = ('', '.')) → DataFrame

Normalize locale-formatted decimal strings and cast them to a decimal type.

This handles common CSV exports where decimal values are represented as text with a comma separator, while empty strings or placeholder markers such as "." mean null.

Parameters:

• df – Input DataFrame.

• columns – Columns to normalize. When None, all columns are processed.

• decimal_type – Spark SQL decimal type used for casting.

• decimal_separator – Separator to replace with "." before casting.

• null_markers – Trimmed string values that should become null.

Returns:

DataFrame with normalized decimal columns.

Example:

# "1.234,56" -> 1234.56
normalize_locale_decimal_columns(df, columns=["revenue"], decimal_type="DECIMAL(12,2)")

scamander.transformations.cleaning.normalize_to_list(data: dict, key: str) → dict

Ensure a dict field is always a list.

Useful when an API returns null, a single object, or a list for the same field depending on the number of results (common in Dutch government APIs).

Modifies data in-place and also returns it for convenience.

Parameters:

• data – Input dict.

• key – The key whose value should be normalised to a list.

Returns:

The same dict with data[key] guaranteed to be a list.

Example:

normalize_to_list(record, "children")

scamander.transformations.cleaning.rename_columns(df: DataFrame, mapping: dict[str, str]) → DataFrame

Rename columns according to a source → target mapping.

Columns not present in the mapping are left unchanged.

Parameters:

• df – Input DataFrame.

• mapping – Dict of {source_name: target_name}.

Returns:

DataFrame with renamed columns.

Example:

rename_columns(df, {"klant_id": "customer_id"})

scamander.transformations.cleaning.replace_empty_strings(df: DataFrame, replacement: str | None = None) → DataFrame

Replace empty strings with null (default) or a custom value in all STRING columns.

Parameters:

• df – Input DataFrame.

• replacement – Value to substitute for empty strings. None means null.

Returns:

DataFrame with empty strings replaced.

Example:

replace_empty_strings(df)

scamander.transformations.cleaning.sanitize_column_names(df: DataFrame, replacement: str = '', collapse_replacement: bool = True, strip_replacement: bool = True) → DataFrame

Remove or replace characters that are unsafe in Spark/Delta column names.

Unlike standardize_column_names, this function preserves case and focuses on removing punctuation and other invalid characters from externally supplied file headers.

Parameters:

• df – Input DataFrame.

• replacement – Replacement for every character outside [a-zA-Z0-9_]. Defaults to an empty string, matching legacy notebook behaviour.

• collapse_replacement – Collapse repeated replacement characters into one. Only applies when replacement is not empty.

• strip_replacement – Remove replacement characters from the start/end of the resulting column name. Only applies when replacement is not empty.

Returns:

DataFrame with sanitized column names.

Example:

# "Klant#ID (oud)" -> "KlantID oud"
sanitize_column_names(df)

scamander.transformations.cleaning.select_columns(df: DataFrame, columns: list[str]) → DataFrame

Select only the specified columns, dropping all others.

Columns listed but absent from df are silently ignored.

Parameters:

• df – Input DataFrame.

• columns – Column names to keep.

Returns:

DataFrame with only the requested columns (in the given order).

Example:

select_columns(df, ["customer_id", "name", "email"])

scamander.transformations.cleaning.standardize_column_names(df: DataFrame) → DataFrame

Lowercase all column names and replace spaces and hyphens with underscores.

Parameters:

df – Input DataFrame.

Returns:

DataFrame with standardized column names.

Example:

# "Customer ID" -> "customer_id"
standardize_column_names(df)

scamander.transformations.cleaning.trim_strings(df: DataFrame) → DataFrame

Trim leading and trailing whitespace from all STRING columns.

Parameters:

df – Input DataFrame.

Returns:

DataFrame with all string columns trimmed.

Example:

trim_strings(df)