helpers

General helper utilities for Databricks DAME project.

This module provides utility functions for common operations including: - DataFrame schema validation and alignment - CSV file output - Databricks context retrieval - Table management - Reminder submission

Requires: - pandas - pyspark (for some functions) - datetime - re - dbutils (Databricks utilities) - spark (Databricks SparkSession)

helpers._get_notebook_value(name)

Return a value from the notebook globals when available.

Parameters:

Name	Type	Description	Default
name	str	Global name to look up, for example "spark" or "dbutils".	required

Returns:

Type	Description
Any	Value bound to name in the __main__ module, falling back to this module's globals. None when the name is unbound in both. A None value in __main__ is treated as absent.

helpers.get_spark()

Return the active Spark session from the notebook or create one.

Returns:

Type	Description
Any	The notebook's spark object when running in Databricks, otherwise a session obtained from SparkSession.builder.getOrCreate().

helpers.get_dbutils()

Return dbutils from the notebook, Spark, or a local SDK fallback.

Returns:

Type	Description
Any	The notebook's dbutils when running in Databricks, otherwise a DBUtils instance built from the Spark session, otherwise the Databricks SDK WorkspaceClient().dbutils.

Raises:

Type	Description
RuntimeError	If none of the three sources is available, which normally means neither Databricks Connect nor the Databricks SDK is installed.

helpers.write_pyspark_df_to_single_csv(df, path, verbose=False, delimiter=',')

Write a PySpark DataFrame to a single CSV file.

This function coalesces the DataFrame to a single partition and writes it as CSV, then moves the resulting file to the desired path.

Parameters:

Name	Type	Description	Default
df		PySpark DataFrame to write.	required
path	str	Output path for the CSV file (without extension). If it doesn't end with '.csv', it will be added.	required
verbose	bool	Print detailed progress messages. Defaults to False.	False
delimiter	str	CSV delimiter character. Defaults to ",".	','

Raises:

Type	Description
FileNotFoundError	If no CSV file is generated in the temporary directory.

Note

• Uses coalesce(1) instead of repartition(1) for efficiency
• Creates a temporary directory that is cleaned up after successful write
• Requires dbutils to be available in the Databricks environment

Example

write_pyspark_df_to_single_csv( df, "s3://bucket/output/data.csv", verbose=True )

helpers.ensure_schema(df, reference_df)

Align a DataFrame's schema to match a reference DataFrame's schema.

Ensures the input DataFrame has the same columns, in the same order, with the same data types as the reference DataFrame.

Parameters:

Name	Type	Description	Default
df		Input DataFrame (Pandas or PySpark).	required
reference_df		Reference DataFrame defining the target schema (Pandas or PySpark).	required

Returns:

Type	Description
	DataFrame with the same structure as reference_df, either Pandas
	or PySpark depending on the input type.

Raises:

Type	Description
ValueError	If column type casting fails.
TypeError	If DataFrames are not Pandas or PySpark types.
ImportError	If PySpark not available for PySpark DataFrames.

Note

• Handles Pandas and PySpark DataFrames
• Creates missing columns with NA values
• Casts data types to match reference
• Drops extra columns not in reference
• Reorders columns to match reference

TODO: Add support for nested schemas (structs, arrays)

Example

aligned_df = ensure_schema(df, reference_df)

helpers.retrieve_current_user()

Return the username from the active Databricks notebook context.

Returns:

Type	Description
	Current Databricks username.

helpers.retrieve_current_notebook()

Return the path of the active Databricks notebook.

Returns:

Type	Description
	Current notebook workspace path.

helpers.dbutils_path_exists(path)

Return whether a DBFS/S3 path exists using Databricks utilities.

Parameters:

Name	Type	Description	Default
path	str	DBFS or S3 path to test.	required

Returns:

Type	Description
bool	True when dbutils.fs.ls lists the path.

Raises:

Type	Description
IOError	If the listing fails for a reason not recognizable as a missing path, so genuine failures such as permission errors are not reported as False.

helpers.find_latest_date_partition(base_path, partition_names=('yyyy', 'mm', 'dd'), max_date=None)

Find the latest nested date partition under a DBFS/S3 path.

The path is expected to use Hive-style partition folders, for example: s3://bucket/table/yyyy=2026/mm=07/dd=02.

Parameters:

Name	Type	Description	Default
base_path	str	Parent path containing the first partition level.	required
partition_names	Tuple[str, str, str]	Names of the year, month, and day partition columns. Defaults to ("yyyy", "mm", "dd").	('yyyy', 'mm', 'dd')
max_date	Optional[Union[date, datetime, str]]	Optional inclusive upper bound for the partition date. Accepts a date, a datetime (using its date component), or a string in YYYY-MM-DD format.	None

Returns:

Type	Description
tuple	((year, month, day), partition_path) where partition_path is the latest day-level partition path.

Raises:

Type	Description
FileNotFoundError	If no complete, valid partition exists within the optional upper bound.
ValueError	If partition_names does not contain exactly three names or max_date has an unsupported type or invalid string format.

helpers.write_to_table(df, table_name, schema, catalog='panel-management', mode='overwrite')

Write a DataFrame to a Delta table in Unity Catalog.

Sanitizes column names and writes the DataFrame as a managed Delta table in the specified catalog and schema.

Parameters:

Name	Type	Description	Default
df		Input DataFrame (Pandas or PySpark). Pandas DataFrames are automatically converted to PySpark.	required
table_name	str	Name of the table to create/overwrite.	required
schema	str	Schema (database) name where the table will be created.	required
catalog	str	Catalog name in Unity Catalog. Defaults to "panel-management".	'panel-management'
mode	str	Write mode ("overwrite" or "append"). Defaults to "overwrite". Use "append" for incremental loads.	'overwrite'

Returns:

Type	Description
	None (prints confirmation message)

Raises:

Type	Description
Exception	If table write fails.

Note

• Column names are sanitized by replacing special characters with '_'
• Requires Spark session and Unity Catalog access
• overwriteSchema option enabled to auto-update schema
• Creates schema if it doesn't exist

Example

write_to_table( df, table_name="users", schema="raw_data", catalog="my-catalog", mode="overwrite" )

helpers.hash_string(input_string, ciffre)

Create a deterministic fixed-width numeric hash.

Parameters:

Name	Type	Description	Default
input_string	str	Text to hash with SHA-256.	required
ciffre	int	Maximum number of decimal digits in the returned value.	required

Returns:

Type	Description
	The SHA-256 digest reduced modulo 10 ** ciffre.

helpers.generate_random_alias()

Generate a 50-character alphanumeric alias.

Returns:

Type	Description
	Random string drawn from ASCII letters and decimal digits.

Notes

This helper uses random.choices and is not suitable for secrets or other security-sensitive identifiers.

helpers.cleanup(path)

Remove every file in a local directory and then remove the directory.

Parameters:

Name	Type	Description	Default
path		Local directory containing only removable files.	required

Returns:

Type	Description
	None. Failures are printed and not re-raised.

Notes

Nested directories are not removed by this helper.

helpers.ensure_directory(dir)

Create a local directory and missing parents when it does not exist.

Parameters:

Name	Type	Description	Default
dir		Local directory path.	required

Returns:

Type	Description
	None.

helpers.print_warning(warning_text, logger=None, end='\n')

Print a warning and optionally write the same message to a logger.

Parameters:

Name	Type	Description	Default
warning_text		Message to display.	required
logger		Optional logger exposing a log method.	None
end		String appended after the console message.	'\n'

helpers.print_info(info_text, logger=None, end='\n')

Print an informational message and optionally write it to a logger.

Parameters:

Name	Type	Description	Default
info_text		Message to display.	required
logger		Optional logger exposing a log method.	None
end		String appended after the console message.	'\n'

helpers.blockPrint()

Redirect sys.stdout to the operating-system null device.

Returns:

Type	Description
	None. Call enablePrint to restore standard output.

helpers.enablePrint()

Restore sys.stdout to Python's original standard-output stream.

Returns:

Type	Description
	None.

helpers.suppress_display()

Temporarily replace the notebook display function with a no-op.

Yields:

Type	Description
	Control to the wrapped context while display calls are suppressed.

Notes

The previous global display function is restored even if the wrapped code raises an exception.

helpers.generate_timestamp(zone='Europe/Copenhagen')

Return the current time formatted for identifiers.

Parameters:

Name	Type	Description	Default
zone		Timezone name understood by pytz.	'Europe/Copenhagen'

Returns:

Type	Description
	Timestamp string formatted as YYYYMMDDHHMM.

Raises:

Type	Description
UnknownTimeZoneError	If zone is not recognized.

helpers.retrieve_current_user()

Return the username from the active Databricks notebook context.

Returns:

Type	Description
	Current Databricks username.

helpers.retrieve_current_notebook()

Return the path of the active Databricks notebook.

Returns:

Type	Description
	Current notebook workspace path.

helpers.s3_uri_to_url(s3_uri)

Convert an S3 URI into an AWS console URL.

Parameters:

Name	Type	Description	Default
s3_uri		URI beginning with s3://.	required

Returns:

Type	Description
	AWS S3 console URL targeting the bucket and object-key prefix in
	us-east-1.

Raises:

Type	Description
ValueError	If s3_uri does not begin with s3://.

helpers.generate_combinations(channels)

Generate every non-empty combination of peers for each item.

Parameters:

Name	Type	Description	Default
channels		Iterable of distinct channel values.	required

Returns:

Type	Description
	Dictionary mapping each channel to tuples containing all non-empty
	combinations of the other channels.

helpers.compare_schemas(df1, df2)

Print the differences between two PySpark DataFrame schemas.

Parameters:

Name	Type	Description	Default
df1	DataFrame | str	First Spark DataFrame, or an S3/DBFS path to a Parquet dataset to read.	required
df2	DataFrame | str	Second Spark DataFrame, or a path, compared against df1.	required

Returns:

Type	Description
None	Both schemas and their unified diff are printed rather than returned.

Notes

Struct fields are sorted by name before comparison, so column order differences are ignored and only genuine name, type, or nullability changes are reported.

helpers.silent(func, *args, **kwargs)

Run a function while suppressing everything it prints to stdout.

Useful for calling helpers that emit a lot of console output when only the return value matters.

Parameters:

Name	Type	Description	Default
func		Callable to invoke.	required
*args		Positional arguments forwarded to func.	()
**kwargs		Keyword arguments forwarded to func.	{}

Returns:

Type	Description
Any	Whatever func returns.

Notes

Only stdout is redirected. Output written to stderr, and log records emitted through the logging module, still appear.

helpers.sanitize_path(path, replacement='-')

Sanitize a file path by replacing forbidden characters.

Replaces characters that are not allowed in file paths with a specified replacement character.

Parameters:

Name	Type	Description	Default
path	str	The file path to sanitize.	required
replacement	str	The character to use as replacement for forbidden characters. Defaults to "-".	'-'

Returns:

Name	Type	Description
str	str	The sanitized path with forbidden characters replaced.

Note

Forbidden characters: < > : " / \ | ? *

Example

sanitize_path("file.txt") 'file-name.txt' sanitize_path("path/to:file", "_") 'path/to_file'

helpers.month_int_to_short_name(month)

Return the three-letter English short name for a month number.

Parameters:

Name	Type	Description	Default
month		Month number from 1 to 12.	required

Returns:

Type	Description
str	Short month name such as "Jan".

Raises:

Type	Description
ValueError	If month is outside the range 1 to 12.

helpers.extract_age_categories(df, age_col='age', categories=['18-24', '25-34', '35-44', '45-54', '55-64', '65+'], max_age=120)

Bucket an age column into labeled categories.

Parameters:

Name	Type	Description	Default
df		Spark DataFrame containing age_col.	required
age_col	str	Name of the age column. A trailing + is stripped and the column is cast to integer before bucketing.	'age'
categories	list	Category labels in "lower-upper" or "lower+" form. Order defines evaluation order, so overlapping ranges resolve to the first match.	['18-24', '25-34', '35-44', '45-54', '55-64', '65+']
max_age	int	Upper bound applied to an open-ended "lower+" category.	120

Returns:

Type	Description
DataFrame	Input DataFrame with an added {age_col}_cat column, filtered to rows that fall into one of categories. Ages outside every category are dropped.

Raises:

Type	Description
Exception	If df does not contain age_col.

helpers.extract_date_from_path(path, pattern='yyyy=(\\d{4})/mm=(\\d{1,2})/dd=(\\d{1,2})')

Extract a partition date from a path string.

Parameters:

Name	Type	Description	Default
path		Path containing Hive-style date partitions, for example s3://bucket/table/yyyy=2026/mm=07/dd=02.	required
pattern		Regular expression with three capture groups for year, month, and day.	'yyyy=(\\d{4})/mm=(\\d{1,2})/dd=(\\d{1,2})'

Returns:

Type	Description
tuple	(year, month, day) as integers.

Raises:

Type	Description
ValueError	If pattern does not match anywhere in path.

helpers.convert_date_list_to_string(date_list)

Convert a (year, month, day) tuple to a YYYY-MM-DD string.

Parameters:

Name	Type	Description	Default
date_list		Three-element sequence of year, month, and day. Values are zero-padded, so (2026, 7, 2) becomes "2026-07-02".	required

Returns:

Type	Description
str	Date formatted as YYYY-MM-DD.