Transformations DataFrame

Column name resolution (physical vs clean_data-style normalized) and helpers.

Filter DataFrames by a list of values on one column (type-aware, no casts).

Join a main DataFrame to a right side with prefixed, normalized column names.

Row-level filters based on how many columns are empty (see empty_or_null).

Column expressions for normalized text and dimension defaults.

Extract year and month integers from free-form French text labels (Spark Column expressions).

Unpivot and reshape wide month columns that share a block suffix (e.g. `` [CA Monthly]``).

DataFrame transforms: column tools, filters, merges, and wide-month reshaping.

See submodules fabrictools.transform.columns, fabrictools.transform.filter, fabrictools.transform.merge, fabrictools.transform.pivot, fabrictools.transform.rows, fabrictools.transform.text, fabrictools.transform.fr_text_period, fabrictools.transform.wide_month_suffix.

class fabrictools.transform.ParamResolver(df_params: pyspark.sql.DataFrame, *, key_col: str = 'parametre', value_col: str = 'valeur', small_threshold: int = 10000, row_count_strategy: Literal['estimate', 'count', 'auto'] = 'auto', duplicate_policy: Literal['first', 'last', 'max', 'min', 'error'] = 'last', order_col: str | None = None, persist_params: bool = False, spark_map_cache: bool = False, broadcast_local_map: bool = True, allow_count_fallback: bool = True, estimate_sample_fraction: float | None = None, logger: _LoggerLike | None = None)

Bases: object

Resolve parameter values from a Spark DataFrame with adaptive caching.

The resolver accepts a two-column parameter DataFrame, deduplicates it according to duplicate_policy, then chooses a lookup strategy.

Performance notes:

• Small parameter tables are collected once into a Python dict. Startup cost is O(n), driver memory is O(n), and each later get() is O(1) without launching another Spark job.

• Medium and large parameter tables stay in Spark. Each get() uses filter + select + first and launches one Spark job. Use this for occasional lookups only.

• spark_map_cache=True builds a single-row Spark MapType cache via map_from_entries(collect_list(...)). It is useful when parameter resolution should happen inside Spark expressions, but it aggregates all parameters and should be enabled deliberately.

• For enriching a large DataFrame, use enrich(), which performs a Spark join and can broadcast the deduplicated parameter table.

If duplicate_policy="last" or "first" is used without order_col, Spark has no stable row order. A warning is emitted because the chosen value is not deterministic across executions.

Example

>>> resolver = ParamResolver(
...     df_params,
...     key_col="parametre",
...     value_col="valeur",
...     small_threshold=10_000,
...     row_count_strategy="auto",
...     duplicate_policy="last",
...     order_col="ts",
... )
>>> x_value = resolver.get("x")
>>> enriched = resolver.enrich(
...     df_data,
...     key_col="parametre",
...     output_col="param_value",
... )

Parameters:

• df_params – Parameter DataFrame.

• key_col – Column containing parameter names.

• value_col – Column containing parameter values.

• small_threshold – Maximum row count for driver-side dictionary mode.

• row_count_strategy – "estimate", "count", or "auto".

• duplicate_policy – How to resolve duplicate keys.

• order_col – Optional ordering column for first and last.

• persist_params – Persist the deduplicated parameter DataFrame.

• spark_map_cache – Build a Spark MapType cache for Spark-side use.

• broadcast_local_map – Broadcast the local dictionary when using the small-table strategy.

• allow_count_fallback – Allow auto to run an exact count() when estimates and bounded counts are insufficient.

• estimate_sample_fraction – Optional sampling fraction used by row_count_strategy="estimate" when Catalyst statistics are missing. Sampling still launches a Spark job; leave as None to avoid it.

• logger – Optional logger with debug, info and warning.

clear() → None

Release broadcast and persisted Spark caches owned by this resolver.

enrich(df_data: pyspark.sql.DataFrame, *, key_col: str = 'parametre', output_col: str | None = None, how: str = 'left', broadcast_params: bool = True) → pyspark.sql.DataFrame

Join parameter values onto df_data.

This is the recommended strategy when a large DataFrame needs parameter values in Spark. It avoids repeated driver lookups and can broadcast the deduplicated parameter table.

Parameters:

• df_data – DataFrame to enrich.

• key_col – Key column in df_data.

• output_col – Output value column. Defaults to value_col.

• how – Spark join type.

• broadcast_params – Wrap the parameter table with broadcast().

Returns:

Enriched DataFrame.

get(key: Any, default: Any = None) → Any

Return the value associated with key.

For small parameter tables this method initializes a local dictionary on first use and then avoids further Spark jobs. For larger parameter tables it performs a Spark filter + select + first lookup per call.

Parameters:

• key – Parameter key to resolve.

• default – Value returned when the key is absent.

Returns:

The resolved parameter value or default.

property lookup_mode: Literal['local_dict', 'spark_filter', 'spark_map'] | None

Return the selected lookup mode after the first lookup, if any.

property row_count_info: RowCountInfo

Return cached row-count information, computing it once if needed.

spark_map() → pyspark.sql.DataFrame

Return a one-row DataFrame containing a Spark MapType column.

The returned DataFrame has one column named params_map. Building it aggregates the deduplicated parameters into a single Spark map and materializes it once with count() when spark_map_cache is enabled.

fabrictools.transform.build_tcd(df: pyspark.sql.DataFrame, rows: str | Sequence[str] | None = None, columns: str | Sequence[str] | None = None, values: str | Sequence[str] | Dict[str, str] | None = None, filters: str | None = None, custom_columns_names: Sequence[str] | None = None) → pyspark.sql.DataFrame

Build a Pivot Table (TCD) from a DataFrame, similar to Excel.

Parameters:

• df (DataFrame) – Source dataframe.

• rows (str | collections.abc.Sequence[str] | None) – Column(s) to group by (rows of the pivot table).

• columns (str | collections.abc.Sequence[str] | None) – Column(s) to pivot (columns of the pivot table).

• values (str | collections.abc.Sequence[str] | dict[str, str] | None) – Column(s) to aggregate, or a dict mapping column names to aggregation functions (e.g., {"amount": "sum", "id": "count"}). Defaults to sum for numeric columns, count for others.

• filters (str | None) – Optional SQL filter expression to apply before pivoting.

• custom_columns_names (collections.abc.Sequence[str] | None) – Optional list of names to rename all output columns in order. Must match the exact number of resulting columns.

Returns:

Pivoted dataframe.

Return type:

DataFrame

Example

>>> import pandas as pd
>>> from pyspark.sql import SparkSession
>>> spark = SparkSession.builder.getOrCreate()
>>> data = [
...     {"Year": 2023, "Region": "North", "Product": "A", "Sales": 100},
...     {"Year": 2023, "Region": "North", "Product": "B", "Sales": 150},
...     {"Year": 2023, "Region": "North", "Product": "C", "Sales": 50},
...     {"Year": 2023, "Region": "South", "Product": "A", "Sales": 200},
...     {"Year": 2024, "Region": "North", "Product": "A", "Sales": 120},
...     {"Year": 2022, "Region": "South", "Product": "C", "Sales": 80},
... ]
>>> df = spark.createDataFrame(pd.DataFrame(data))
>>> # TCD: Rows = Region, Columns = Year, Values = Sum of Sales, Filter = Product A or C and Year > 2022
>>> tcd_df = build_tcd(
...     df,
...     rows="Region",
...     columns="Year",
...     values={"Sales": "sum"},
...     filters="Product IN ('A', 'C') AND Year > 2022",
...     custom_columns_names=["Region", "Year 2023", "Year 2024"]
... )
>>> tcd_df.show()
+------+---------+---------+
|Region|Year 2023|Year 2024|
+------+---------+---------+
| North|      150|      120|
| South|      200|     null|
+------+---------+---------+

fabrictools.transform.cast_columns(df: pyspark.sql.DataFrame, type_map: dict[str, str | pyspark.sql.types.DataType]) → pyspark.sql.DataFrame

Cast multiple columns to new types using a mapping dictionary.

Uses physical column names, clean_data-style normalized labels, or snake_case to resolve columns in the dataframe. Non-matching keys are ignored.

Parameters:

• df (DataFrame) – Input dataframe.

• type_map (dict[str, str | DataType]) – A dictionary mapping column names to target Spark types (as strings or DataType).

Returns:

Dataframe with cast columns.

Return type:

DataFrame

Example

>>> df_cast = cast_columns(df, {
...     "Mon ID Client": "int",
...     "Montant_Total": "decimal(18,2)",
...     "date_creation": "date",
... })

fabrictools.transform.coalesce_dim(src: pyspark.sql.Column) → pyspark.sql.Column

String cast of src; null or blank becomes the literal 0 as string (dimension-friendly).

Parameters:

src (Column) – Source column.

Returns:

String Column.

Return type:

Column

Example

>>> df.withColumn("dim_id", coalesce_dim(F.col("legacy_code")))

fabrictools.transform.dataframe_last_nonnull_wide_month_from_long(long_df: pyspark.sql.DataFrame, *, order_column: str, variable_column: str = 'MoisCol', value_column: str = 'Valeur', month_start_column: str = 'MonthStart', output_month_start: str = 'MonthStart', output_year: str = 'Year', output_month: str = 'Month', output_value: str = 'Value') → pyspark.sql.DataFrame

For each distinct variable_column, keep the row with greatest order_column where value_column is non-null; emit typed month/value columns.

Parameters:

• long_df (DataFrame) – Long dataframe (e.g. from dataframe_unpivot_wide_month_suffix()).

• order_column (str) – Tie-break column (descending); must exist on long_df.

• variable_column (str) – Month variable name column.

• value_column (str) – Measure column.

• month_start_column (str) – Parsed month start on long_df.

• output_month_start (str) – Output date column name.

• output_year (str) – Output year column name.

• output_month (str) – Output month-of-year column name.

• output_value (str) – Output numeric value column name.

Returns:

One row per variable_column with cast types, or empty schema if inputs missing.

Return type:

DataFrame

Example

>>> latest = dataframe_last_nonnull_wide_month_from_long(
...     long_df, order_column="as_of_date"
... )

fabrictools.transform.dataframe_pivot_category_wide_month_from_long(long_df: pyspark.sql.DataFrame, *, category_column: str, pivot_categories: Sequence[str], fill_value: float = 0.0, variable_column: str = 'MoisCol', value_column: str = 'Valeur', month_start_column: str = 'MonthStart', output_year: str = 'Year', output_month: str = 'Month', montant_column: str = 'Montant') → pyspark.sql.DataFrame

Sum value_column by month_start_column and category_column, pivot categories wide, add year/month columns.

Parameters:

• long_df (DataFrame) – Long dataframe with month, category, and value.

• category_column (str) – Dimension to pivot.

• pivot_categories (collections.abc.Sequence[str]) – Category values that become column names.

• fill_value (float) – Fill null pivot cells after aggregation.

• variable_column (str) – Variable column name (must exist on long_df for early-exit checks).

• value_column (str) – Measure to sum.

• month_start_column (str) – Date key for grouping.

• output_year (str) – Name of year output column.

• output_month (str) – Name of month output column.

• montant_column (str) – Internal aggregate column name before pivot.

Returns:

Wide dataframe Year, Month, one column per category.

Return type:

DataFrame

Raises:

ValueError – If pivot_categories is empty.

Example

>>> wide = dataframe_pivot_category_wide_month_from_long(
...     long_df,
...     category_column="cost_type",
...     pivot_categories=("Actual", "Forecast"),
... )

fabrictools.transform.dataframe_unpivot_wide_month_suffix(df: pyspark.sql.DataFrame, *, id_columns: ~collections.abc.Sequence[str], value_columns_suffix: str | None = None, value_columns: ~collections.abc.Sequence[str] | None = None, exclude_columns: ~collections.abc.Collection[str] = (), variable_column: str = 'MoisCol', value_column: str = 'Valeur', month_start_column: str = 'MonthStart', month_start_from_column_name: ~collections.abc.Callable[[str], ~datetime.date | None] = <function month_start_from_ca_monthly_col>) → pyspark.sql.DataFrame

Unpivot wide month columns to long form and parse month_start_column from the variable name.

If value_columns is set, it takes precedence over value_columns_suffix.

Parameters:

• df (DataFrame) – Wide dataframe.

• id_columns (collections.abc.Sequence[str]) – Identifier columns kept as-is; labels that do not resolve on df are omitted.

• value_columns_suffix (str | None) – Suffix selecting value columns (via wide_value_columns()).

• value_columns (collections.abc.Sequence[str] | None) – Explicit list of value column names (optional).

• exclude_columns (collections.abc.Collection[str]) – Excluded from value detection when using suffix.

• variable_column (str) – Unpivot variable column name.

• value_column (str) – Unpivot value column name.

• month_start_column (str) – Output column for parsed month start dates.

• month_start_from_column_name (collections.abc.Callable[[str], date | None]) – Callable mapping variable name to date (default: fabrictools.month_start_from_ca_monthly_col()).

Returns:

Long dataframe with ids, variable, value, and month start.

Return type:

DataFrame

Example

>>> long_df = dataframe_unpivot_wide_month_suffix(
...     wide_df,
...     id_columns=["project_id"],
...     value_columns_suffix=" [CA Monthly]",
... )

fabrictools.transform.drop_rows_over_empty_percent(df: pyspark.sql.DataFrame, max_empty_percent: float, *, columns: Sequence[str] | None = None) → pyspark.sql.DataFrame

Drop rows where the fraction of empty cells (see fabrictools.empty_or_null()) exceeds max_empty_percent.

Parameters:

• df (DataFrame) – Input dataframe.

• max_empty_percent (float) – Upper bound in [0, 1]; rows with empty ratio strictly greater than this are removed.

• columns (collections.abc.Sequence[str] | None) – Columns to score; None means all columns. Names resolved like fabrictools.resolve_dataframe_column(); unknown labels are skipped, and if none remain all columns are used.

Returns:

Filtered dataframe.

Return type:

DataFrame

Raises:

ValueError – If max_empty_percent is outside [0, 1], if columns is an empty sequence, or if no columns remain to score.

Example

>>> pruned = drop_rows_over_empty_percent(
...     df, 0.5, columns=["col_a", "col_b", "col_c"]
... )

fabrictools.transform.empty_or_null(c: pyspark.sql.Column) → pyspark.sql.Column

Boolean column: true if c is null or blank after string cast and trim.

Parameters:

c (Column) – Input column expression.

Returns:

Boolean Column.

Return type:

Column

Example

>>> df.filter(empty_or_null(F.col("notes")))

fabrictools.transform.filter_column_by_values(df: pyspark.sql.DataFrame, column: str, values: Sequence[Any], *, exclude: bool = True) → pyspark.sql.DataFrame

Keep or drop rows where column is in values (no column cast).

For string-like dtypes, compares trim(column) to values. str entries in values are stripped.

Parameters:

• df (DataFrame) – Input dataframe.

• column (str) – Logical or physical column name (resolved like fabrictools.resolve_dataframe_column()). If it does not resolve, df is returned unchanged.

• values (collections.abc.Sequence) – Membership list; non-strings kept as-is.

• exclude (bool) – If True (default), drop rows in values; if False, keep only those rows.

Returns:

Filtered dataframe.

Return type:

DataFrame

Example

>>> filtered = filter_column_by_values(
...     df, "status", ["VOID", "CANCELLED"], exclude=True
... )

fabrictools.transform.merge_dataframes(main: pyspark.sql.DataFrame, join_df: pyspark.sql.DataFrame, join_columns: Sequence[str], keys: Sequence[tuple[str, str]], how: str = 'left', *, join_prefix: str | None = None, join_column_names: Sequence[str] | None = None) → pyspark.sql.DataFrame

Left-join main to join_df and project right-side columns with normalized names.

When join_column_names is omitted, each join_columns label is snake_cased. If that name is not already present on main (normalized labels), it is kept as-is; otherwise it is prefixed {prefix}_{suffix}. The prefix is snake_case from, in order: inferred join_df variable name at the call site, else first SubqueryAlias on join_df’s analyzed plan, else join. Pass join_prefix to force a value. Output names are made unique against main and among joined columns (_2, _3, …). Suffixes match fabrictools.clean_data() uniqueness rules.

Parameters:

• main (DataFrame) – Left dataframe.

• join_df (DataFrame) – Right dataframe (only join_columns projected, plus key temps).

• join_columns (collections.abc.Sequence[str]) – Right-side columns to expose; labels that do not resolve on join_df are omitted.

• join_column_names (collections.abc.Sequence[str] | None) – Optional output names for join_columns (same order, same length). If provided, names are used as-is (no prefix), still uniquified against main.

• keys (collections.abc.Sequence[tuple[str, str]]) – (main_col, join_col) pairs for the join predicate (AND); names resolved per frame. Pairs where either side does not resolve are skipped; if none resolve, main is returned unchanged.

• how (str) – Spark join type (e.g. left, inner).

• join_prefix (str | None) – Optional explicit prefix (snake_cased); overrides inference.

Returns:

Joined dataframe with temporary key columns dropped.

Return type:

DataFrame

Raises:

ValueError – If keys is empty.

Example

>>> out = merge_dataframes(
...     orders,
...     customers,
...     join_columns=["name", "segment"],
...     keys=[("customer_id", "id")],
...     join_prefix="cust",
... )

fabrictools.transform.metric_value_for_class(df: pyspark.sql.DataFrame, *, class_col_candidates: str | Sequence[str], metric_col_candidates: str | Sequence[str], class_value: Any, missing: Any = None) → Any

Return the metric cell for one class key from a pre-aggregated table.

Intended for dataframes with at most one row per class column value, for example the output of build_tcd() when grouping only on rows (no pivot columns). Uses filter + select + first without Spark casts on the metric column; the scalar is returned as Spark provides it. Callers are responsible for any coercion (e.g. int(...)).

If several rows share the same class_value, only the first row matched by Spark is used (unlike sum after groupBy on duplicates).

Parameters:

• df – Dataframe (e.g. aggregated / TCD).

• class_col_candidates – Class column name or ordered resolution candidates.

• metric_col_candidates – Metric column name or ordered resolution candidates.

• class_value – Value to match in the class column (passed to lit).

• missing – Value returned when columns do not resolve, no row matches the filter, or the metric cell is SQL NULL. Defaults to None.

Returns:

Metric value, or missing in the cases described above.

fabrictools.transform.month_from_fr_text(expr: pyspark.sql.Column | str) → pyspark.sql.Column

Extract month number (1–12) from a French text label (full or abbreviated month).

Tokens are matched as whole words on accent-stripped lowercase text (longest token first, e.g. fevrier before fev).

Parameters:

expr (Column | str) – Spark column or string literal (e.g. "OIT fev 2026", "févr").

Returns:

Integer month column (1–12), or null when no month token matches.

Return type:

Column

Example

>>> df.withColumn("mois", month_from_fr_text("periode_label"))

fabrictools.transform.month_start_from_ca_monthly_col(col_name: str) → date | None

Parse first-of-month from a column name: French mois année head, optional `` [label]`` suffix stripped.

Parameters:

col_name (str) – Wide column name (e.g. janvier_2024 [CA Monthly]).

Returns:

Parsed month start, or None if parsing fails.

Return type:

datetime.date | None

Example

>>> d0 = month_start_from_ca_monthly_col("janvier_2024 [CA Monthly]")

fabrictools.transform.norm_text(expr: pyspark.sql.Column | str) → pyspark.sql.Column

Lowercase string with control chars stripped and spaces removed (Power Query Text.Clean style).

If expr is a str, it is wrapped with F.lit.

Parameters:

expr (Column | str) – Spark column or string literal.

Returns:

Transformed column expression.

Return type:

Column

Example

>>> df.withColumn("key_norm", norm_text("Customer Name"))

fabrictools.transform.remove_columns(df: pyspark.sql.DataFrame, *names: str, columns: Sequence[str] | None = None, keep_columns: bool = False) → pyspark.sql.DataFrame

Drop columns by physical name or by the same resolution rules as fabrictools.merge_dataframes().

Pass column labels either as positional arguments or as columns= (not both). With keep_columns=True, only the resolved columns are kept and all others are dropped.

Parameters:

• df (DataFrame) – Input dataframe.

• names (str) – One or more labels (positional); duplicates resolving to the same physical column are dropped once. Labels that do not resolve to a column on df are ignored (drop mode) or skipped (keep mode).

• columns (collections.abc.Sequence[str] | None) – Optional sequence of labels; same resolution rules as names. Use when the list is already in a variable.

• keep_columns (bool) – If False (default), drop the resolved columns. If True, drop every column not in the resolved set (keep-only).

Returns:

Dataframe with columns removed per keep_columns. In drop mode, unchanged if every label is unknown.

Return type:

DataFrame

Raises:

ValueError – If no names are passed, if both positional names and columns are provided, or if keep_columns=True and no label resolves to a column on df.

Example

>>> slim = remove_columns(df, "temp_flag", "raw_json_blob")
>>> slim = remove_columns(df, columns=["temp_flag", "raw_json_blob"])
>>> keep_few = remove_columns(df, "id", "ts", keep_columns=True)

fabrictools.transform.rename_columns_month_year_block_labels(df: pyspark.sql.DataFrame, *, labels: Sequence[str] = ('Coûts prévisionnels (par mois)', 'Coûts prévisionnels cumulés', 'Avancement prévisionnel', 'CA prévisionnel cumulé', 'CA Monthly'), exclude_columns: Collection[str] = ('__spark_row_order__',)) → pyspark.sql.DataFrame

Rename contiguous French mois année column blocks using ordered labels (projection-style).

Order follows df.columns after exclude_columns. Rename targets disambiguate with __2, __3, … among new names, then _2, _3, … against the rest of the schema.

Parameters:

• df (DataFrame) – Input wide dataframe.

• labels (collections.abc.Sequence[str]) – Block markers in column order (defaults to built-in forecast / CA block set).

• exclude_columns (collections.abc.Collection[str]) – Column names ignored when scanning contiguous runs.

Returns:

Dataframe with renamed month columns.

Return type:

DataFrame

Example

>>> tagged = rename_columns_month_year_block_labels(
...     wide_projection_df, labels=("Block A", "Block B")
... )

fabrictools.transform.rename_columns_normalized(df: pyspark.sql.DataFrame) → pyspark.sql.DataFrame

Rename every column to snake_case with _2, _3, … disambiguation.

Uses the same name scheme as the rename step in fabrictools.clean_data(). Does not cast types, replace blanks, deduplicate rows, or drop rows.

Parameters:

df (DataFrame) – Input dataframe.

Returns:

Dataframe with updated column names where needed.

Return type:

DataFrame

Example

>>> renamed = rename_columns_normalized(messy_cols_df)

fabrictools.transform.rename_columns_pq_serial_to_dates(df: pyspark.sql.DataFrame, *, date_format: str = '%Y-%m-%d', prefix: str = '', include_suffix_in_name: bool = True) → pyspark.sql.DataFrame

Rename columns whose names embed a Power Query / Excel day serial (epoch PQ_EPOCH).

Non-matching columns are unchanged. Target collisions get _2, _3, … suffixes.

Parameters:

• df (DataFrame) – Input dataframe.

• date_format (str) – strftime format for the date portion of new names.

• prefix (str) – Text prepended before the formatted date.

• include_suffix_in_name (bool) – If True, append parsed numeric suffix after the serial segment.

Returns:

Dataframe with renamed columns.

Return type:

DataFrame

Example

>>> dated = rename_columns_pq_serial_to_dates(
...     pq_wide_df, date_format="%Y-%m-%d", prefix="d_"
... )

fabrictools.transform.rename_columns_pq_serial_to_mois_annee(df: pyspark.sql.DataFrame, *, prefix: str = '', include_suffix_in_name: bool = True, capitalize_month: bool = True) → pyspark.sql.DataFrame

Like rename_columns_pq_serial_to_dates() but labels use French mois année (e.g. janvier_2024).

Parameters:

• df (DataFrame) – Input dataframe.

• prefix (str) – Prepended before the month-year token.

• include_suffix_in_name (bool) – Append _{suffix} when a numeric suffix follows the serial in the source name.

• capitalize_month (bool) – If True, capitalize the month word (e.g. Janvier_2024).

Returns:

Renamed dataframe.

Return type:

DataFrame

Example

>>> labeled = rename_columns_pq_serial_to_mois_annee(
...     pq_wide_df, prefix="m_", capitalize_month=True
... )

fabrictools.transform.resolve_dataframe_column(df: pyspark.sql.DataFrame, name: str | Sequence[str]) → str | None

Resolve name to the physical column name on df.

Accepts the physical name, a fabrictools.clean_data()-style normalized label, or snake_case (same rules as fabrictools.merge_dataframes() / fabrictools.remove_columns()).

If name is a sequence of strings, each entry is tried in order and the first that resolves wins.

Parameters:

• df (DataFrame) – Dataframe whose schema is searched.

• name (str | collections.abc.Sequence[str]) – Logical, normalized, or physical column label, or ordered candidates.

Returns:

Physical column name present on df, or None if none resolve.

Return type:

str | None

Example

>>> physical = resolve_dataframe_column(df, "Customer ID")

fabrictools.transform.transform_wide_month_suffix(df: pyspark.sql.DataFrame, *, id_columns: ~collections.abc.Sequence[str], aggregation: ~typing.Literal['last_nonnull', 'pivot_sum'], value_columns_suffix: str | None = None, value_columns: ~collections.abc.Sequence[str] | None = None, exclude_columns: ~collections.abc.Collection[str] = (), variable_column: str = 'MoisCol', value_column: str = 'Valeur', month_start_column: str = 'MonthStart', month_start_from_column_name: ~collections.abc.Callable[[str], ~datetime.date | None] = <function month_start_from_ca_monthly_col>, order_column: str | None = None, output_value: str = 'Value', output_month_start: str = 'MonthStart', output_year: str = 'Year', output_month: str = 'Month', category_column: str | None = None, pivot_categories: ~collections.abc.Sequence[str] | None = None, fill_value: float = 0.0, montant_column: str = 'Montant') → pyspark.sql.DataFrame

Run dataframe_unpivot_wide_month_suffix() then last_nonnull or pivot_sum aggregation.

Parameters:

• df (DataFrame) – Wide source dataframe.

• id_columns (collections.abc.Sequence[str]) – Passed through to unpivot.

• aggregation (Literal['last_nonnull', 'pivot_sum']) – last_nonnull (needs order_column) or pivot_sum (needs category_column and pivot_categories).

• value_columns_suffix (str | None) – Passed through to unpivot.

• value_columns (collections.abc.Sequence[str] | None) – Passed through to unpivot.

• exclude_columns (collections.abc.Collection[str]) – Passed through to unpivot.

• variable_column (str) – Long-form variable column name.

• value_column (str) – Long-form value column name.

• month_start_column (str) – Long-form month start column name.

• month_start_from_column_name (collections.abc.Callable[[str], date | None]) – Parser for month start from variable name.

• order_column (str | None) – Source-wide column for last_nonnull ordering (resolved on df). If it does not resolve, the long unpivot result is returned unchanged.

• output_value (str) – Output value column for last_nonnull.

• output_month_start (str) – Output month start for last_nonnull.

• output_year (str) – Output year for both aggregations where applicable.

• output_month (str) – Output month for both aggregations where applicable.

• category_column (str | None) – Source column for pivot_sum (resolved on df). If it does not resolve, the long unpivot result is returned unchanged.

• pivot_categories (collections.abc.Sequence[str] | None) – Category list for pivot_sum.

• fill_value (float) – Pivot fill for pivot_sum.

• montant_column (str) – Internal sum column name for pivot path.

Returns:

Aggregated dataframe per selected mode, or the long unpivot only when order_column / category_column does not resolve as above.

Return type:

DataFrame

Raises:

ValueError – If aggregation is unknown or required parameters are missing.

Example

>>> summary = transform_wide_month_suffix(
...     wide_df,
...     id_columns=["project_id"],
...     aggregation="last_nonnull",
...     value_columns_suffix=" [CA Monthly]",
...     order_column="snapshot_date",
... )

fabrictools.transform.wide_value_columns(df: pyspark.sql.DataFrame, *, suffix: str, exclude: Collection[str] = ()) → list[str]

List physical columns whose names end with suffix and are not in exclude.

Parameters:

• df (DataFrame) – Wide dataframe.

• suffix (str) – Suffix substring to match (e.g. block label including leading space if stored that way).

• exclude (collections.abc.Collection[str]) – Column names to skip.

Returns:

Ordered column names from df.columns.

Return type:

list[str]

Example

>>> cols = wide_value_columns(df, suffix=" [CA Monthly]")

fabrictools.transform.with_year_month_from_fr_text(df: pyspark.sql.DataFrame, source_col: str, year_col: str | None = 'annee', month_col: str | None = 'mois', yyyymm_col: str | None = 'annee_mois') → pyspark.sql.DataFrame

Add year and/or month columns parsed from source_col (resolved like other transform helpers).

When both year_col and month_col are set, also adds a year-month key column (année × 100 + mois, e.g. February 2026 → 202602) unless yyyymm_col is None.

Parameters:

• df (DataFrame) – Input dataframe.

• source_col (str) – Source text column (physical, normalized, or snake_case label).

• year_col (str | None) – Output year column name (default "annee"). Pass None to skip.

• month_col (str | None) – Output month column name (default "mois"). Pass None to skip.

• yyyymm_col (str | None) – Output year-month key column (default "yyyymm"). Pass None to skip.

Returns:

Dataframe with one to three added columns.

Return type:

DataFrame

Raises:

ValueError – If source_col does not resolve on df, or if both output names are None.

Example

>>> out = with_year_month_from_fr_text(df, "libelle_periode")
>>> out = with_year_month_from_fr_text(df, "libelle_periode", "Annee", "Mois")
>>> out = with_year_month_from_fr_text(
...     df, "libelle_periode", year_col="Year", month_col="Month"
... )

fabrictools.transform.year_from_fr_text(expr: pyspark.sql.Column | str) → pyspark.sql.Column

Extract the first calendar year (1900–2099) from a French text label.

If expr is a str, it is wrapped with F.lit.

Parameters:

expr (Column | str) – Spark column or string literal (e.g. "OIT fev 2026").

Returns:

Integer year column, or null when no year is found.

Return type:

Column

Example

>>> df.withColumn("annee", year_from_fr_text("periode_label"))