Building an End-to-End ETL Pipeline Project

Introduction

Welcome to Chapter 12! So far, we’ve explored the foundational concepts of Databricks, delved into PySpark, understood the magic of Delta Lake, and even optimized some queries. Now, it’s time to bring all those pieces together and build something truly practical: an End-to-End ETL Pipeline Project.

In this chapter, you’ll learn how to design, implement, and manage a complete Extract, Transform, Load (ETL) pipeline using Databricks. We’ll simulate a real-world scenario where data flows from raw sources, gets cleaned and enriched, and is finally prepared for analysis. This hands-on project will solidify your understanding of data engineering principles and demonstrate Databricks’ power as a unified platform for data processing. Get ready to put your skills to the test and build something awesome!

Before we dive in, make sure you’re comfortable with:

• Basic Databricks workspace navigation and notebook usage.
• PySpark DataFrames and common transformations.
• The core concepts of Delta Lake (ACID properties, time travel, upserts).
• Basic SQL queries within Databricks.

If any of these feel a bit fuzzy, a quick recap of previous chapters might be helpful. Ready? Let’s build some data magic!

Core Concepts: Understanding ETL and the Lakehouse Architecture

Before we jump into coding, let’s ensure we’re all on the same page about what an ETL pipeline is and why Databricks is an exceptional choice for building them.

What is ETL? The Data Journey Explained

ETL stands for Extract, Transform, Load. It’s a fundamental process in data warehousing and data engineering, describing the steps involved in collecting data from various sources, cleaning and reshaping it, and then delivering it to a destination for analysis or operational use.

1. Extract (E): This is where you pull data from its original sources. These sources can be incredibly diverse: relational databases, flat files (CSV, JSON, Parquet), APIs, streaming data feeds, and more. The goal here is simply to get the raw data out.
2. Transform (T): This is often the most complex and critical step. Raw data is rarely in a usable format. Transformation involves a series of operations to clean, standardize, filter, aggregate, and enrich the data. Think about removing duplicates, converting data types, joining disparate datasets, or calculating new metrics. The goal is to make the data consistent, accurate, and ready for its intended purpose.
3. Load (L): Finally, the transformed data is moved into its target destination. This could be a data warehouse, a data lake, an analytical database, or even another application. The loading process might involve appending new data, updating existing records, or completely overwriting tables.

Why is ETL important? Imagine trying to analyze sales data if half your records have “USA” and the other half “United States” for the same country, or if product prices are sometimes strings and sometimes numbers. ETL ensures data quality, consistency, and usability, making it reliable for business intelligence, machine learning, and reporting.

Why Databricks for ETL? The Lakehouse Advantage

Databricks, powered by Apache Spark and built upon the Delta Lake format, offers a unique and powerful platform for ETL, often referred to as a “Lakehouse” architecture.

• Unified Platform: Databricks brings together data engineering, data science, and machine learning on a single platform. This means you can use Python, SQL, Scala, or R within the same environment, leveraging the same underlying data.
• Apache Spark’s Power: At its core, Databricks uses Apache Spark, a lightning-fast engine for large-scale data processing. Spark’s distributed computing capabilities are perfect for handling the “large data” aspect of our ETL.
• Delta Lake’s Reliability: Delta Lake, an open-source storage layer, adds ACID (Atomicity, Consistency, Isolation, Durability) transactions to data lakes. This means your data operations are reliable, even in distributed environments. It also provides schema enforcement, schema evolution, and time travel, which are game-changers for robust ETL pipelines.
• Scalability & Performance: Databricks clusters can automatically scale up and down based on workload demands, ensuring efficient resource utilization and optimal performance for even the most demanding ETL jobs. Modern features like Databricks Serverless Compute further simplify resource management, allowing you to focus purely on your data logic.
• Simplified Orchestration: Databricks Workflows (formerly Jobs) allow you to schedule and monitor your ETL pipelines, making it easy to automate complex data flows.

The Medallion Architecture (Bronze, Silver, Gold Layers)

A common and highly recommended architectural pattern for ETL on Databricks (and in a Lakehouse generally) is the Medallion Architecture, which organizes data into three distinct layers: Bronze, Silver, and Gold. Think of it as refining raw ore into valuable jewelry!

1. Bronze Layer (Raw Data):

   • Purpose: Ingests raw data from source systems “as-is” with minimal (if any) transformations. This layer acts as a landing zone and a faithful, immutable copy of the source data.
   • Characteristics: Data is typically stored in Delta Lake format to leverage its features like schema evolution and time travel, allowing you to reprocess data from any point in time if transformations in later layers need adjustment.
   • Analogy: This is like the raw ore, straight from the mine. You haven’t cleaned it or shaped it yet, but you’ve got it safely stored.

2. Silver Layer (Cleaned & Conformed Data):

   • Purpose: This layer applies initial transformations, cleaning, filtering, and standardizing the raw data from the Bronze layer. It often involves joining data from multiple Bronze tables.
   • Characteristics: Data here is structured, validated, and ready for business-specific transformations. Common operations include handling nulls, correcting data types, filtering invalid records, and enriching data with basic lookups.
   • Analogy: Now you’ve cleaned the ore, removed the dirt, and perhaps separated different types of metals. It’s much more refined but not yet a finished product.

3. Gold Layer (Curated & Business-Ready Data):

   • Purpose: This layer contains highly refined, aggregated, and business-specific data, optimized for specific analytical workloads, dashboards, and machine learning models.
   • Characteristics: Data is typically denormalized, aggregated, and designed for fast querying. Examples include sales summaries, customer churn metrics, or daily operational reports. This is the layer directly consumed by business users, analysts, and data scientists.
   • Analogy: This is the finished jewelry – polished, shaped, and ready to be worn or displayed. It’s easy to understand and provides immediate value.

This layered approach promotes data quality, reusability, and maintainability, making our ETL pipelines robust and easier to manage.

Step-by-Step Implementation: Building Our First ETL Pipeline

Let’s get our hands dirty and build a simple ETL pipeline. Our scenario will involve processing simulated customer order data. We’ll extract raw order details, transform them to calculate order totals and clean customer names, and then load them into a curated table for analysis.

We’ll use a Databricks Notebook for this project.

Step 1: Setting Up Your Databricks Environment

First, log into your Databricks workspace.

1. Create a New Notebook:

   • In your workspace, click “New” > “Notebook”.
   • Give it a descriptive name, like ETL_Order_Pipeline.
   • Ensure the default language is Python.
   • Attach it to a running cluster. For this exercise, any standard cluster with a recent, stable Databricks Runtime (e.g., a 16.x or 17.x LTS release, which would be widely available and stable by December 2025) will work perfectly.

2. Define Our Data Storage Path: We’ll use a path in Databricks File System (DBFS) to store our data. In a real-world scenario, you’d likely use Unity Catalog external locations pointing to cloud storage (like S3, ADLS Gen2, or GCS), but for simplicity, we’ll use a path relative to the user’s home directory.

   In your new notebook, add the following code to the first cell:

   # Define our base path for storing data
   # We'll use a path relative to the user's home directory in DBFS
   # This ensures a unique path for each user and avoids conflicts
   username = spark.sql("SELECT current_user()").collect()[0][0]
   base_path = f"/Users/{username}/etl_project_data"
   
   print(f"Base path for ETL project data: {base_path}")
   
   # Create the directory if it doesn't exist (idempotent)
   dbutils.fs.mkdirs(base_path)
   

   Explanation:

   • spark.sql("SELECT current_user()"): This command executes a SQL query to get the current Databricks user’s email address.
   • collect()[0][0]: Extracts the actual string username from the result of the SQL query.
   • base_path = f"/Users/{username}/etl_project_data": We construct a unique path for your data within DBFS. This is a good practice for personal projects to avoid polluting shared directories.
   • print(f"..."): Just a friendly print statement to show you where your data will live.
   • dbutils.fs.mkdirs(base_path): This dbutils command ensures that the directory specified by base_path exists. If it doesn’t, it creates it. If it does, it does nothing, making it “idempotent” (safe to run multiple times).

   Run this cell by pressing Shift + Enter. You should see the base path printed.

Step 2: Data Extraction (Bronze Layer)

Our first step is to extract raw data. We’ll simulate receiving new order data as a CSV file.

1. Simulate Raw Data: Let’s create a simple CSV file with some dummy order data directly within our notebook and save it to our base_path.

   Add this to a new cell:

   # Define our raw data
   raw_orders_data = """
   order_id,customer_name,product_name,quantity,price,order_date,status
   1001,Alice Smith,Laptop,1,1200.00,2025-11-01,completed
   1002,Bob Johnson,Mouse,2,25.50,2025-11-01,pending
   1003,Alice Smith,Keyboard,1,75.00,2025-11-02,completed
   1004,Charlie Brown,Monitor,1,300.00,2025-11-02,shipped
   1005,Bob Johnson,Webcam,1,50.00,2025-11-03,cancelled
   1006,Alice Smith,Headphones,1,150.00,2025-11-03,completed
   1007,David Lee,,1,99.99,2025-11-04,pending
   """
   
   # Define the file path for our raw data
   raw_csv_path = f"{base_path}/raw_orders/orders_20251101.csv"
   
   # Write the data to a file in DBFS
   dbutils.fs.put(raw_csv_path, raw_orders_data.strip(), True)
   
   print(f"Raw data saved to: {raw_csv_path}")
   

   Explanation:

   • raw_orders_data: A multi-line string representing our CSV content. Notice the empty customer_name for order 1007. This will be useful for transformation.
   • raw_csv_path: We define the full path where this raw CSV will be saved. We’re creating a raw_orders sub-directory within our base_path.
   • dbutils.fs.put(raw_csv_path, raw_orders_data.strip(), True): This dbutils command writes the string content to the specified file path.
     • raw_csv_path: The destination file.
     • raw_orders_data.strip(): The content to write, .strip() removes leading/trailing whitespace.
     • True: This argument means “overwrite if the file already exists,” which is useful for re-running the notebook.

   Run this cell.

2. Read Raw Data into a DataFrame: Now, let’s read this CSV file into a Spark DataFrame.

   # Read the raw CSV data into a Spark DataFrame
   raw_orders_df = spark.read \
       .format("csv") \
       .option("header", "true") \
       .option("inferSchema", "true") \
       .load(raw_csv_path)
   
   # Display the raw DataFrame and its schema
   print("--- Raw Orders DataFrame ---")
   raw_orders_df.printSchema()
   raw_orders_df.display()
   

   Explanation:

   • spark.read.format("csv"): We tell Spark to read data in CSV format.
   • .option("header", "true"): This crucial option tells Spark that the first line of the CSV file contains column headers, not data.
   • .option("inferSchema", "true"): This option instructs Spark to automatically detect the data types of each column (e.g., quantity as integer, price as double, order_date as date). While convenient for quick exploration, in production, you’d often define a fixed schema to ensure consistency and prevent performance overhead.
   • .load(raw_csv_path): Specifies the path to the CSV file to load.
   • raw_orders_df.printSchema(): Shows the inferred schema (column names and data types).
   • raw_orders_df.display(): A Databricks-specific command to show the DataFrame content in a nicely formatted table.

   Run this cell and observe the schema and data. Notice how inferSchema correctly identified price as double and order_date as date.

3. Load into Bronze Delta Table: Finally, we’ll save this raw DataFrame into our Bronze layer as a Delta table. This makes it a robust, versioned, and queryable source.

   # Define the path for our Bronze layer Delta table
   bronze_orders_path = f"{base_path}/bronze/raw_orders"
   bronze_orders_table_name = "bronze_raw_orders"
   
   # Write the raw DataFrame to the Bronze layer as a Delta table
   raw_orders_df.write \
       .format("delta") \
       .mode("append") \
       .option("mergeSchema", "true") \
       .save(bronze_orders_path)
   
   # Create a table entry in the Metastore for easy querying
   spark.sql(f"CREATE TABLE IF NOT EXISTS {bronze_orders_table_name} USING DELTA LOCATION '{bronze_orders_path}'")
   
   print(f"Raw data loaded into Bronze Delta table at: {bronze_orders_path}")
   print(f"Table '{bronze_orders_table_name}' created/updated.")
   
   # Verify by querying the Bronze table
   print("\n--- Data in Bronze Layer ---")
   spark.sql(f"SELECT * FROM {bronze_orders_table_name}").display()
   

   Explanation:

   • bronze_orders_path: Defines the file system location for our Bronze Delta table.
   • bronze_orders_table_name: The name we’ll use to reference this table via SQL (e.g., SELECT * FROM bronze_raw_orders).
   • raw_orders_df.write.format("delta"): Specifies that we want to write the DataFrame in Delta Lake format.
   • .mode("append"): This is crucial. It tells Delta Lake to add new records to the table. Other common modes include "overwrite" (replaces the entire table) and "ignore" (does nothing if the table exists).
   • .option("mergeSchema", "true"): An important Delta Lake feature. If the schema of the incoming DataFrame differs slightly from the existing table (e.g., new columns are added), Delta Lake will attempt to merge them without failing the write operation. This is powerful for handling schema evolution in raw data.
   • .save(bronze_orders_path): Saves the DataFrame as a Delta table at the specified path.
   • spark.sql(f"CREATE TABLE IF NOT EXISTS ..."): This registers the Delta table with the Databricks Metastore (often backed by Unity Catalog or a Hive Metastore). Once registered, you can query it using standard SQL by its name, making it very convenient. IF NOT EXISTS ensures it only creates it once.

   Run this cell. You’ve successfully extracted data and landed it in your Bronze layer!

Step 3: Data Transformation (Silver Layer)

Now that our raw data is safely in the Bronze layer, let’s clean and enrich it for the Silver layer. We’ll perform a few transformations:

1. Calculate a total_amount for each order (quantity * price).
2. Standardize customer_name by replacing empty strings or nulls with “Anonymous”.
3. Add a processing_timestamp to track when the record was processed.

Add this to a new cell:

from pyspark.sql.functions import col, when, lit, current_timestamp

# Read data from the Bronze layer
bronze_df = spark.read.format("delta").load(bronze_orders_path)

# Perform transformations for the Silver layer
silver_df = bronze_df.select(
    col("order_id"),
    # Calculate total_amount: quantity * price
    (col("quantity") * col("price")).alias("total_amount"),
    # Standardize customer_name: replace empty strings or nulls with 'Anonymous'
    when(col("customer_name").isNull() | (col("customer_name") == ""), lit("Anonymous"))
        .otherwise(col("customer_name"))
        .alias("customer_name_cleaned"),
    col("product_name"),
    col("quantity"),
    col("price"),
    col("order_date"),
    col("status"),
    # Add a processing timestamp
    current_timestamp().alias("processing_timestamp")
)

# Display the transformed DataFrame and its schema
print("--- Silver Orders DataFrame ---")
silver_df.printSchema()
silver_df.display()

Explanation:

• from pyspark.sql.functions import ...: We import necessary PySpark SQL functions.
• bronze_df = spark.read.format("delta").load(bronze_orders_path): We read directly from our Bronze Delta table. This is the beauty of the layered architecture – each layer builds upon the previous one.
• silver_df = bronze_df.select(...): We use the select transformation to choose columns and apply new expressions.
  • (col("quantity") * col("price")).alias("total_amount"): Calculates the product of quantity and price and names the new column total_amount.
  • when(col("customer_name").isNull() | (col("customer_name") == ""), lit("Anonymous")).otherwise(col("customer_name")).alias("customer_name_cleaned"): This is a conditional expression.
    • col("customer_name").isNull(): Checks if customer_name is null.
    • (col("customer_name") == ""): Checks if customer_name is an empty string.
    • |: This is the logical OR operator.
    • lit("Anonymous"): If either condition is true, set the value to the literal string “Anonymous”.
    • .otherwise(col("customer_name")): Otherwise, keep the original customer_name.
    • .alias("customer_name_cleaned"): Names this new, cleaned column.
  • current_timestamp().alias("processing_timestamp"): Adds a column with the current timestamp, marking when this record was processed in the Silver layer.

Run this cell. Observe the total_amount column and how the customer_name for order_id 1007 is now “Anonymous”.

Step 4: Data Loading (Silver Delta Table)

Now, let’s load our cleaned and transformed data into the Silver layer as a Delta table.

Add this to a new cell:

# Define the path for our Silver layer Delta table
silver_orders_path = f"{base_path}/silver/cleaned_orders"
silver_orders_table_name = "silver_cleaned_orders"

# Write the transformed DataFrame to the Silver layer as a Delta table
silver_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .save(silver_orders_path)

# Create a table entry in the Metastore for easy querying
spark.sql(f"CREATE TABLE IF NOT EXISTS {silver_orders_table_name} USING DELTA LOCATION '{silver_orders_path}'")

print(f"Cleaned data loaded into Silver Delta table at: {silver_orders_path}")
print(f"Table '{silver_orders_table_name}' created/updated.")

# Verify by querying the Silver table
print("\n--- Data in Silver Layer ---")
spark.sql(f"SELECT * FROM {silver_orders_table_name}").display()

Explanation:

• silver_orders_path and silver_orders_table_name: Similar to Bronze, these define the location and name for our Silver table.
• .mode("overwrite"): Here, we use "overwrite". This is common for Silver layers, especially if the transformations are idempotent (meaning running them again with the same source data produces the exact same result). It effectively replaces the entire Silver table with the latest cleaned data derived from Bronze. Be cautious with overwrite in production; append or merge (for upserts) are often preferred depending on the specific use case and data volume.
• The rest is similar to the Bronze write operation.

Run this cell. You now have a clean, conformed dataset in your Silver layer!

Step 5: Data Loading (Gold Layer)

Finally, let’s create our Gold layer, which will contain aggregated, business-ready data. For this example, we’ll create a daily sales summary.

Add this to a new cell:

# Read data from the Silver layer
silver_df_for_gold = spark.read.format("delta").load(silver_orders_path)

# Aggregate data for the Gold layer (daily sales summary)
gold_daily_sales_df = silver_df_for_gold.groupBy("order_date") \
    .agg(
        sum(col("total_amount")).alias("daily_revenue"),
        count(col("order_id")).alias("total_orders"),
        countDistinct(col("customer_name_cleaned")).alias("unique_customers")
    ) \
    .orderBy("order_date")

# Display the Gold DataFrame
print("--- Gold Layer: Daily Sales Summary ---")
gold_daily_sales_df.display()

Explanation:

• from pyspark.sql.functions import sum, count, countDistinct: Importing more aggregation functions.
• silver_df_for_gold = spark.read.format("delta").load(silver_orders_path): Reading from our Silver table.
• gold_daily_sales_df = silver_df_for_gold.groupBy("order_date"): We group our data by the order_date to perform daily aggregations.
• .agg(...): This is where we define our aggregation functions:
  • sum(col("total_amount")).alias("daily_revenue"): Calculates the sum of total_amount for each day, aliased as daily_revenue.
  • count(col("order_id")).alias("total_orders"): Counts the number of orders each day.
  • countDistinct(col("customer_name_cleaned")).alias("unique_customers"): Counts the number of unique customers each day.
• .orderBy("order_date"): Sorts the results by date for better readability.

Run this cell. You should see a summary of daily revenue, total orders, and unique customers.

Step 6: Load into Gold Delta Table and Final Query

Let’s save this aggregated data into our Gold layer and then query it.

Add this to a new cell:

# Define the path for our Gold layer Delta table
gold_daily_sales_path = f"{base_path}/gold/daily_sales"
gold_daily_sales_table_name = "gold_daily_sales_summary"

# Write the aggregated DataFrame to the Gold layer as a Delta table
gold_daily_sales_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .save(gold_daily_sales_path)

# Create a table entry in the Metastore for easy querying
spark.sql(f"CREATE TABLE IF NOT EXISTS {gold_daily_sales_table_name} USING DELTA LOCATION '{gold_daily_sales_path}'")

print(f"Aggregated data loaded into Gold Delta table at: {gold_daily_sales_path}")
print(f"Table '{gold_daily_sales_table_name}' created/updated.")

# Final verification: Query the Gold table using SQL
print("\n--- Final Data in Gold Layer (SQL Query) ---")
spark.sql(f"SELECT * FROM {gold_daily_sales_table_name} WHERE daily_revenue > 100 ORDER BY order_date DESC").display()

# Clean up our temporary raw data file after pipeline is complete
dbutils.fs.rm(raw_csv_path, True)
print(f"\nCleaned up raw CSV file: {raw_csv_path}")

Explanation:

• This cell follows the same pattern as loading to Bronze and Silver, but for our Gold layer.
• We use .mode("overwrite") again, as the daily summary will be fully recalculated each time.
• The final spark.sql query demonstrates how a business user or analyst might interact with this curated Gold layer data, filtering for revenues above a certain threshold and ordering by date.
• dbutils.fs.rm(raw_csv_path, True): This is a good practice to clean up temporary raw files once they’ve been successfully ingested into the Bronze layer, preventing clutter. True means recursive deletion.

Run this cell. Congratulations! You’ve successfully built an end-to-end ETL pipeline using Databricks and the Medallion Architecture!

Mini-Challenge: Enhance Your Silver Layer!

You’ve built a solid foundation. Now, let’s make it a bit more robust!

Challenge: Modify the Silver Layer transformation to add two new columns:

1. order_month: Extract the month (as an integer, e.g., 11 for November) from the order_date.
2. is_high_value_order: A boolean (True/False) column that is True if total_amount is greater than or equal to 200.00, and False otherwise.

After adding these, re-run your notebook from the Silver layer transformation cell onwards to see the changes propagate to the Gold layer. What new aggregation could you add to the Gold layer now that you have order_month?

Hint:

• For order_month, look for PySpark SQL functions related to date manipulation, specifically month().
• For is_high_value_order, the when() function (which we already used!) will be very useful, combined with a comparison operator.

What to observe/learn:

• How easily you can add new transformations to existing layers.
• How changes in lower layers (Silver) propagate and can enrich higher layers (Gold).
• The modularity of the Medallion Architecture.

(Pause here and try to implement the challenge yourself before looking at a potential solution!)

Potential Solution (Don’t peek until you’ve tried!):

# (Re-run from here if you're doing the challenge)
from pyspark.sql.functions import col, when, lit, current_timestamp, month, sum, count, countDistinct

# Read data from the Bronze layer
bronze_df = spark.read.format("delta").load(bronze_orders_path)

# Perform transformations for the Silver layer
silver_df = bronze_df.select(
    col("order_id"),
    (col("quantity") * col("price")).alias("total_amount"),
    when(col("customer_name").isNull() | (col("customer_name") == ""), lit("Anonymous"))
        .otherwise(col("customer_name"))
        .alias("customer_name_cleaned"),
    col("product_name"),
    col("quantity"),
    col("price"),
    col("order_date"),
    col("status"),
    current_timestamp().alias("processing_timestamp"),
    # New column 1: order_month
    month(col("order_date")).alias("order_month"),
    # New column 2: is_high_value_order
    (col("quantity") * col("price") >= 200.00).alias("is_high_value_order")
)

# Display the transformed DataFrame and its schema
print("--- Silver Orders DataFrame (with new columns) ---")
silver_df.printSchema()
silver_df.display()

# --- Re-run Silver Delta Table Load ---
silver_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .save(silver_orders_path)
spark.sql(f"CREATE TABLE IF NOT EXISTS {silver_orders_table_name} USING DELTA LOCATION '{silver_orders_path}'")
print(f"Cleaned data loaded into Silver Delta table at: {silver_orders_path}")

# --- Re-run Gold Layer Aggregation ---
silver_df_for_gold = spark.read.format("delta").load(silver_orders_path)

gold_daily_sales_df = silver_df_for_gold.groupBy("order_date", "order_month") \
    .agg(
        sum(col("total_amount")).alias("daily_revenue"),
        count(col("order_id")).alias("total_orders"),
        countDistinct(col("customer_name_cleaned")).alias("unique_customers"),
        sum(when(col("is_high_value_order"), 1).otherwise(0)).alias("high_value_orders_count") # New aggregation
    ) \
    .orderBy("order_date")

print("\n--- Gold Layer: Daily Sales Summary (with new aggregations) ---")
gold_daily_sales_df.display()

# --- Re-run Gold Delta Table Load ---
gold_daily_sales_df.write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .save(gold_daily_sales_path)
spark.sql(f"CREATE TABLE IF NOT EXISTS {gold_daily_sales_table_name} USING DELTA LOCATION '{gold_daily_sales_path}'")
print(f"Aggregated data loaded into Gold Delta table at: {gold_daily_sales_path}")

# Final verification: Query the Gold table using SQL
print("\n--- Final Data in Gold Layer (SQL Query) with new columns ---")
spark.sql(f"SELECT * FROM {gold_daily_sales_table_name} WHERE daily_revenue > 100 ORDER BY order_date DESC").display()

You’ll notice that the total_amount calculation for order 1001 (Laptop) is 1200.00, which is >= 200.00, so is_high_value_order would be True. For order 1002 (Mouse), total_amount is 51.00, so is_high_value_order would be False.

In the Gold layer, you could now group by order_month to get monthly summaries, or add a count of high_value_orders_count per day/month!

Common Pitfalls & Troubleshooting

Building ETL pipelines can be complex. Here are a few common issues and how to approach them:

1. Schema Evolution Issues:

   • Pitfall: Source data schemas change unexpectedly (e.g., a new column is added, or a column’s data type changes). Without proper handling, this can break your pipeline.
   • Troubleshooting: Delta Lake’s mergeSchema option (which we used!) is your best friend for handling new columns. For data type changes, you might need explicit casting (df.withColumn("column_name", col("column_name").cast("new_type"))) or a more robust schema validation step in your Bronze layer. Always keep an eye on your printSchema() output!

2. Performance Bottlenecks with Large Data:

   • Pitfall: Your pipeline runs slowly, especially as data volumes grow. This could be due to inefficient joins, too many small files, or incorrect cluster sizing.
   • Troubleshooting:
     • Small Files: Many small files can overwhelm Spark’s metadata processing. Use OPTIMIZE commands with ZORDER BY (if applicable) on your Delta tables to compact files and improve query performance. For example: spark.sql(f"OPTIMIZE {bronze_orders_table_name} ZORDER BY (order_date)").
     • Partitioning: While Delta Lake often reduces the need for manual partitioning, consider partitioning large tables by low-cardinality columns (like order_date or country) if those are frequently used in filters.
     • Cluster Sizing: Ensure your Databricks cluster has enough cores and memory for your workload. Databricks Serverless Compute handles this automatically, but for fixed-size clusters, monitor Spark UI for bottlenecks.
     • Caching: For frequently accessed intermediate DataFrames, df.cache() can sometimes help, but use it judiciously.

3. Data Quality Issues:

   • Pitfall: Nulls, incorrect data types, duplicates, or inconsistent values sneak through, leading to inaccurate analyses in the Gold layer.
   • Troubleshooting:
     • Validation Steps: Implement explicit validation steps in your Silver layer. Use filter() to remove bad records, or when() statements to correct values (as we did for customer_name).
     • Expectations (Delta Live Tables): For production-grade pipelines, Databricks Delta Live Tables (DLT) offer powerful “expectations” to define data quality rules and automatically quarantine or alert on bad data. While beyond this chapter, it’s a critical concept for robust ETL.
     • Monitoring: Set up monitoring and alerts on data quality metrics in your Silver and Gold layers.

Summary

Phew! You’ve just completed a significant milestone in your Databricks journey. Let’s recap what we’ve learned and achieved:

• ETL Fundamentals: You now have a solid understanding of what Extract, Transform, and Load entail and why they are crucial for data processing.
• Databricks for ETL: You’ve seen how Databricks, with Apache Spark and Delta Lake, provides a powerful and unified platform for building robust ETL pipelines.
• Medallion Architecture: You’ve implemented the industry-standard Bronze, Silver, and Gold layering strategy, promoting data quality, reusability, and maintainability.
  • Bronze: Ingested raw, immutable data.
  • Silver: Cleaned, conformed, and enriched data.
  • Gold: Curated, aggregated, business-ready data for analytics.
• Hands-on Implementation: You successfully built a complete pipeline, from simulating raw data ingestion to generating an analytical summary, using PySpark DataFrame operations and Delta Lake writes.
• Troubleshooting: You’re aware of common pitfalls like schema evolution, performance, and data quality, and know initial steps to address them.

This project laid the groundwork for building complex, production-ready data pipelines. You now have the practical experience to design and implement your own ETL solutions on Databricks.

What’s Next? In upcoming chapters, we’ll explore how to automate these pipelines using Databricks Workflows (Jobs), delve into more advanced transformations, discuss streaming ETL, and even touch upon data governance with Unity Catalog in more detail. Keep building, keep learning!

References

• Databricks Official Documentation: What is the Lakehouse?
• Databricks Official Documentation: Delta Lake
• Databricks Official Documentation: Medallion Architecture
• Apache Spark Official Documentation
• Microsoft Learn: Best practices for performance efficiency (Azure Databricks)

This page is AI-assisted and reviewed. It references official documentation and recognized resources where relevant.