Transform Dimension and Fact Data into the Silver Layer

Overview

After ingesting raw datasets into the Bronze Layer, the next step is to transform the data into trusted, analytics-ready datasets within the Silver Layer.

This stage processes both Dimension and Fact tables.

Dimension tables undergo cleansing, validation, and standardization to create reliable reference data, while Fact tables are incrementally processed using Databricks Structured Streaming, Auto Loader, foreachBatch(), and Delta MERGE to keep transactional data synchronized.

The Silver layer contains validated, standardized, and business-ready Delta tables that serve as the trusted source for downstream Gold layer aggregations, dashboards, and analytical workloads.

Learning Objectives

After completing this guide, you will be able to:

Dimension Tables

Read Bronze Delta tables
Remove duplicate records
Handle NULL values
Standardize text values
Apply business validation rules
Convert columns to appropriate data types
Create Silver Dimension Delta tables

Fact Tables

Read Bronze streaming Delta tables
Process incremental data using Structured Streaming
Configure checkpoint locations
Perform Delta MERGE operations
Apply business validation
Create trusted Silver Fact Delta tables

Architecture

                       Bronze Delta Tables
                                │
               ┌────────────────┴────────────────┐
               ▼                                 ▼
      Dimension Tables                    Fact Tables
               │                                 │
               ▼                                 ▼
   Silver Dimension Notebook      Structured Streaming Notebook
               │                                 │
      Remove Duplicates                Remove Duplicates
      Handle NULL Values               Standardize Data
      Standardize Values               Business Validation
      Business Validation              foreachBatch()
      Convert Data Types               Delta MERGE
               │                                 │
               └────────────────┬────────────────┘
                                ▼
                       Silver Delta Tables
                                │
            ┌────────────────────────────────────┐
            ▼                                    ▼
      Dimension Tables                      Fact Tables
      • slv_brands                          • slv_order_items
      • slv_category
      • slv_products
      • slv_customers
      • slv_calendar

Source Data

The Silver notebooks read Dimension and Fact tables stored in the Bronze schema.

Dimension Tables

bronze
│
├── brz_brands
├── brz_calendar
├── brz_category
├── brz_customers
└── brz_products

Fact Tables

bronze
│
└── brz_order_items

Transformation Process

The Silver notebooks perform the following tasks.

Dimension Tables

Read Bronze Delta tables
Remove duplicate records
Handle missing values
Correct invalid values
Standardize text formatting
Convert columns to appropriate data types
Rename columns where required
Create business-friendly attributes
Write cleansed data into the Silver schema

Fact Tables

Read Bronze streaming Delta tables
Remove duplicate transactions
Validate business rules
Standardize data formats
Configure checkpoint locations
Process streaming micro-batches
Perform Delta MERGE operations
Update Silver Delta tables incrementally

Data Quality Transformations

The Silver layer applies multiple cleansing and validation techniques before publishing trusted datasets.

Transformation	Dimension	Fact
Remove Duplicates	✅	✅
Handle NULL Values	✅	✅
Standardize Text	✅	✅
Correct Invalid Values	✅	✅
Convert Data Types	✅	✅
Business Validation	✅	✅
Delta MERGE	❌	✅
Structured Streaming	❌	✅
Checkpointing	❌	✅
foreachBatch()	❌	✅

Silver Tables Created

After successful execution, the following Delta tables are created.

Silver Table	Type	Description
slv_brands	Dimension	Cleansed product brand dimension
slv_calendar	Dimension	Standardized calendar dimension
slv_category	Dimension	Cleansed product category dimension
slv_customers	Dimension	Standardized customer dimension
slv_products	Dimension	Cleansed product dimension
slv_order_items	Fact	Cleaned and incrementally updated order item transactions

Notebook Workflow

The notebooks are organized into the following sections.

1. Import Required Libraries

Imports the required PySpark SQL modules.

from pyspark.sql import functions as F
from pyspark.sql.types import *

Brand Dimension

Bronze to Silver: Data Cleaning and Transformation

%run "./CommonUtils"

from pyspark.sql import functions as F
from pyspark.sql.types import StringType, IntegerType, DateType, TimestampType, FloatType

df_bronze = spark.table(bronze_table('brz_brands'))
df_bronze.show(10)

Trim Whitespace

This step removes leading and trailing whitespace from one or more text columns using the trim() function.

Purpose:

Standardize text values.
Eliminate unwanted spaces introduced during data entry or ingestion.
Improve data quality for joins, filtering, and reporting.

Example:

Before	After
`" Nike"`	`"Nike"`
`"Apple "`	`"Apple"`
`" Samsung "`	`"Samsung"`

Only leading and trailing spaces are removed. Spaces between words are preserved.

df_silver = df_bronze.withColumn(
        'brand_name',
        F.trim(F.col('brand_name'))
    )

df_silver.show(10)

Remove Special Characters

This step removes all special characters from one or more columns, retaining only alphanumeric characters (A-Z, a-z, 0-9). This standardizes text values and improves data quality for reporting, joins, and downstream processing.

Example:

Before	After
SONY#123	SONY123
NIKE@2025	NIKE2025
HP-LAPTOP	HPLAPTOP

The transformation uses the regexp_replace() function with the regular expression [^A-Za-z0-9] to remove any character that is not a letter or a number.

df_silver =  df_silver.withColumn(
        'brand_code',
        F.regexp_replace(F.col('brand_code'), r'[^A-Za-z0-9]', '')
    )

df_silver.show(10)

Validate Distinct Category Codes

This step displays the unique values present in the category_code column after data cleansing and standardization.

Purpose:

Verify that category codes have been standardized correctly.
Identify any unexpected or invalid category values.
Validate the quality of transformed data before writing it to the Silver layer.

Using distinct() helps ensure that only the expected category codes are present in the dataset.

show_distinct_values(df_silver, "category_code")

Standardize Column Values

This step replaces predefined values in a specified column using a mapping dictionary.

Purpose:

Standardize inconsistent values.
Apply business-approved codes or abbreviations.
Improve data consistency for reporting and downstream processing.

The transformation uses a reusable utility function that accepts:

The input DataFrame
The target column name
A dictionary containing the old and new values to replace

This approach promotes code reuse and simplifies maintenance across multiple ETL notebooks.

anomalies = {
    "GROCERY": "GRCY",
    "TOYS": "TOY",
    "BOOKS": "BKS"
}

df_silver = df_silver.replace(
        to_replace=anomalies,
        subset=['category_code']
    )

display(df_silver)

show_distinct_values(df_silver, "category_code")

Persist Data to Delta Table

This step writes the transformed DataFrame to a Delta table in the specified Unity Catalog schema.

Purpose:

Persist cleansed and transformed data.
Store data in Delta format for ACID transactions and schema enforcement.
Support configurable write modes such as overwrite and append.
Reuse the same utility function across Bronze, Silver, and Gold layers.

After the write operation, the table can optionally be read back to validate that the data has been successfully persisted.

Category Dimension
Bronze to Silver: Data Cleaning and Transformation

df_bronze = spark.table(bronze_table('brz_category'))
df_bronze.show(10, truncate=False)

Identify Duplicate Records
This step identifies duplicate records based on one or more key columns.

Purpose:

Detect duplicate business keys.
Validate data quality before applying transformations.
Support data cleansing by identifying records that require deduplication.
The output includes the duplicate values and the number of occurrences for each duplicate key.

df_dublicates = find_duplicates(df_bronze, 'category_code')
df_dublicates.show()

Remove Duplicate Records

This step removes duplicate records from the dataset based on one or more business key columns.

Purpose:

Ensure each business key is unique.
Improve data quality before loading into the Silver layer.
Prevent duplicate records from affecting downstream reporting and analytics.
The dropDuplicates() function retains the first occurrence of each unique key and removes any subsequent duplicates.

df_silver = df_bronze.dropDuplicates(['category_code'])
display(df_silver)

Convert Text to Uppercase

This step converts the values of text column to uppercase to ensure consistent formatting across the dataset.

Purpose:

Standardize text values.
Eliminate case inconsistencies.
Improve data quality for joins, filtering, and reporting.
Ensure consistent business key values.
Example:

Before After
grocery GROCERY
Toys TOYS
books BOOKS

df_silver = df_silver.withColumn('category_code', F.upper(F.col('category_code')))
display(df_silver)

Persist Data to the Silver Layer

This step writes the cleansed and transformed category data to the Silver layer in Delta format.

Purpose:

Store validated and standardized data.
Support ACID transactions using Delta Lake.
Enable schema evolution with mergeSchema.
Make the data available for downstream Gold layer transformations and analytics.

write_delta_table(
df_silver,
CATALOG,
"silver",
"slv_category"
)
display(read_delta_table(spark, CATALOG, "silver", "slv_category"))

Products Dimension
Bronze to Silver: Data Cleaning and Transformation

# Read the raw data from the bronze table (ecommerce.bronze.brz_calendar)
df_bronze = spark.table(bronze_table('brz_products'))

display(df_bronze.limit(5))

# Get row and column count
row_count, column_count = df_bronze.count(), len(df_bronze.columns)

# Print the results
print(f"Row count: {row_count}")
print(f"Column count: {column_count}")

Displays the first few values of a specified column (weight_grams).

# Check weight_grams column
df_bronze.select("weight_grams").show(5, truncate=False)

Clean and Convert weight_grams
This step cleans the weight_grams column by removing all non-numeric characters and converting the values to an integer data type.

Purpose:

Remove unwanted characters (such as “g”, spaces, or other text) from the weight_grams column.
Standardize the column into a numeric format for consistency.
Convert the cleaned values to an integer data type for accurate calculations, filtering, and aggregations in the Silver layer.
Improve data quality before downstream processing and analytics.

df_silver = df_bronze.withColumn(
    "weight_grams",
    F.regexp_replace(F.col("weight_grams"), r"[^0-9]", "").cast(IntegerType())
)

df_silver.select("weight_grams").show(5, truncate=False)

df_silver.select(“length_cm”).show(3)

Clean and Convert `length_cm`

This step standardizes the length_cm column by replacing comma (",") decimal separators with a period (".") and converting the values to a float data type.

Purpose:

Replace comma decimal separators with periods to ensure a consistent numeric format.
Convert the length_cm column to a float data type for accurate decimal representation.
Enable numerical calculations, filtering, and aggregations on the column.
Improve data quality and ensure compatibility with downstream processing and analytics in the Silver layer.

df_silver = df_silver.withColumn(
    "length_cm",
    F.regexp_replace(F.col("length_cm"), ",", ".").cast(FloatType())
)

df_silver.select("length_cm").show(3)

category_code and brand_code are in lower case. we need to make it all upper case

df_silver.select("category_code", "brand_code").show(2)

Standardize `category_code` and `brand_code`

This step standardizes the category_code and brand_code columns by converting all values to uppercase.

Purpose:

Ensure consistent formatting across categorical data.
Eliminate differences caused by mixed letter casing (for example, "abc" and "ABC").
Improve the accuracy of filtering, grouping, joins, and aggregations.
Enhance data quality and maintain consistency for downstream processing and analytics in the Silver layer.

df_silver = df_silver.withColumn(
    "category_code",
    F.upper(F.col("category_code"))
).withColumn(
    "brand_code",
    F.upper(F.col("brand_code"))
)

df_silver.show(2)

Spelling mistakes in material column

show_distinct_values(df_silver, "material")

Standardize `material` Values

This step corrects spelling mistakes in the material column by replacing inconsistent or misspelled values with standardized values.

Purpose:

Correct common spelling errors in the material column.
Standardize material names to ensure data consistency.
Prevent duplicate categories caused by inconsistent text values.
Improve the accuracy of filtering, grouping, reporting, and analytics in the Silver layer.

# Method 1: Using when()

# df_silver = df_silver.withColumn(

#     "material",

#     F.when(F.col("material") == "Coton", "Cotton")

#      .when(F.col("material") == "Alumium", "Aluminum")

#      .when(F.col("material") == "Ruber", "Rubber")

#      .otherwise(F.col("material"))

# )

#  Method 2: Using replace()

std_column_val = {

    "Coton": "Cotton",

    "Alumium": "Aluminum",

    "Ruber": "Rubber"

}

df_silver = df_silver.replace(

    to_replace=std_column_val,

    subset=["material"]

)

display(df_silver.limit(10))

Negative values in rating_count

df_silver.filter(F.col('rating_count')<0).select("rating_count").show(3)

Standardize `rating_count`

This step standardizes the rating_count column by converting negative values to their absolute values and replacing null values with 0.

Purpose:

Convert invalid negative rating counts to positive values using the absolute value function.
Replace missing (null) values with a default value of 0.
Ensure the rating_count column contains valid, non-negative numeric values.
Improve data quality and enable accurate calculations, reporting, and analytics in the Silver layer.

df_silver = df_silver.withColumn(

    "rating_count",

    F.when(F.col("rating_count").isNotNull(), F.abs(F.col("rating_count")))

    .otherwise(F.lit(0))

)

df_silver.select("rating_count").show(3)

Check final cleaned data

df_silver.select(

    "weight_grams",

    "length_cm",

    "category_code",

    "brand_code",

    "material",

    "rating_count"

).show(10, truncate=False)

Persist Data to the Silver Delta Table

This step writes the transformed Silver DataFrame to a Delta table and reads the table to verify that the data has been successfully stored.

Purpose:

Persist the cleansed and standardized data in the Silver layer.
Store the data in Delta format for reliable, scalable, and ACID-compliant storage.
Make the Silver table available for downstream transformations and analytics.
Validate the data by reading and displaying the contents of the Delta table after the write operation.

write_delta_table(
    df_silver,
    CATALOG,
    SILVER_SCHEMA,
    "slv_products"
)

display(read_delta_table(spark, CATALOG, SILVER_SCHEMA, "slv_products").show(10))

Customers Dimension

Bronze to Silver: Data Cleaning and Transformation

# Read the raw data from the bronze table (ecommerce.bronze.brz_customers)
df_bronze = spark.table(bronze_table('brz_customers'))
display(df_bronze.limit(5))

# Get row and column count
row_count, column_count = df_bronze.count(), len(df_bronze.columns)

# Print the results
print(f"Row count: {row_count}")
print(f"Column count: {column_count}")

Handle NULL Values in customer_id

This step removes records where the customer_id column contains NULL values using the dropna() function.

Purpose:

Remove records with missing customer_id values.
Ensure every record has a valid customer identifier.
Improve data quality and maintain data integrity.
Prevent incomplete records from affecting downstream processing, reporting, and analytics in the Silver layer.

null_count = df_bronze.filter(F.col("customer_id").isNull()).count()
null_count

# There are 300 null values in customer_id column. Display some of those
df_bronze.filter(F.col("customer_id").isNull()).show(3)

df_silver = df_bronze.dropna(subset=["customer_id"])
df_silver.count()

Handle NULL Values in phone

This step replaces NULL values in the phone column with the default value “Not Available” using the fillna() function.

Purpose:

Replace missing phone numbers with a meaningful default value.
Ensure the phone column does not contain NULL values.
Improve data completeness and consistency.
Prevent issues caused by missing values in downstream processing, reporting, and analytics in the Silver layer.

null_count = df_silver.filter(F.col("phone").isNull()).count()
print(f"Number of nulls in phone: {null_count}")

df_silver.filter(F.col("phone").isNull()).show(3)

df_silver = df_silver.fillna("Not Available", subset=["phone"])
df_silver.filter(F.col("phone").isNull()).show()

Persist Data to the Silver Delta Table

This step writes the cleansed and transformed customer data to the slv_customers Delta table in the Silver layer. The mergeSchema option is used to automatically merge any schema changes with the existing table. After the write operation, the table is read and displayed to verify that the data has been successfully stored.

Purpose:

Persist the cleansed and standardized customer data in the Silver layer.
Automatically merge schema changes with the existing Delta table.
Store the data in Delta format for reliable, scalable, and ACID-compliant storage.
Validate the write operation by reading and displaying the contents of the Silver Delta table.

write_delta_table(

    df_silver,

    CATALOG,

    SILVER_SCHEMA,

    'slv_customers',

    overwriteSchema="mergeSchema",

)

display(read_delta_table(spark, CATALOG, SILVER_SCHEMA, 'slv_customers'))

Calendar/Date Dimension

Bronze to Silver: Data Cleaning and Transformation

# Read the raw data from the bronze table (ecommerce.bronze.brz_calendar)

df_bronze = spark.table(bronze_table("brz_calendar"))

# Get row and column count

row_count, column_count = df_bronze.count(), len(df_bronze.columns)

# Print the results

print(f"Row count: {row_count}")

print(f"Column count: {column_count}")

df_bronze.show(3)

Remove Duplicate Records
This step identifies and removes duplicate records from the dataset based on the date column using the dropDuplicates() function. Before removing duplicates, the duplicate records are identified and counted to validate the data quality issue.

Purpose:

Identify duplicate records based on the date column.
Remove duplicate records while retaining the first occurrence of each unique date.
Ensure data uniqueness and improve data quality.
Prevent duplicate records from affecting downstream processing, reporting, and analytics in the Silver layer.

dublicate = find_duplicates(df_bronze, "date")

print(f"Duplicate count: {dublicate.count()}")

dublicate.show()

df_silver = df_bronze.dropDuplicates(["date"])

row_count = df_silver.count()

print(f"Total number of rows in silver table: {row_count}")

Standardize day_name Casing
This step standardizes the day_name column by converting each value to Proper Case using the initcap() function. This ensures that the first letter of each day name is uppercase and the remaining letters are lowercase.

Purpose:

Normalize the casing of day names for consistency.
Eliminate inconsistencies caused by mixed or incorrect letter casing (for example, “monday”, “MONDAY”, or “MoNdAy”).
Improve data quality and readability.
Ensure consistent values for filtering, grouping, reporting, and analytics in the Silver layer.

df_silver = df_silver.withColumn("day_name", F.initcap(F.col("day_name")))
df_silver.show()

Standardize week_of_year
This step standardizes the week_of_year column by converting any negative values to their absolute values using the abs() function. This ensures that all week numbers are valid and non-negative.

Purpose:

Convert negative week numbers to positive values.
Ensure the week_of_year column contains valid, non-negative values.
Improve data quality and consistency.
Enable accurate filtering, reporting, and time-based analytics in the Silver layer.

df_silver = df_silver.withColumn("week_of_year", F.abs(F.col("week_of_year")))
df_silver.show()

Format quarter and week_of_year
This step formats the quarter and week_of_year columns into more descriptive, human-readable values by concatenating the quarter or week number with the corresponding year.

Purpose:

Create standardized labels for quarter and week values.
Improve the readability of time-based dimensions.
Provide a consistent format for reporting and analytics.
Enhance the usability of date-related attributes in the Silver layer.

df_silver = df_silver.withColumn(
"quarter",
F.concat(F.lit("Q"), F.col("quarter"), F.lit("-"), F.col("year"))
)

df_silver = df_silver.withColumn(
"week_of_year",
F.concat(F.lit("Week"), F.col("week_of_year"), F.lit("-"), F.col("year"))
)

df_silver.show()

Rename week_of_year Column
This step renames the week_of_year column to week to provide a simpler and more meaningful column name while preserving the existing data.

df_silver = df_silver.withColumnRenamed("week_of_year", "week")

df_silver.show()

write_delta_table(

    df_silver,

    CATALOG,

    SILVER_SCHEMA,

    "slv_calendar",

    overwriteSchema="mergeSchema"

)

display(read_delta_table(spark, CATALOG, SILVER_SCHEMA, "slv_calendar"))

spark-databricks-ecommerce-pipeline

Curriculum

Transform Dimension and Fact Data into the Silver Layer

Overview

Learning Objectives

Dimension Tables

Fact Tables

Architecture

Source Data

Dimension Tables

Fact Tables

Transformation Process

Dimension Tables

Fact Tables

Data Quality Transformations

Silver Tables Created

Notebook Workflow

1. Import Required Libraries

Brand Dimension

Bronze to Silver: Data Cleaning and Transformation

Trim Whitespace

Remove Special Characters

Validate Distinct Category Codes

Standardize Column Values

Persist Data to Delta Table

Remove Duplicate Records

Purpose:

Convert Text to Uppercase

Persist Data to the Silver Layer

Clean and Convert length_cm

Standardize category_code and brand_code

Standardize material Values

Standardize rating_count

Check final cleaned data

Persist Data to the Silver Delta Table

Customers Dimension

Bronze to Silver: Data Cleaning and Transformation

Handle NULL Values in customer_id

Handle NULL Values in phone

Persist Data to the Silver Delta Table

Calendar/Date Dimension

Bronze to Silver: Data Cleaning and Transformation

Leave a Reply Cancel reply

Modal title

Clean and Convert `length_cm`

Standardize `category_code` and `brand_code`

Standardize `material` Values

Standardize `rating_count`