Skip to main content

loaders.table_utils

spark_operation_with_retries​

def spark_operation_with_retries(operation_func,
                                 config_manager: ConfigManager = None,
                                 operation_name: str = "Spark Operation")

Executes a Spark/Delta operation with retry logic for concurrency conflicts. Retry count and delay come from the silverloadretry and silverloaddelay configuration settings (defaults: 5 retries, 10 seconds).

Arguments:

  • operation_func - A callable that performs the Spark/Delta operation.
  • config_manager ConfigManager - Configuration manager for retry settings. Defaults to the initialized global configuration.
  • operation_name str - Name of the operation for logging.

refresh_table​

def refresh_table(table_config: TableConfig,
                  config_manager: ConfigManager = None,
                  layer: str = "bronze",
                  return_table: bool = False,
                  history: bool = False) -> DataFrame | None

Refresh the metadata of a Spark table and optionally return it as a DataFrame, with retry logic on failure.

Arguments:

  • table_config TableConfig - Table configuration object.
  • config_manager ConfigManager - Configuration manager for environment/layer. Defaults to the initialized global configuration.
  • layer str - Data layer name. Defaults to "bronze".
  • return_table bool, optional - If True, return the refreshed Spark DataFrame. Defaults to False.
  • history bool, optional - If True, get historical version of table. Defaults to False.

Returns:

  • DataFrame | None - Spark DataFrame if return_table is True, else None.

truncate_bronze_table​

def truncate_bronze_table(table_config: TableConfig,
                          config_manager: ConfigManager = None)

Deletes all data from the specified bronze lakehouseA place where you store both "raw" data (like files) and "organized" data (like tables). It combines the best of a File Cabinet and a Database. table using TRUNCATE and refreshes the table cache.

Arguments:

  • table_config TableConfig - Configuration object containing the table details.
  • config_manager ConfigManager - Configuration manager to retrieve lakehouse and schema details. Defaults to the initialized global configuration.

Raises:

  • Exception - If truncating the bronze table fails.

delete_silver​

def delete_silver(table_config: TableConfig,
                  config_manager: ConfigManager = None)

Deletes all data from the specified silver table using TRUNCATE and refreshes the cache to ensure the latest state of the table is visible for subsequent operations.

Arguments:

  • table_config TableConfig - Configuration object that provides table-related information such as table names. It is used to determine the silver table name linked to the provided configuration.
  • config_manager ConfigManager - Configuration manager instance that provides access to various lakehouse configurations, including the silver lakehouse details required for this operation. Defaults to the initialized global configuration.

Raises:

  • Exception - If truncating the silver table fails.