Module Deep Dives

transformations — 15 DataFrame Transforms

Reusable, composable functions that each take a DataFrame and return a new DataFrame. Designed to be chained using PySpark's .transform() method.

Column Cleaning

from pyspark_utils.transformations import clean_column_names, rename_columns_bulk

# Normalize all column names to lowercase snake_case
df = clean_column_names(df)
# "Order ID" → "order_id", "customer-name" → "customer_name"

# Bulk rename with a mapping dict
df = rename_columns_bulk(df, {"old_col": "new_col", "amt": "amount"})

Type Casting

from pyspark_utils.transformations import cast_columns, type_cast_dataframe

# Cast specific columns
df = cast_columns(df, {"amount": "double", "quantity": "integer"})

# Auto-cast all columns based on a target schema
df = type_cast_dataframe(df, target_schema)

Struct & Array Operations

from pyspark_utils.transformations import flatten_struct, explode_array

# Flatten nested structs into top-level columns (address.city → address_city)
df = flatten_struct(df)

# Explode an array column with position index
df = explode_array(df, array_column="line_items", position_alias="item_index")

Pivoting & Unpivoting

from pyspark_utils.transformations import pivot_aggregate, unpivot

# Pivot: rows → columns
df_pivoted = pivot_aggregate(
    df,
    group_columns=["customer_id"],
    pivot_column="metric_name",
    agg_column="metric_value",
    agg_function="sum",
)

# Unpivot: columns → rows
df_unpivoted = unpivot(
    df,
    id_columns=["customer_id"],
    value_columns=["revenue", "cost", "profit"],
    var_col_name="metric",
    val_col_name="value",
)

Metadata & Hashing

Included with PySpark Utils Library

Ready for the complete PySpark Utils Library?