Skip to main content
Unlisted page
This page is unlisted. Search engines will not index it, and only users having a direct link can access it.

Load Bronze

The load_data_bronze module is the entry point for extracting data from source systems and landing it in the Bronze layer. It handles multiple file formats (CSV, JSON, XML, etc.) and ensures that raw data is captured with minimal transformation while managing pre- and post-processing notebook hooks. The module includes intelligent file change detection to skip unnecessary loads and validation checks to ensure data quality.

A Load Bronze run completing — the activity panel reports 5 files loaded for table adv_products in 1 min 39 sec.

Overview​

The bronze loader performs the following workflow:

  1. Source Check (optional): For Azure Blob sources, checks blob metadata without downloading to determine if any files have changed
  2. File Pulling: Downloads files from the configured source system
  3. Freshness Validation: Verifies that source files are not stale based on configurable age thresholds
  4. Change Detection (optional): Compares current files against a snapshot to skip loading if nothing has changed
  5. Data Loading: Processes files according to their file type and loads into Bronze tables
  6. History Tracking (optional): Maintains history records and validates that file changes correlate with data changes

Key Features​

Skip-if-Unchanged (skipifsourceunchanged)​

This feature prevents unnecessary bronze loads when source files have not changed. It works in two stages:

Stage 1: Azure Blob Pre-Check (for Azure Blob sources)

  • Compares blob metadata (MD5 hash and size) without downloading files
  • If all blobs match previous snapshot, entire load is skipped
  • Significantly reduces bandwidth and processing time
  • The freshness check (Validation 1) still runs against the listed metadata before skipping, so an unchanged-but-stale source — a supplier that stopped delivering, leaving an old file in place — is surfaced rather than silently skipped

Stage 2: File Change Detection (after file pull)

  • Compares pulled files against previously saved snapshot
  • Uses MD5 hash or file size for comparison
  • Skips load if no changes detected

Enable this feature by setting skipifsourceunchanged: true in your table configuration.

File Freshness Validation (Validation 1)​

Ensures source files are recent enough for processing:

  • Checks the last_modified timestamp for each file. Files that have no last_modified timestamp are exempt and always pass the check.
  • Compares against maxfileagehours (default: 2 hours)
  • Runs both after the file pull and on the Azure Blob unchanged fast-path, so a stale source is caught even when the load would otherwise be skipped as unchanged
  • The response is governed by bronzeloadviolationaction: continue logs a warning and proceeds (or skips, if unchanged), stop raises FabricStaleFileError
  • Bypass per-run with the skip_stale_check override

The check runs on the whole batch before the Bronze table is truncated and before any file is loaded, and it is all-or-nothing:

  • Under the default stop, the first stale file in the batch raises FabricStaleFileError and aborts the entire Bronze run for that table. Nothing is written — not the stale file, and not the fresh files pulled alongside it. Because the abort happens before the truncate, the existing Bronze table is left untouched.
  • Under continue, every stale file is logged as a warning and the run proceeds normally, loading all files (the stale one included). Stale files are never skipped individually.

History Validation (Validation 2)​

When history is enabled (keephistory: true), validates data integrity:

  • Tracks MD5 hash and file size for each source file
  • Compares file changes against the number of new records added to the history table (Bronze.his)
  • Warns if files changed but no new records were added (potential data issue)
  • Logs detailed change information for debugging

File Tracker​

The file tracker maintains snapshots of source files in the Bronze layer:

  • Stores metadata (MD5, size, timestamp) in NDJSON format
  • Located at the stable Bronze path for the table
  • Used for change detection without re-downloading files
  • Automatically updated after successful loads

Function Reference​

run​

def run(tablefile: str, config_manager: ConfigManager = None,
        overrides: LoadOverrides = None) -> str | None

Runs the bronze loader process for a specified table configuration and pulls files from the source, processes them, and loads them into the bronze layer.

Workflow:

  1. Validates table configuration and layer settings
  2. Computes stable Bronze folder path for file tracker
  3. Loads previous file snapshot from tracker (if exists)
  4. Optionally performs Azure Blob pre-check for unchanged sources (runs the freshness check before skipping, so a stale-but-unchanged source is still surfaced)
  5. Executes pre-bronze notebook (if configured)
  6. Pulls files from source system
  7. Validates file freshness (Validation 1) — under the default stop, a single stale file aborts the run here, before any file is loaded
  8. Checks for file changes using tracker snapshot (if skipifsourceunchanged enabled)
  9. Truncates Bronze table and loads file data by type
  10. Executes mid-bronze notebook and reloads data (if configured)
  11. Loads history records and validates correlation (Validation 2)
  12. Saves file snapshot for next run
  13. Executes post-bronze notebook (if configured)

Arguments:

  • tablefile str - Path to the YAMLA simple way to write configurations. It's basically a list that computers can read easily. file representing a table's configuration.
  • config_manager ConfigManager - An instance of ConfigManager used for accessing the application's configuration settings. Defaults to global config if not provided.
  • overrides LoadOverrides - Optional per-run overrides that flip default validation/gating behaviour for this call only — skip notebook hooks, skip the stale-file check, or force a reload of unchanged source. Defaults to no overrides (use YAML / ConfigManager defaults).

Returns:

  • str - A message indicating the outcome, such as file count, skip reason, or error details. Returns None if table is inactive or skipped.

Raises:

  • Exception - If ConfigManager is not initialized, table config is invalid, or filetype is unsupported.

Configuration Options:

  • skipifsourceunchanged (bool) - Enable skip-if-unchanged detection
  • bronzeloadskip (bool) - Skip entire bronze load for this table
  • keephistory (bool) - Maintain history records and validation
  • prebronzenotebook (str) - Path to notebook to run before loading
  • midbronzenotebook (str) - Path to notebook to run between load and history
  • postbronzenotebook (str) - Path to notebook to run after loading
  • sourceorder (bool) - Sort files by sourceorder instead of name
  • bronzefolder (str) - Override Bronze folder from connection
  • maxfileagehours (int) - Maximum file age for freshness validation (per table, falling back to the connection)

dataframeloader​

def dataframeloader(data_frame: DataFrame, load_config: LoadConfig,
                    table_config: TableConfig,
                    config_manager: ConfigManager = None) -> str | None

Loads a DataFrame into a specified data platform table using the provided configuration and manager.

This function handles the loading operation by using detailed configurations for the DataFrame, table, and the application configuration manager. It sets up logging, ensures required parameters are initialized, and supports specific settings for different layers (e.g., bronze layer). The function handles exception logging and provides mechanisms to stop processing upon encountering errors based on configuration settings.

Arguments:

  • data_frame DataFrame - The data to be loaded into the specified table.
  • load_config LoadConfig - Contains configuration for the loading process, including destination table.
  • table_config TableConfig - Holds table-specific settings, e.g., table name identifiers and layers.
  • config_manager ConfigManager - Manages and validates application-level configurations.

Returns:

  • str - Message indicating the result of the DataFrame loading process, including the target table name and error details if applicable.

Raises:

  • Exception - If the destination table name is missing from LoadConfig.
  • Exception - If the ConfigManager is not properly initialized.

Supported File Types​

The bronze loader supports the following file formats:

  • CSV - Comma-separated values with automatic schema detection
  • JSON - JSON objects and arrays
  • XML - Extensible Markup Language files
  • Parquet - Apache Parquet columnar format
  • XLSX - Excel spreadsheets
  • Notebook - Databricks notebooks for custom processing

Error Handling​

The module provides comprehensive error handling:

  • Pre-check failures: Logs warnings and continues with regular load if pre-check fails
  • File change detection failures: Proceeds with load if change detection fails
  • Stale files: Behavior depends on bronzeloadviolationaction. Under the default stop, the first stale file raises FabricStaleFileError before any file loads, aborting the whole Bronze run for that table (no files written, including fresh ones). Under continue, old files are logged as warnings and processing continues
  • Configuration errors: Raises exceptions for missing or invalid configurations
  • Stop at error: Respects the stop_at_error config to halt on first error or continue with warnings

Example Usage​

from easyfabric.load_data_bronze import run
from easyfabric.data import ConfigManager

# Initialize config manager
config_manager = ConfigManager.initialize()

# Run bronze loader for a specific table
result = run(
    tablefile="/path/to/table_config.yaml",
    config_manager=config_manager
)

print(result)
# Output: "5 files loaded for table: my_table" or "Bronze: my_table skipped (source unchanged)"

Best Practices​

  1. Enable skip-if-unchanged for tables with infrequent source changes to save processing time and bandwidth
  2. Monitor validation warnings in logs to identify data quality issues early
  3. Configure maxfileagehours appropriately for your data refresh schedule
  4. Use keephistory for critical tables to maintain an audit trail and detect data anomalies
  5. Set sourceorder if file processing order matters for your use case
  6. Pre- and post-notebooks for custom transformations and cleanup before/after loading