Transformer

Fluent DataFrame transformation builder.

Wraps the functions in cleaning and enrichment modules behind a chainable API that optionally integrates with the platform logger.

class scamander.transformations.transformer.Transformer(df: DataFrame, log: HumanLogger | None = None)

Basisklassen: object

Fluent builder for platform-independent DataFrame transformations.

Each method applies a transformation and returns self, enabling method chaining. Retrieve the resulting DataFrame via the .df property.

Parameters:

• df – The starting DataFrame.

• log – Optional HumanLogger from the active Platform instance. When supplied, each transformation step is logged at DEBUG level.

Example:

# Bronze → Silver
silver_df = (
    Transformer(bronze_df, log=platform.log)
    .cast_to_schema(metadata["columns"])
    .rename_columns({"klant_id": "customer_id"})
    .trim_strings()
    .replace_empty_strings()
    .filter_nulls(["customer_id"])
    .drop_duplicates(subset=["customer_id"])
    .add_audit_columns(source_file="customers.csv")
    .df
)

# Silver → Gold
gold_df = (
    Transformer(silver_df, log=platform.log)
    .join(platform.spark.read.table("silver.orders"),
          on="customer_id", how="left", broadcast_right=True)
    .aggregate(
        group_by=["region"],
        aggregations=[
            {"column": "revenue",  "function": "sum",   "alias": "total_revenue"},
            {"column": "order_id", "function": "count", "alias": "order_count"},
        ],
    )
    .df
)

add_audit_columns(source_file: str | None = None, batch_id: str | None = None) → Transformer

Append _ingestion_timestamp, _source_file, and _batch_id audit columns.

add_derived_columns(expressions: dict[str, Column | str]) → Transformer

Add columns computed from Column expressions or SQL strings.

aggregate(group_by: list[str], aggregations: list[dict]) → Transformer

Group by columns and apply aggregation functions.

Each entry in aggregations must be a dict with ‘column’, ‘function’, and ‘alias’ keys. See enrichment.aggregate for supported functions.

anonymize(config: AnonymizationConfig) → Transformer

Apply an anonymization config to the current DataFrame.

Rules are applied in the order they appear in config. This step returns self so the chain can continue.

Parameters:

config – Anonymization configuration with one rule per column.

Returns:

This Transformer with anonymization rules applied.

Veroorzaakt:

ImportError – When a rule requires cryptography or faker and the [anonymization] extra is not installed.

Example:

from scamander.models.anonymization_config import AnonymizationConfig, AnonymizationRule

config = AnonymizationConfig(
    name="customer_pii",
    rules=[
        AnonymizationRule(column="full_name", method="fake", faker_provider="name"),
        AnonymizationRule(column="email",     method="hash"),
        AnonymizationRule(column="phone",     method="mask", visible_suffix=4),
        AnonymizationRule(column="age",       method="generalize",
                          bins=[0.0, 18.0, 65.0, 120.0],
                          labels=["child", "adult", "senior"]),
    ],
)

silver_df = (
    Transformer(bronze_df, log=platform.log)
    .trim_strings()
    .filter_nulls(["customer_id"])
    .anonymize(config)
    .df
)

cast_to_schema(metadata: list[dict]) → Transformer

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

deduplicate_with_ranking(partition_by: list[str], order_by: list[str], descending: bool = True) → Transformer

Keep the highest-ranked row per partition using row_number().

Parameters:

• partition_by – Columns defining the uniqueness key.

• order_by – Columns to determine rank within each partition.

• descending – Keep the row with the highest value when True (default).

property df: DataFrame

The current transformed DataFrame.

drop_columns(columns: list[str]) → Transformer

Drop the specified columns when they are present.

drop_duplicates(subset: list[str] | None = None) → Transformer

Remove duplicate rows, optionally scoped to a column subset.

filter_nulls(columns: list[str]) → Transformer

Remove rows where any of the listed columns is null.

join(other: DataFrame, on: str | list[str], how: str = 'inner', broadcast_right: bool = False) → Transformer

Join the current DataFrame with another.

Parameters:

• other – Right DataFrame.

• on – Join column(s).

• how – Join type — inner, left, right, full, left_semi, left_anti.

• broadcast_right – Broadcast the right DataFrame (for small lookup tables).

normalize_locale_decimal_columns(columns: list[str] | None = None, decimal_type: str = 'DECIMAL(10,4)', decimal_separator: str = ',', null_markers: tuple[str, ...] = ('', '.')) → Transformer

Normalize locale-formatted decimal strings and cast them to decimals.

pivot_table(group_by: list[str], pivot_col: str, agg_col: str, agg_func: str = 'sum', pivot_values: list[str] | None = None) → Transformer

Pivot a column into multiple columns, aggregating values.

Parameters:

• group_by – Columns to keep as rows.

• pivot_col – Column whose distinct values become column headers.

• agg_col – Column to aggregate into each cell.

• agg_func – Aggregation function name. Default ‘sum’.

• pivot_values – Explicit pivot values to skip the auto-discovery scan.

rename_columns(mapping: dict[str, str]) → Transformer

Rename columns using a source → target mapping dict.

replace_empty_strings(replacement: str | None = None) → Transformer

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

sanitize_column_names(replacement: str = '', collapse_replacement: bool = True, strip_replacement: bool = True) → Transformer

Remove or replace unsafe characters in column names while preserving case.

select_columns(columns: list[str]) → Transformer

Keep only the specified columns, dropping all others.

standardize_column_names() → Transformer

Lowercase column names and replace spaces/hyphens with underscores.

trim_strings() → Transformer

Trim leading/trailing whitespace from all STRING columns.

validate(suite_config: QualitySuiteConfig) → QualityResult

Run a Great Expectations quality suite against the current DataFrame.

This is a terminal inspection step — it returns a QualityResult rather than self, so the chain ends here. Access .df beforehand if you need the DataFrame after validation.

Parameters:

suite_config – Quality suite definition.

Returns:

QualityResult with per-expectation outcomes.

Veroorzaakt:

DataQualityError – When suite_config.raise_on_failure is True and one or more expectations fail.

Example:

result = (
    Transformer(df, log=platform.log)
    .trim_strings()
    .filter_nulls(["id"])
    .validate(suite_config)
)
if not result.success:
    raise RuntimeError(str(result))