Enrichment

Silver-to-gold DataFrame transformation functions.

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

scamander.transformations.enrichment.add_derived_columns(df: DataFrame, expressions: dict[str, Column | str]) → DataFrame

Add or overwrite columns computed from PySpark Column expressions or SQL strings.

SQL strings are wrapped in F.expr() automatically, so any valid Spark SQL expression can be used without importing Column types.

Parameters:

• df – Input DataFrame.

• expressions – Dict of {alias: Column_expression_or_sql_string}.

Returns:

DataFrame with the new or overwritten columns.

Example:

add_derived_columns(df, {
    "full_name":        F.concat_ws(" ", F.col("first_name"), F.col("last_name")),
    "revenue_per_unit": "revenue / quantity",
    "is_premium":       "total_spend > 10000",
})

scamander.transformations.enrichment.aggregate(df: DataFrame, group_by: list[str], aggregations: list[dict]) → DataFrame

Group and aggregate a DataFrame.

Parameters:

• df – Input DataFrame.

• group_by – Column names to group by.

• aggregations –

  List of aggregation specs. Each dict must have: - ‘column’ (str): Column to aggregate. - ‘function’ (str): One of: sum, count, avg, mean, min, max, first,

  last, count_distinct, stddev, variance, collect_list, collect_set.

  • ’alias’ (str): Name for the resulting column.

Returns:

Aggregated DataFrame.

Veroorzaakt:

ValueError – If an unknown aggregation function is specified.

Example:

aggregate(df, group_by=["region"], aggregations=[
    {"column": "revenue",  "function": "sum",          "alias": "total_revenue"},
    {"column": "order_id", "function": "count_distinct","alias": "unique_orders"},
])

scamander.transformations.enrichment.deduplicate_with_ranking(df: DataFrame, partition_by: list[str], order_by: list[str], descending: bool = True) → DataFrame

Keep one row per partition, selected by rank on an ordering column.

Uses row_number over a Window — no extra shuffle beyond what grouping already requires. Typically used to retain the most-recent record per entity key.

Parameters:

• df – Input DataFrame.

• partition_by – Columns defining the uniqueness key (e.g. [‘customer_id’]).

• order_by – Columns to sort within each partition to determine rank 1.

• descending – When True (default), the highest value ranks first (e.g. retain the most-recent record by a timestamp column).

Returns:

DataFrame with exactly one row per partition (the rank-1 row).

Example:

# Keep only the most-recent row per customer_id
deduplicate_with_ranking(df, partition_by=["customer_id"], order_by=["updated_at"])

scamander.transformations.enrichment.join_dataframes(left: DataFrame, right: DataFrame, on: str | list[str], how: str = 'inner', broadcast_right: bool = False) → DataFrame

Join two DataFrames with an optional broadcast hint for the right side.

Broadcast the right DataFrame when it is small enough to fit in executor memory (typically < 200 MB) to eliminate the shuffle join.

Parameters:

• left – Left DataFrame.

• right – Right DataFrame.

• on – Column name or list of column names to join on.

• how – Join type — ‘inner’, ‘left’, ‘right’, ‘full’, ‘left_semi’, ‘left_anti’, ‘cross’. Defaults to ‘inner’.

• broadcast_right – Apply a broadcast hint to the right DataFrame. Defaults to False.

Returns:

Joined DataFrame.

Example:

join_dataframes(orders_df, customers_df, on="customer_id",
                 how="left", broadcast_right=True)

scamander.transformations.enrichment.pivot_table(df: DataFrame, group_by: list[str], pivot_col: str, agg_col: str, agg_func: str = 'sum', pivot_values: list[str] | None = None) → DataFrame

Pivot a column into multiple columns, aggregating values.

Providing explicit pivot_values avoids a full scan to discover distinct values — always preferred in production when the value set is known.

Parameters:

• df – Input DataFrame.

• group_by – Columns to keep as rows.

• pivot_col – Column whose distinct values become the new column headers.

• agg_col – Column to aggregate into each pivoted cell.

• agg_func – Aggregation function (key from _AGG_FUNCTIONS). Default ‘sum’.

• pivot_values – Optional explicit list of pivot values. Skips the auto-discovery scan when provided.

Returns:

Pivoted DataFrame.

Veroorzaakt:

ValueError – If an unknown aggregation function is specified.

Example:

pivot_table(df, group_by=["region"], pivot_col="product",
            agg_col="revenue", agg_func="sum")