ETL Pipelines with Azure Databricks and Data Factory

Course: Build and Run ETL Pipelines with Azure Databricks and Azure Data Factory Level: Intermediate / Advanced | Platform: Azure Databricks + Azure Data Factory

Table of Contents

1. Introduction to ETL and Apache Spark
2. Azure Databricks Architecture
3. Unity Catalog — Centralized Governance
4. Databricks vs Azure Data Factory
5. Environment Setup
6. Connecting to Azure Data Lake Storage from Databricks
7. Apache Spark DataFrames — Fundamentals
8. Schema Definition
9. Data Analysis and Cleaning
10. Business Transformations with PySpark
11. SQL Queries on DataFrames
12. Handling Corrupted Data
13. Delta Lake — Foundations and Architecture
14. Writing to Data Lake and Delta Tables
15. DML Operations on Delta Tables
16. Delta Lake Performance Optimizations
17. Delta Lake Auto-Optimization
18. Automation with Databricks Workflows
19. Parameterizing Notebooks with Widgets
20. Task Values and Dependencies Between Tasks
21. Triggers and Job Automation
22. Git Integration with Databricks
23. Orchestration with Azure Data Factory
24. Invoking Databricks from Data Factory
25. Automating ADF Pipelines with Triggers
26. Advanced ETL Architecture Patterns
27. Summary and Best Practices
28. Glossary

1. Introduction to ETL and Apache Spark

1.1 What is ETL?

ETL stands for Extract, Transform, Load — it is the fundamental process of data engineering:

flowchart LR
    subgraph Extract["1. EXTRACT"]
        S1["(Customer DB\nOracle/SQL)"]
        S2[CSV/JSON Files\nFTP/S3/ADLS]
        S3["(NoSQL\nMongoDB/Cassandra)"]
        S4[REST API\nSalesforce/SAP]
    end

    subgraph Transform["2. TRANSFORM"]
        T1[Cleaning\nnull values, duplicates]
        T2[Enrichment\njoins, lookups]
        T3[Aggregation\ngroupBy, pivots]
        T4[Business Logic\nmargins, KPIs]
    end

    subgraph Load["3. LOAD"]
        L1["(Data Warehouse\nSynapse Analytics)"]
        L2[Data Lake\nADLS Gen2]
        L3["(Delta Table\nDatabricks)"]
        L4[Power BI\nTableau]
    end

    S1 --> T1
    S2 --> T1
    S3 --> T1
    S4 --> T1
    T1 --> T2 --> T3 --> T4
    T4 --> L1
    T4 --> L2
    T4 --> L3
    L3 --> L4

1.2 Challenges of Traditional ETL Tools

Traditional ETL tools (Informatica, SSIS, Talend) face growing limitations:

1.3 Apache Spark Architecture

Apache Spark is the in-memory distributed processing engine that powers Azure Databricks:

graph TB
    subgraph "Spark Application"
        Driver["Driver Process (JVM)\n• Analyzes the code\n• Creates the execution plan\n• Distributes the work"]
        
        subgraph "Worker Cluster"
            E1["Executor 1 (JVM)\n• Executes tasks\n• Caches data\n• Returns results"]
            E2["Executor 2 (JVM)\n• Executes tasks\n• Caches data"]
            E3["Executor N (JVM)\n• Executes tasks\n• Caches data"]
        end
    end
    
    subgraph "Storage Layer"
        ADLS["(Azure Data Lake\nStorage Gen2)"]
        Delta[Delta Lake\nParquet + Log]
    end
    
    Driver -->|Distributes tasks| E1
    Driver -->|Distributes tasks| E2
    Driver -->|Distributes tasks| E3
    E1 -->|Read/Write| ADLS
    E2 -->|Read/Write| ADLS
    E3 -->|Read/Write| ADLS
    ADLS <--> Delta

Key Spark characteristics:

1.4 Lazy Evaluation — Fundamental Principle

# Demonstration of Lazy Evaluation in PySpark

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("DemoLazyEval").getOrCreate()

# These lines do NOT execute yet — they are transformations
df = spark.read.csv("abfss://data@storage.dfs.core.windows.net/taxi.csv",
                     header=True, inferSchema=True)  # LAZY
df_filtered = df.filter(F.col("passenger_count") > 0)           # LAZY
df_enriched = df_filtered.withColumn(
    "trip_duration_min",
    (F.col("dropoff_datetime").cast("long") - 
     F.col("pickup_datetime").cast("long")) / 60
)                                                                  # LAZY
df_aggregated = df_enriched.groupBy("VendorID").agg(
    F.count("*").alias("num_trips"),
    F.avg("trip_duration_min").alias("avg_duration_min"),
    F.sum("total_amount").alias("total_revenue")
)                                                                  # LAZY

# Only here does Spark execute the entire optimized plan at once
df_aggregated.show()  # ACTION → triggers execution
df_aggregated.write.format("delta").save("/output/aggregated_taxi")  # ACTION

2. Azure Databricks Architecture

2.1 Unified Data Intelligence Platform

graph LR
    subgraph "Azure Databricks Platform"
        subgraph "Persona Layer"
            DE[Data Engineering\nETL, Pipelines]
            DS[Data Science\nML, Analytics]
            SQL2[Data Warehousing\nDatabricks SQL]
        end
        
        subgraph "Compute Layer"
            Spark[Apache Spark\nOpen Source]
            Photon[Photon Engine\nNative Databricks]
            Serverless[Serverless\nOn-Demand]
        end
        
        subgraph "Data Layer"
            Delta[Delta Lake\nACID Transactions]
            UC[Unity Catalog\nGovernance]
            DBFS[DBFS / Volumes\nStorage]
        end
    end
    
    subgraph "Azure Services"
        ADLS["(ADLS Gen2)"]
        KV[Key Vault]
        ADF[Data Factory]
        AEH[Event Hubs]
    end
    
    DE & DS & SQL2 --> Spark & Photon & Serverless
    Spark & Photon --> Delta
    Delta --> UC
    Delta <--> ADLS
    ADF -->|Orchestration| DE
    AEH -->|Streaming| DE
    KV -->|Secrets| DE

2.2 Detailed Components

3. Unity Catalog — Centralized Governance

3.1 3-Level Architecture

Unity Catalog organizes data in a 3-level hierarchy:

graph TB
    UC[Unity Catalog Metastore] --> C1[Catalog: taxicatalog]
    UC --> C2[Catalog: ml_catalog]
    UC --> C3[Catalog: hive_metastore\nlegacy]
    
    C1 --> S1[Schema: rides]
    C1 --> S2[Schema: dimensions]
    C1 --> S3[Schema: staging]
    
    S1 --> T1[Table: yellow_taxis]
    S1 --> T2[Table: green_taxis]
    S2 --> T3[Table: rate_codes]
    S2 --> T4[Table: taxi_zones]
    S3 --> V1[View: vw_daily_revenue]

3-level notation: catalog.schema.table

Example: taxicatalog.rides.yellow_taxis

3.2 Unity Catalog vs Hive Metastore Advantages

3.3 Configure Unity Catalog — Steps

# 1. Create an Access Connector (managed identity for Databricks)
az databricks access-connector create \
  --name "databricks-access-connector" \
  --resource-group "ETL-RG" \
  --location "eastus2" \
  --identity-type "SystemAssigned"

# 2. Create an ADLS container for the metastore
az storage container create \
  --account-name "myunitymetastoredls" \
  --name "unity-metastore" \
  --auth-mode login

# 3. Assign Storage Blob Data Contributor role to the Access Connector
CONNECTOR_PRINCIPAL_ID=$(az databricks access-connector show \
  --name "databricks-access-connector" \
  --resource-group "ETL-RG" \
  --query "identity.principalId" -o tsv)

az role assignment create \
  --assignee "$CONNECTOR_PRINCIPAL_ID" \
  --role "Storage Blob Data Contributor" \
  --scope "/subscriptions/{sub-id}/resourceGroups/ETL-RG/providers/Microsoft.Storage/storageAccounts/myunitymetastoredls"

# Create catalog and schema in Databricks
# (in a notebook with Unity Catalog admin)

# Create a catalog
spark.sql("CREATE CATALOG IF NOT EXISTS taxicatalog")
spark.sql("USE CATALOG taxicatalog")

# Create schemas
spark.sql("CREATE SCHEMA IF NOT EXISTS taxicatalog.rides COMMENT 'Taxi ride data'")
spark.sql("CREATE SCHEMA IF NOT EXISTS taxicatalog.dimensions COMMENT 'Dimension tables'")
spark.sql("CREATE SCHEMA IF NOT EXISTS taxicatalog.staging COMMENT 'Staging area'")

4. Databricks vs Azure Data Factory

4.1 Comparison for ETL Pipelines

4.2 Recommended Combined Architecture

graph LR
    subgraph "Sources"
        SQL["(Azure SQL DB\n+ On-Prem SQL)"]
        S3["(S3 / ADLS\nFiles)"]
        API[REST APIs\nSalesforce/SAP]
    end

    subgraph "Azure Data Factory"
        Copy[Copy Activity\nIngestion]
        Control[Control Flow\nLookup/ForEach/If]
        Trigger1[ADF Triggers\nSchedule/Event]
    end

    subgraph "Azure Databricks"
        Raw[Raw Zone\nDelta Lake]
        Trans[Transformations\nPySpark/SQL]
        Curated[Curated Zone\nDelta Tables]
        Workflow[Databricks\nWorkflows]
    end

    subgraph "Consumption"
        PBI[Power BI]
        ML[ML Models]
        API2[Serving APIs]
    end

    SQL --> Copy
    S3 --> Copy
    API --> Copy
    Copy --> Raw
    Control --> Trans
    Trigger1 -->|Triggers| Workflow
    Workflow --> Trans
    Raw --> Trans
    Trans --> Curated
    Curated --> PBI
    Curated --> ML
    Curated --> API2

5. Environment Setup

5.1 Required Azure Resources

# Variables
RESOURCE_GROUP="ETL-RG"
LOCATION="eastus2"
STORAGE_ACCOUNT="taxidatadls"
CONTAINER_NAME="taxidata"
ADF_NAME="etl-data-factory"
ADB_WORKSPACE="psworkspace-eastus2"

# Create the resource group
az group create --name $RESOURCE_GROUP --location $LOCATION

# Create the ADLS Gen2 account
az storage account create \
  --name $STORAGE_ACCOUNT \
  --resource-group $RESOURCE_GROUP \
  --location $LOCATION \
  --sku Standard_LRS \
  --kind StorageV2 \
  --hierarchical-namespace true  # Enables ADLS Gen2

# Create the container
az storage container create \
  --account-name $STORAGE_ACCOUNT \
  --name $CONTAINER_NAME \
  --auth-mode login

# Create directories
az storage fs directory create \
  --account-name $STORAGE_ACCOUNT \
  --file-system $CONTAINER_NAME \
  --name "raw" \
  --auth-mode login

az storage fs directory create \
  --account-name $STORAGE_ACCOUNT \
  --file-system $CONTAINER_NAME \
  --name "output" \
  --auth-mode login

# Create Azure Data Factory
az datafactory create \
  --resource-group $RESOURCE_GROUP \
  --factory-name $ADF_NAME \
  --location $LOCATION

# Create Premium Databricks workspace (required for Unity Catalog)
az databricks workspace create \
  --name $ADB_WORKSPACE \
  --resource-group $RESOURCE_GROUP \
  --location $LOCATION \
  --sku premium

5.2 Spark Cluster Configuration

{
  "cluster_name": "etl-production-cluster",
  "spark_version": "13.3.x-scala2.12",
  "node_type_id": "Standard_DS4_v2",
  "driver_node_type_id": "Standard_DS4_v2",
  "autoscale": {
    "min_workers": 2,
    "max_workers": 8
  },
  "autotermination_minutes": 60,
  "spark_conf": {
    "spark.sql.adaptive.enabled": "true",
    "spark.databricks.delta.optimizeWrite.enabled": "true",
    "spark.databricks.delta.autoCompact.enabled": "true",
    "spark.sql.shuffle.partitions": "200"
  },
  "custom_tags": {
    "team": "data-engineering",
    "environment": "production",
    "project": "taxi-etl"
  }
}

6. Connecting to Azure Data Lake Storage from Databricks

6.1 Available Connection Methods

graph LR
    DB[Databricks] --> M1[Without Unity Catalog\nLegacy Access Patterns]
    DB --> M2[With Unity Catalog\nCredentials + External Location]
    
    M1 --> L1[Access Keys]
    M1 --> L2[SAS Token]
    M1 --> L3[Service Principal\nEntra ID]
    M1 --> L4[Credential Passthrough]
    
    M2 --> U1[Storage Credentials\nService Principal / Managed Identity]
    M2 --> U2[External Locations\nMapping credential → path]

6.2 Connection via Access Key (Legacy — development only)

# ⚠️ WARNING: Never hardcode keys in production code!
# Use Azure Key Vault or Databricks secrets

# Configuration via Access Key (legacy, for local testing only)
storage_account_name = "taxidatadls"
storage_account_key = dbutils.secrets.get(
    scope="kv-secrets",
    key="adls-access-key"
)
container_name = "taxidata"

# Configure access in SparkContext
spark.conf.set(
    f"fs.azure.account.key.{storage_account_name}.dfs.core.windows.net",
    storage_account_key
)

# ABFSS path (Azure Blob File System Secured)
file_path = f"abfss://{container_name}@{storage_account_name}.dfs.core.windows.net/raw/"
print(f"Configured path: {file_path}")

# Read a file
df = spark.read.csv(f"{file_path}YellowTaxis_2023.csv", header=True)
df.show(5)

6.3 Connection via Service Principal (recommended for production)

# Connection via Service Principal (OAuth 2.0)
# Secrets are stored in Azure Key Vault and retrieved via dbutils.secrets

storage_account_name = "taxidatadls"
client_id = dbutils.secrets.get(scope="kv-secrets", key="sp-client-id")
client_secret = dbutils.secrets.get(scope="kv-secrets", key="sp-client-secret")
tenant_id = dbutils.secrets.get(scope="kv-secrets", key="tenant-id")

# OAuth configuration
spark.conf.set(
    f"fs.azure.account.auth.type.{storage_account_name}.dfs.core.windows.net",
    "OAuth"
)
spark.conf.set(
    f"fs.azure.account.oauth.provider.type.{storage_account_name}.dfs.core.windows.net",
    "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider"
)
spark.conf.set(
    f"fs.azure.account.oauth2.client.id.{storage_account_name}.dfs.core.windows.net",
    client_id
)
spark.conf.set(
    f"fs.azure.account.oauth2.client.secret.{storage_account_name}.dfs.core.windows.net",
    client_secret
)
spark.conf.set(
    f"fs.azure.account.oauth2.client.endpoint.{storage_account_name}.dfs.core.windows.net",
    f"https://login.microsoftonline.com/{tenant_id}/oauth2/token"
)

# Guaranteed access
raw_path = f"abfss://taxidata@{storage_account_name}.dfs.core.windows.net/raw/"
df = spark.read.csv(f"{raw_path}YellowTaxis_2023.csv", header=True)
print(f"Number of rows read: {df.count()}")

6.4 Connection with Unity Catalog (best practice)

-- Create a Storage Credential (once per admin)
CREATE STORAGE CREDENTIAL adls_production_credential
USING MANAGED IDENTITY;

-- Create an External Location
CREATE EXTERNAL LOCATION taxidata_raw
URL 'abfss://taxidata@taxidatadls.dfs.core.windows.net/raw'
WITH (STORAGE CREDENTIAL adls_production_credential);

-- Verify access
SHOW EXTERNAL LOCATIONS;
LIST 'abfss://taxidata@taxidatadls.dfs.core.windows.net/raw';

# With Unity Catalog, direct reading without access configuration
# The credential is transparently managed by Unity Catalog
df = spark.read.csv(
    "abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/YellowTaxis_2023.csv",
    header=True
)

7. Apache Spark DataFrames — Fundamentals

7.1 Key DataFrame Concepts

A Spark DataFrame is a distributed collection of data organized in columns, similar to a relational table:

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

spark = SparkSession.builder.appName("TaxiETL").getOrCreate()

# Read a CSV file with options
yellow_taxi_df = (
    spark.read
    .option("header", "true")
    .option("inferSchema", "true")
    .option("nullValue", "")
    .option("emptyValue", None)
    .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/YellowTaxis_2023.csv")
)

# Basic DataFrame information
print(f"Number of rows: {yellow_taxi_df.count()}")
print(f"Number of columns: {len(yellow_taxi_df.columns)}")
print(f"Columns: {yellow_taxi_df.columns}")

# Display the schema
yellow_taxi_df.printSchema()

# Show first rows
yellow_taxi_df.show(10, truncate=False)

7.2 Supported Spark Data Types

7.3 Reading Different File Formats

# Read a CSV with all options
df_csv = (
    spark.read
    .option("header", "true")
    .option("sep", ",")
    .option("quote", '"')
    .option("escape", '"')
    .option("encoding", "UTF-8")
    .option("dateFormat", "yyyy-MM-dd")
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
    .csv("abfss://container@storage.dfs.core.windows.net/raw/data.csv")
)

# Read JSON
df_json = (
    spark.read
    .option("multiline", "true")
    .json("abfss://container@storage.dfs.core.windows.net/raw/ratecodes.json")
)

# Read Parquet (performant columnar format)
df_parquet = spark.read.parquet(
    "abfss://container@storage.dfs.core.windows.net/output/yellowtaxis.parquet"
)

# Read a Delta table
df_delta = spark.read.format("delta").load(
    "abfss://container@storage.dfs.core.windows.net/output/yellowtaxis.delta"
)

# Read a Delta table by name (Unity Catalog)
df_table = spark.table("taxicatalog.rides.yellow_taxis")

8. Schema Definition

8.1 Automatic Inference vs Manual Schema

graph LR
    subgraph "InferSchema"
        I1[Full file scan]
        I2[Automatic type detection]
        I3[Slow on large files]
        I4[Can be incorrect]
        I1 --> I2 --> I3
        I3 --> I4
    end
    
    subgraph "Manual Schema"
        M1[Explicit definition]
        M2[Immediate validation]
        M3[Fast and reliable]
        M4[Recommended for production]
        M1 --> M2 --> M3
        M3 --> M4
    end

8.2 Defining a Schema Manually

from pyspark.sql.types import (
    StructType, StructField,
    IntegerType, StringType, DoubleType,
    TimestampType, LongType
)

# Complete schema definition for Yellow Taxi data
yellow_taxi_schema = StructType([
    StructField("VendorID",            IntegerType(),   nullable=True),
    StructField("tpep_pickup_datetime", TimestampType(), nullable=True),
    StructField("tpep_dropoff_datetime", TimestampType(), nullable=True),
    StructField("passenger_count",     IntegerType(),   nullable=True),
    StructField("trip_distance",       DoubleType(),    nullable=True),
    StructField("RatecodeID",          IntegerType(),   nullable=True),
    StructField("store_and_fwd_flag",  StringType(),    nullable=True),
    StructField("PULocationID",        IntegerType(),   nullable=True),
    StructField("DOLocationID",        IntegerType(),   nullable=True),
    StructField("payment_type",        IntegerType(),   nullable=True),
    StructField("fare_amount",         DoubleType(),    nullable=True),
    StructField("extra",               DoubleType(),    nullable=True),
    StructField("mta_tax",             DoubleType(),    nullable=True),
    StructField("tip_amount",          DoubleType(),    nullable=True),
    StructField("tolls_amount",        DoubleType(),    nullable=True),
    StructField("improvement_surcharge", DoubleType(),  nullable=True),
    StructField("total_amount",        DoubleType(),    nullable=True),
    StructField("congestion_surcharge", DoubleType(),   nullable=True),
    StructField("airport_fee",         DoubleType(),    nullable=True),
])

# Read with manually defined schema
yellow_taxi_df = (
    spark.read
    .schema(yellow_taxi_schema)
    .option("header", "true")
    .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/YellowTaxis_2023.csv")
)

# Validate the schema
yellow_taxi_df.printSchema()
print(f"Rows read: {yellow_taxi_df.count():,}")

9. Data Analysis and Cleaning

9.1 Data Quality Framework

mindmap
  root((Data\nQuality))
    Completeness
      Remove nulls
      Fill missing values
      Identify required fields
    Uniqueness
      Detect duplicates
      Keep the most recent version
      Deduplication key
    Timeliness
      Check date ranges
      Filter future dates
      Consistent pickup ≤ dropoff dates
    Accuracy
      Values within expected ranges
      Logical consistency
      Valid reference data

9.2 Exploratory Analysis with Spark

# Statistical analysis of numeric columns
print("=== Descriptive Statistics ===")
yellow_taxi_df.describe(
    "passenger_count", 
    "trip_distance", 
    "fare_amount", 
    "total_amount"
).show()

# Count null values per column
from pyspark.sql import functions as F

null_counts = yellow_taxi_df.select([
    F.count(F.when(F.col(c).isNull(), c)).alias(c)
    for c in yellow_taxi_df.columns
])
print("=== Null Values Per Column ===")
null_counts.show(vertical=True)

# Distribution of values in a categorical column
yellow_taxi_df.groupBy("VendorID").count().orderBy("VendorID").show()

# Date range
yellow_taxi_df.agg(
    F.min("tpep_pickup_datetime").alias("min_date"),
    F.max("tpep_pickup_datetime").alias("max_date")
).show()

9.3 Complete Data Cleaning

from pyspark.sql import functions as F
from pyspark.sql.window import Window
from datetime import datetime

def clean_taxi_data(df):
    """
    Applies all data quality rules to a Yellow Taxi DataFrame.
    
    Rules applied:
    - Remove records with passengers = 0 or > 9
    - Remove negative or null distances
    - Remove negative amounts
    - Date validation (not in the future, pickup before dropoff)
    - Remove duplicates
    
    Returns:
        Cleaned DataFrame with an 'etl_load_date' column
    """
    
    print(f"Rows before cleaning: {df.count():,}")
    
    # 1. Completeness: remove rows with critical nulls
    df_clean = df.dropna(subset=[
        "VendorID", 
        "tpep_pickup_datetime", 
        "tpep_dropoff_datetime",
        "passenger_count",
        "trip_distance",
        "total_amount"
    ])
    print(f"After removing critical nulls: {df_clean.count():,}")
    
    # 2. Accuracy: values within acceptable ranges
    df_clean = df_clean.filter(
        (F.col("passenger_count") >= 1) & 
        (F.col("passenger_count") <= 9)
    )
    print(f"After passenger_count filter (1-9): {df_clean.count():,}")
    
    df_clean = df_clean.filter(
        (F.col("trip_distance") > 0) & 
        (F.col("trip_distance") < 500)  # Max 500 miles
    )
    print(f"After trip_distance filter: {df_clean.count():,}")
    
    df_clean = df_clean.filter(
        (F.col("total_amount") > 0) & 
        (F.col("total_amount") < 5000)  # Max $5000
    )
    print(f"After total_amount filter: {df_clean.count():,}")
    
    # 3. Timeliness: date validation
    today = datetime.now()
    df_clean = df_clean.filter(
        (F.col("tpep_pickup_datetime") <= F.lit(today)) &
        (F.col("tpep_pickup_datetime") < F.col("tpep_dropoff_datetime")) &
        (F.col("tpep_pickup_datetime") >= F.lit("2009-01-01"))  # TLC data minimum
    )
    print(f"After date validation: {df_clean.count():,}")
    
    # 4. Uniqueness: remove exact duplicates
    df_clean = df_clean.dropDuplicates([
        "VendorID", "tpep_pickup_datetime", "tpep_dropoff_datetime",
        "PULocationID", "DOLocationID", "total_amount"
    ])
    print(f"After deduplication: {df_clean.count():,}")
    
    # 5. Add provenance flag
    df_clean = df_clean.withColumn("etl_load_date", F.current_date())
    df_clean = df_clean.withColumn("etl_source", F.lit("yellow_taxi_csv"))
    
    return df_clean

# Apply cleaning
yellow_taxi_clean = clean_taxi_data(yellow_taxi_df)

10. Business Transformations with PySpark

10.1 Column Selection and Renaming

from pyspark.sql import functions as F

# Select and rename relevant columns
yellow_taxi_transformed = yellow_taxi_clean.select(
    # VendorID as-is (by column name string)
    F.col("VendorID"),
    
    # passenger_count cast to Integer (F.col allows applying methods)
    F.col("passenger_count").cast("integer").alias("PassengerCount"),
    
    # trip_distance renamed
    F.col("trip_distance").alias("TripDistance"),
    
    # Temporal columns renamed
    F.col("tpep_pickup_datetime").alias("PickupTime"),
    F.col("tpep_dropoff_datetime").alias("DropTime"),
    
    # Locations
    F.col("PULocationID").alias("PickupLocationId"),
    F.col("DOLocationID").alias("DropLocationId"),
    
    # Amounts
    F.col("fare_amount").alias("FareAmount"),
    F.col("tip_amount").alias("TipAmount"),
    F.col("total_amount").alias("TotalAmount"),
    
    # ETL metadata
    F.col("etl_load_date"),
    F.col("etl_source")
)

print(f"Columns after transformation: {yellow_taxi_transformed.columns}")
yellow_taxi_transformed.printSchema()

10.2 Creating Derived Columns

# Enrichment with calculated columns
yellow_taxi_enriched = (
    yellow_taxi_transformed
    # Derived temporal columns
    .withColumn("PickupYear",  F.year("PickupTime"))
    .withColumn("PickupMonth", F.month("PickupTime"))
    .withColumn("PickupDay",   F.dayofmonth("PickupTime"))
    .withColumn("PickupHour",  F.hour("PickupTime"))
    .withColumn("PickupDayOfWeek", F.dayofweek("PickupTime"))  # 1=Sunday, 7=Saturday
    
    # Trip duration in minutes
    .withColumn(
        "TripDurationMinutes",
        (F.unix_timestamp("DropTime") - F.unix_timestamp("PickupTime")) / 60
    )
    
    # Average speed (miles per hour)
    .withColumn(
        "AvgSpeedMph",
        F.when(F.col("TripDurationMinutes") > 0,
               (F.col("TripDistance") / (F.col("TripDurationMinutes") / 60))
        ).otherwise(F.lit(None).cast("double"))
    )
    
    # Cost per mile
    .withColumn(
        "CostPerMile",
        F.when(F.col("TripDistance") > 0,
               F.col("TotalAmount") / F.col("TripDistance")
        ).otherwise(F.lit(None).cast("double"))
    )
    
    # Tip percentage
    .withColumn(
        "TipPercentage",
        F.when(F.col("FareAmount") > 0,
               (F.col("TipAmount") / F.col("FareAmount")) * 100
        ).otherwise(F.lit(0.0))
    )
    
    # Trip category
    .withColumn(
        "TripCategory",
        F.when(F.col("TripDistance") < 1, "Short")
         .when(F.col("TripDistance") < 5, "Medium")
         .when(F.col("TripDistance") < 15, "Long")
         .otherwise("Very Long")
    )
    
    # Filter invalid calculated values
    .filter(F.col("TripDurationMinutes") > 0)
    .filter(F.col("TripDurationMinutes") < 300)  # Max 5 hours
)

print(f"Rows after enrichment: {yellow_taxi_enriched.count():,}")
yellow_taxi_enriched.show(5, truncate=False)

10.3 Aggregations and Analysis

# Analysis by month and vendor
monthly_summary = (
    yellow_taxi_enriched
    .groupBy("PickupYear", "PickupMonth", "VendorID", "TripCategory")
    .agg(
        F.count("*").alias("NumTrips"),
        F.sum("TotalAmount").alias("TotalRevenue"),
        F.avg("TripDistance").alias("AvgDistance"),
        F.avg("TripDurationMinutes").alias("AvgDuration"),
        F.avg("TipPercentage").alias("AvgTipPct"),
        F.percentile_approx("TotalAmount", 0.5).alias("MedianAmount"),
        F.percentile_approx("TripDistance", 0.95).alias("P95Distance")
    )
    .orderBy("PickupYear", "PickupMonth", "VendorID")
)

monthly_summary.show(20)

# Top 10 pickup zones by revenue
top_pickup_zones = (
    yellow_taxi_enriched
    .groupBy("PickupLocationId")
    .agg(
        F.count("*").alias("NumPickups"),
        F.sum("TotalAmount").alias("TotalRevenue"),
        F.avg("TipPercentage").alias("AvgTipPct")
    )
    .orderBy(F.desc("TotalRevenue"))
    .limit(10)
)

top_pickup_zones.show()

11. SQL Queries on DataFrames

11.1 Creating Temporary Views

# Create a temporary view for using SQL
yellow_taxi_enriched.createOrReplaceTempView("YellowTaxis")

# Temporary views last for the Spark session
# For a longer duration, use a Global Temp View
yellow_taxi_enriched.createOrReplaceGlobalTempView("YellowTaxisGlobal")
# Access: SELECT * FROM global_temp.YellowTaxisGlobal

11.2 Complex SQL Queries

# Direct SQL query on the temporary view
result_sql = spark.sql("""
    SELECT 
        VendorID,
        PickupYear,
        PickupMonth,
        TripCategory,
        COUNT(*) AS NumTrips,
        SUM(TotalAmount) AS TotalRevenue,
        AVG(TripDurationMinutes) AS AvgDurationMin,
        AVG(TipPercentage) AS AvgTipPercent,
        PERCENTILE_APPROX(TotalAmount, 0.5) AS MedianAmount
    FROM YellowTaxis
    WHERE PickupYear = 2023
        AND TripDurationMinutes BETWEEN 1 AND 120
        AND TotalAmount > 0
    GROUP BY VendorID, PickupYear, PickupMonth, TripCategory
    HAVING COUNT(*) > 100
    ORDER BY PickupMonth, VendorID, NumTrips DESC
""")

result_sql.show(20)

# Query with joins (zone lookup join)
zones_df = spark.read.csv(
    "abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/taxi_zones.csv",
    header=True
)
zones_df.createOrReplaceTempView("TaxiZones")

trips_with_zones = spark.sql("""
    SELECT 
        t.VendorID,
        t.PickupTime,
        t.TotalAmount,
        t.TripDistance,
        pz.Zone AS PickupZone,
        pz.Borough AS PickupBorough,
        dz.Zone AS DropZone,
        dz.Borough AS DropBorough
    FROM YellowTaxis t
    LEFT JOIN TaxiZones pz ON t.PickupLocationId = pz.LocationID
    LEFT JOIN TaxiZones dz ON t.DropLocationId = dz.LocationID
    WHERE t.PickupYear = 2023
    ORDER BY t.TotalAmount DESC
    LIMIT 1000
""")

trips_with_zones.show(10)

11.3 Python ↔ SQL Interoperability

# The result of spark.sql() is a Python DataFrame
# Python and SQL transformations are equivalent in performance

# PYTHON VERSION
result_python = (
    yellow_taxi_enriched
    .filter((F.col("PickupYear") == 2023) & (F.col("TotalAmount") > 0))
    .groupBy("VendorID", "PickupMonth", "TripCategory")
    .agg(
        F.count("*").alias("NumTrips"),
        F.sum("TotalAmount").alias("TotalRevenue"),
        F.avg("TipPercentage").alias("AvgTipPercent")
    )
    .orderBy("PickupMonth", "VendorID")
)

# SQL VERSION — identical result
result_sql_equiv = spark.sql("""
    SELECT VendorID, PickupMonth, TripCategory,
           COUNT(*) AS NumTrips, SUM(TotalAmount) AS TotalRevenue,
           AVG(TipPercentage) AS AvgTipPercent
    FROM YellowTaxis
    WHERE PickupYear = 2023 AND TotalAmount > 0
    GROUP BY VendorID, PickupMonth, TripCategory
    ORDER BY PickupMonth, VendorID
""")

# Verification: both execution plans are identical
result_python.explain(extended=True)

12. Handling Corrupted Data

12.1 Spark Parsing Modes

12.2 Example with All 3 Modes

# PERMISSIVE mode (default) — captures corrupted records
df_permissive = (
    spark.read
    .option("mode", "PERMISSIVE")
    .option("columnNameOfCorruptRecord", "_corrupt_record")
    .schema(yellow_taxi_schema)
    .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/corrupted_test.csv",
         header=True)
)

# View corrupted records
corrupted = df_permissive.filter(F.col("_corrupt_record").isNotNull())
print(f"Corrupted records: {corrupted.count()}")
corrupted.show(truncate=False)

# DROPMALFORMED mode — silently drops errors
df_drop = (
    spark.read
    .option("mode", "DROPMALFORMED")
    .schema(yellow_taxi_schema)
    .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/corrupted_test.csv",
         header=True)
)

# FAILFAST mode — exception if an error is found
try:
    df_fail = (
        spark.read
        .option("mode", "FAILFAST")
        .schema(yellow_taxi_schema)
        .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/corrupted_test.csv",
             header=True)
    )
    df_fail.count()  # Triggers actual reading
except Exception as e:
    print(f"FAILFAST detected an error: {e}")

12.3 badRecordsPath — Quarantine for Errors

# Store corrupted records in a quarantine zone
df_with_quarantine = (
    spark.read
    .option("badRecordsPath",
            "abfss://taxidata@taxidatadls.dfs.core.windows.net/quarantine/yellow_taxi/")
    .schema(yellow_taxi_schema)
    .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/",
         header=True)
)

# Count valid rows
valid_count = df_with_quarantine.count()
print(f"Valid records: {valid_count:,}")

# Read quarantined records for analysis
quarantine_df = spark.read.json(
    "abfss://taxidata@taxidatadls.dfs.core.windows.net/quarantine/yellow_taxi/"
)
print(f"Quarantined records: {quarantine_df.count()}")
quarantine_df.show(truncate=False)

13. Delta Lake — Foundations and Architecture

13.1 What is Delta Lake?

Delta Lake is an open-source storage layer that brings relational database reliability to Data Lakes:

graph LR
    subgraph "Without Delta Lake"
        DF1[Spark DataFrame] -->|Write| P1[Parquet Files\nPartition 1]
        DF1 -->|Write| P2[Parquet Files\nPartition 2]
        P1 & P2 -->|No metadata| NoMeta[No\ntransaction log\nNo ACID\nNo schema]
    end

    subgraph "With Delta Lake"
        DF2[Spark DataFrame] -->|Write| P3[Parquet Files\nPartition 1]
        DF2 -->|Write| P4[Parquet Files\nPartition 2]
        P3 & P4 --> TL["_delta_log/\n000.json, 001.json...\nTransaction Log"]
        TL --> Features[ACID Transactions\nTime Travel\nSchema Evolution\nFull Audit]
    end

13.2 Transaction Log Architecture

sequenceDiagram
    participant App as Application
    participant Delta as Delta Lake
    participant Storage as ADLS Gen2

    App->>Delta: Write version 0 (files F1, F2)
    Delta->>Storage: Write F1.parquet, F2.parquet
    Delta->>Storage: Write _delta_log/00000000000000000000.json
    Note over Storage: 00000000000000000000.json contains:\nadd F1, add F2, schema metadata

    App->>Delta: Update (update)
    Delta->>Storage: Write F3.parquet (updated data)
    Delta->>Storage: Write _delta_log/00000000000000000001.json
    Note over Storage: 00000000000000000001.json contains:\nremove F1, add F3

    App->>Delta: DELETE record
    Delta->>Storage: Write F4.parquet (without deleted records)
    Delta->>Storage: Write _delta_log/00000000000000000002.json
    Note over Storage: Old files remain for Time Travel

13.3 Key Delta Lake Features

14. Writing to Data Lake and Delta Tables

14.1 Writing in Parquet vs Delta Format

# OPTION 1: Write in Parquet (without Delta Lake)
yellow_taxi_enriched.write \
    .mode("overwrite") \
    .partitionBy("VendorID") \
    .format("parquet") \
    .save("abfss://taxidata@taxidatadls.dfs.core.windows.net/output/yellowtaxis.parquet")

# OPTION 2: Write in Delta (recommended)
yellow_taxi_enriched.write \
    .mode("overwrite") \
    .partitionBy("VendorID") \
    .format("delta") \
    .save("abfss://taxidata@taxidatadls.dfs.core.windows.net/output/yellowtaxis.delta")

14.2 Create a Delta Table with Unity Catalog

# Write as Delta table in Unity Catalog
yellow_taxi_enriched.write \
    .mode("overwrite") \
    .partitionBy("VendorID", "PickupYear", "PickupMonth") \
    .format("delta") \
    .option("path", "abfss://taxidata@taxidatadls.dfs.core.windows.net/tables/yellow_taxis") \
    .saveAsTable("taxicatalog.rides.yellow_taxis")

# Verification
spark.sql("DESCRIBE TABLE EXTENDED taxicatalog.rides.yellow_taxis").show(50, truncate=False)
spark.sql("SELECT COUNT(*) FROM taxicatalog.rides.yellow_taxis").show()

14.3 Time Travel — Access Previous Versions

# Time Travel by version number
df_version_0 = spark.read \
    .format("delta") \
    .option("versionAsOf", 0) \
    .load("abfss://taxidata@taxidatadls.dfs.core.windows.net/output/yellowtaxis.delta")

# Time Travel by timestamp
df_historical = spark.read \
    .format("delta") \
    .option("timestampAsOf", "2024-01-15 10:00:00") \
    .load("abfss://taxidata@taxidatadls.dfs.core.windows.net/output/yellowtaxis.delta")

# Via SQL with Unity Catalog
df_old = spark.sql("""
    SELECT * FROM taxicatalog.rides.yellow_taxis
    VERSION AS OF 3
""")

df_timestamp = spark.sql("""
    SELECT * FROM taxicatalog.rides.yellow_taxis
    TIMESTAMP AS OF '2024-01-15 10:00:00'
""")

# View full history
spark.sql("DESCRIBE HISTORY taxicatalog.rides.yellow_taxis").show(10, truncate=False)

15. DML Operations on Delta Tables

15.1 INSERT, UPDATE, DELETE

-- INSERT a single record
INSERT INTO taxicatalog.rides.yellow_taxis
(VendorID, PickupTime, DropTime, PassengerCount, TripDistance, TotalAmount, PickupYear, PickupMonth)
VALUES (3, '2023-12-15 10:30:00', '2023-12-15 11:00:00', 2, 5.3, 28.50, 2023, 12);

-- Verify the insert
SELECT * FROM taxicatalog.rides.yellow_taxis WHERE VendorID = 3;

-- UPDATE records
UPDATE taxicatalog.rides.yellow_taxis
SET TipAmount = TipAmount * 1.1  -- +10% tip
WHERE PickupYear = 2023 AND PickupMonth = 12;

-- DELETE records
DELETE FROM taxicatalog.rides.yellow_taxis
WHERE TripDistance < 0 OR TotalAmount < 0;

-- Verify history after DML
DESCRIBE HISTORY taxicatalog.rides.yellow_taxis;

15.2 DataFrame Append

# Read a file with new data
new_data_df = (
    spark.read
    .schema(yellow_taxi_schema)
    .option("header", "true")
    .csv("abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/YellowTaxis_2024_01.csv")
)

# Clean and transform new data
new_data_clean = clean_taxi_data(new_data_df)
new_data_enriched = (
    new_data_clean.select(yellow_taxi_enriched.columns)
)

# Append to the existing Delta table
new_data_enriched.write \
    .mode("append") \
    .format("delta") \
    .saveAsTable("taxicatalog.rides.yellow_taxis")

# Verify stats after append
spark.sql("""
    SELECT PickupYear, PickupMonth, COUNT(*) AS NumTrips
    FROM taxicatalog.rides.yellow_taxis
    GROUP BY PickupYear, PickupMonth
    ORDER BY PickupYear, PickupMonth
""").show()

15.3 MERGE (UPSERT) — SCD Type 1 Pattern

from delta.tables import DeltaTable

# Reference the target Delta table
delta_table = DeltaTable.forName(spark, "taxicatalog.rides.yellow_taxis")

# Source of updates
updates_df = spark.read.format("delta").load(
    "abfss://taxidata@taxidatadls.dfs.core.windows.net/updates/yellow_taxis_corrections/"
)

# MERGE: INSERT if new, UPDATE if existing
(delta_table.alias("target")
    .merge(
        updates_df.alias("source"),
        """target.VendorID = source.VendorID 
           AND target.PickupTime = source.PickupTime
           AND target.PickupLocationId = source.PickupLocationId"""
    )
    .whenMatchedUpdate(set={
        "TotalAmount": "source.TotalAmount",
        "TipAmount": "source.TipAmount",
        "etl_load_date": "current_date()"
    })
    .whenNotMatchedInsertAll()
    .execute()
)

print("MERGE completed successfully")

16. Delta Lake Performance Optimizations

16.1 Small Files Problem

graph LR
    subgraph "Before OPTIMIZE (problem)"
        W1[Worker 1] --> F1[F1: 2MB]
        W2[Worker 2] --> F2[F2: 1.5MB]
        W3[Worker 3] --> F3[F3: 3MB]
        W4[Worker 4] --> F4[F4: 1MB]
        W5[Worker 5] --> F5[F5: 2.5MB]
        F1 & F2 & F3 & F4 & F5 --> Overhead[Overhead: opening\n200 small files]
    end

    subgraph "After OPTIMIZE (bin packing)"
        OF1[F201: 512MB] 
        OF2[F202: 488MB]
        OF1 & OF2 --> Fast[Fast read:\nonly 2 files]
    end

16.2 OPTIMIZE and Z-ORDER

-- Compact small files (bin packing)
OPTIMIZE taxicatalog.rides.yellow_taxis;

-- OPTIMIZE with Z-ORDER on frequently filtered columns
-- Z-ORDER sorts data in files to co-locate similar values
OPTIMIZE taxicatalog.rides.yellow_taxis
ZORDER BY (PickupLocationId, DropLocationId, PickupYear, PickupMonth);

from delta.tables import DeltaTable

# OPTIMIZE via Python API
delta_table = DeltaTable.forName(spark, "taxicatalog.rides.yellow_taxis")

# Optimize a specific partition (useful for partitioned tables)
delta_table.optimize() \
    .where("PickupYear = 2023 AND PickupMonth = 12") \
    .executeZOrderBy("PickupLocationId", "DropLocationId")

16.3 When to Use Z-ORDER?

16.4 VACUUM — Cleaning Obsolete Files

-- Show what would be deleted (dry run)
VACUUM taxicatalog.rides.yellow_taxis DRY RUN;

-- Delete files > 7 days (default)
VACUUM taxicatalog.rides.yellow_taxis;

-- Delete files > 30 days (custom retention)
VACUUM taxicatalog.rides.yellow_taxis RETAIN 720 HOURS;

-- ⚠️ DANGER: Override minimum retention (disables safety guard)
-- This disables time travel for versions > 0 hours
-- SET spark.databricks.delta.retentionDurationCheck.enabled = false;
-- VACUUM taxicatalog.rides.yellow_taxis RETAIN 0 HOURS;
-- Never do this in production without fully understanding the implications!

VACUUM best practice: Define retention based on your time travel needs. If you keep 30 days of history, use RETAIN 720 HOURS. Run VACUUM at least once per week in production.

17. Delta Lake Auto-Optimization

17.1 Optimized Writes

# Enable write optimization at the cluster level
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# Or at the table level (persistent)
spark.sql("""
    ALTER TABLE taxicatalog.rides.yellow_taxis
    SET TBLPROPERTIES ('delta.autoOptimize.optimizeWrite' = 'true')
""")

17.2 Auto Compaction

# Enable auto compaction at the cluster level
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# Or at the table level
spark.sql("""
    ALTER TABLE taxicatalog.rides.yellow_taxis
    SET TBLPROPERTIES ('delta.autoOptimize.autoCompact' = 'true')
""")

17.3 Comparison of Optimization Strategies

18. Automation with Databricks Workflows

18.1 Databricks Workflow Architecture

graph TB
    subgraph "Databricks Job"
        subgraph "Parallel Tasks — Dimensions"
            T1[Task: Taxi Zones\nNotebook]
            T2[Task: Rate Codes\nNotebook]
            T3[Task: Payment Types\nNotebook]
        end
        
        subgraph "Control Flow"
            IF["Task: Check_Zones_Count\nIf/Else condition\n(TaxiZonesCount > 100)"]
        end
        
        subgraph "Fact Tasks"
            T4[Task: Yellow Taxis\nNotebook]
            T5[Task: Green Taxis\nNotebook]
        end
        
        subgraph "Finalization"
            T6[Task: Data Quality Report\nPython Script]
        end
    end
    
    T1 & T2 & T3 --> IF
    IF -->|True: Count OK| T4 & T5
    IF -->|False: Count KO| FAIL[Fail Job]
    T4 & T5 --> T6

18.2 Compute Options for Tasks

19. Parameterizing Notebooks with Widgets

19.1 Widget Types

import dbutils

# Text widget (free-form input)
dbutils.widgets.text("ProcessMonth", "2023-12", "Month to process")

# Dropdown widget
dbutils.widgets.dropdown("Environment", "dev", ["dev", "staging", "prod"], "Environment")

# Combobox widget (dropdown + free-form)
dbutils.widgets.combobox("VendorId", "1", ["1", "2", "3"], "Vendor ID")

# Multiselect widget
dbutils.widgets.multiselect("Months", "2023-12", ["2023-10", "2023-11", "2023-12"], "Months")

# Get values
process_month = dbutils.widgets.get("ProcessMonth")
environment = dbutils.widgets.get("Environment")
vendor_id = dbutils.widgets.get("VendorId")

print(f"Processing month: {process_month}")
print(f"Environment: {environment}")
print(f"Vendor: {vendor_id}")

# Usage in code
file_path = f"abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/YellowTaxis_{process_month}.csv"
output_table = f"taxicatalog.rides.yellow_taxis_{environment}"

print(f"Source file: {file_path}")
print(f"Target table: {output_table}")

19.2 Parameterized Production Notebook

# Notebook: ETL_Yellow_Taxis_Production.py
# Parameters via widgets

import dbutils
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import *

# Parameters
process_month = dbutils.widgets.get("ProcessMonth")  # e.g. "2023-12"
environment = dbutils.widgets.get("Environment")      # e.g. "prod"

# Configuration based on environment
configs = {
    "dev": {
        "storage_account": "taxidatadev",
        "catalog": "taxicatalog_dev",
        "max_files": 1
    },
    "staging": {
        "storage_account": "taxidatastg",
        "catalog": "taxicatalog_stg",
        "max_files": 5
    },
    "prod": {
        "storage_account": "taxidataprod",
        "catalog": "taxicatalog",
        "max_files": None
    }
}

config = configs[environment]
storage_account = config["storage_account"]
catalog = config["catalog"]

# Paths
raw_path = f"abfss://taxidata@{storage_account}.dfs.core.windows.net/raw/YellowTaxis_{process_month}.csv"
output_table = f"{catalog}.rides.yellow_taxis"

print(f"=== Start Yellow Taxis ETL ===")
print(f"Month: {process_month}")
print(f"Environment: {environment}")
print(f"Source: {raw_path}")
print(f"Target: {output_table}")

# ETL execution
spark = SparkSession.builder.getOrCreate()
df = spark.read.schema(yellow_taxi_schema).option("header", "true").csv(raw_path)
df_clean = clean_taxi_data(df)
df_enriched = transform_taxi_data(df_clean)

# Write
df_enriched.write \
    .mode("append") \
    .partitionBy("VendorID", "PickupYear", "PickupMonth") \
    .format("delta") \
    .saveAsTable(output_table)

row_count = df_enriched.count()
print(f"=== ETL complete: {row_count:,} rows processed ===")

# Return count as task value
dbutils.notebook.exit(str(row_count))

20. Task Values and Dependencies Between Tasks

20.1 Passing Values Between Tasks

# In the Taxi_Zones notebook (dimension task)
zones_df = spark.table("taxicatalog.dimensions.taxi_zones")
zones_count = zones_df.count()

print(f"Number of zones loaded: {zones_count}")

# Store value as Task Value (shareable between tasks)
dbutils.jobs.taskValues.set(key="TaxiZonesCount", value=zones_count)
dbutils.jobs.taskValues.set(key="LoadStatus", value="SUCCESS")

# In a downstream task — retrieve the value
zones_count = dbutils.jobs.taskValues.get(
    taskKey="Dimension_Taxi_Zones",
    key="TaxiZonesCount",
    default=0,
    debugValue=150  # Value in local debug mode
)

print(f"Available zones: {zones_count}")
if zones_count < 100:
    raise Exception(f"Only {zones_count} zones — pipeline stopped for data quality")

20.2 Conditions in Jobs (If/Else)

Databricks Workflows supports control flow tasks:

// If/Else task configuration in a job (via API)
{
  "task_key": "Check_Zones_Count",
  "description": "Verify that the number of zones is sufficient",
  "depends_on": [
    {"task_key": "Dimension_Taxi_Zones"},
    {"task_key": "Dimension_Rate_Codes"},
    {"task_key": "Dimension_Payment_Types"}
  ],
  "condition_task": {
    "op": "GREATER",
    "left": "{{tasks.Dimension_Taxi_Zones.values.TaxiZonesCount}}",
    "right": "100"
  }
}

21. Triggers and Job Automation

21.1 Available Trigger Types

21.2 Configure a Scheduled Trigger

// Trigger configuration via Databricks Jobs API
{
  "job_id": 12345,
  "schedule": {
    "quartz_cron_expression": "0 0 2 * * ?",
    "timezone_id": "America/New_York",
    "pause_status": "UNPAUSED"
  }
}

Useful cron expressions:

21.3 File Arrival Trigger

# Configure a file arrival trigger via SDK API
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import FileArrivalTriggerConfiguration

w = WorkspaceClient()

w.jobs.update(
    job_id=12345,
    new_settings={
        "trigger": {
            "file_arrival": {
                "url": "abfss://taxidata@taxidatadls.dfs.core.windows.net/raw/",
                "min_time_between_triggers_seconds": 900,  # 15 minutes
                "wait_after_last_change_seconds": 60
            }
        }
    }
)

22. Git Integration with Databricks

22.1 Git Configuration in Databricks

sequenceDiagram
    participant Dev as Developer
    participant DB as Databricks Workspace
    participant GH as GitHub Repository

    Dev->>DB: Settings > Linked Accounts
    Dev->>DB: Configure GitHub Token
    DB->>GH: Authorize Databricks App
    
    Dev->>DB: Workspace > Create > Git Folder
    DB->>GH: Clone TaxiRides repository
    GH-->>DB: Synchronized code
    
    Dev->>DB: Modify notebook
    DB->>GH: git commit + push (via Databricks UI)
    GH-->DB: Confirmation

22.2 Git Best Practices with Databricks

# Recommended repository structure for Databricks
TaxiRides/
├── notebooks/
│   ├── ETL Production/
│   │   ├── Dimensions/
│   │   │   ├── ETL_Taxi_Zones.py
│   │   │   ├── ETL_Rate_Codes.py
│   │   │   └── ETL_Payment_Types.py
│   │   └── Facts/
│   │       ├── ETL_Yellow_Taxis.py
│   │       └── ETL_Green_Taxis.py
│   └── Utils/
│       ├── data_quality_utils.py
│       └── spark_utils.py
├── tests/
│   ├── test_yellow_taxi_etl.py
│   └── test_data_quality.py
├── terraform/
│   ├── main.tf
│   └── variables.tf
└── README.md

23. Orchestration with Azure Data Factory

23.1 Azure Data Factory Components

graph LR
    subgraph "Azure Data Factory"
        Pipeline[Pipeline\nData workflow]
        Activity[Activities\nActions in the pipeline]
        Dataset[Datasets\nSource/sink data]
        LinkedSvc[Linked Services\nConnections to sources]
        Trigger[Triggers\nPipeline scheduling]
        
        Pipeline --> Activity
        Activity --> Dataset
        Dataset --> LinkedSvc
        Trigger --> Pipeline
    end
    
    subgraph "Activity Types"
        Copy[Copy Activity\nData ingestion]
        Notebook[Notebook Activity\nDatabricks notebook]
        Job[Job Activity\nDatabricks job]
        If[If Condition\nControl flow]
        ForEach[ForEach\nIteration]
        Lookup[Lookup\nData reading]
    end
    
    Activity --> Copy & Notebook & Job & If & ForEach & Lookup

23.2 ADF Linked Services

// Linked Service for Azure SQL Database
{
  "name": "AzureSqlLinkedService",
  "type": "AzureSqlDatabase",
  "typeProperties": {
    "server": "pstaxiserver.database.windows.net",
    "database": "TaxisDB",
    "encrypt": true,
    "trustServerCertificate": false,
    "authenticationType": "SQL",
    "userName": "sqladmin",
    "password": {
      "type": "AzureKeyVaultSecret",
      "store": {
        "referenceName": "MyKeyVault",
        "type": "LinkedServiceReference"
      },
      "secretName": "sql-password"
    }
  }
}

// Linked Service for Azure Data Lake Gen2
{
  "name": "DataLakeLinkedService",
  "type": "AzureBlobFS",
  "typeProperties": {
    "url": "https://taxidatadls.dfs.core.windows.net",
    "servicePrincipalId": "@{linkedService().servicePrincipalId}",
    "servicePrincipalKey": {
      "type": "AzureKeyVaultSecret",
      "store": {
        "referenceName": "MyKeyVault",
        "type": "LinkedServiceReference"
      },
      "secretName": "sp-client-secret"
    },
    "tenant": "your-tenant-id"
  }
}

24. Invoking Databricks from Data Factory

24.1 Databricks Notebook Activity

// Activity to invoke a Databricks notebook from ADF
{
  "name": "Process_RateCodes",
  "type": "DatabricksNotebook",
  "linkedServiceName": {
    "referenceName": "DatabricksLinkedService",
    "type": "LinkedServiceReference"
  },
  "typeProperties": {
    "notebookPath": "/ETL Production/Dimensions/ETL_Rate_Codes",
    "baseParameters": {
      "ProcessMonth": {
        "value": "@formatDateTime(pipeline().parameters.ProcessMonth, 'yyyy-MM')",
        "type": "Expression"
      },
      "Environment": {
        "value": "@pipeline().parameters.Environment",
        "type": "Expression"
      }
    }
  }
}

24.2 Complete ADF Pipeline with Databricks

// ADF Pipeline to orchestrate the complete ETL
{
  "name": "TaxiETLPipeline",
  "properties": {
    "parameters": {
      "ProcessMonth": {"type": "string", "defaultValue": "2023-12"},
      "Environment": {"type": "string", "defaultValue": "prod"}
    },
    "activities": [
      {
        "name": "Copy_RateCodes_From_SQL",
        "type": "Copy",
        "inputs": [{"referenceName": "AzureSqlRateCodesDataset", "type": "DatasetReference"}],
        "outputs": [{"referenceName": "AdlsRawRateCodesDataset", "type": "DatasetReference"}]
      },
      {
        "name": "Process_Dimensions",
        "type": "DatabricksNotebook",
        "dependsOn": [{"activity": "Copy_RateCodes_From_SQL", "dependencyConditions": ["Succeeded"]}],
        "typeProperties": {
          "notebookPath": "/ETL Production/Dimensions/ETL_Rate_Codes",
          "baseParameters": {
            "ProcessMonth": {"value": "@pipeline().parameters.ProcessMonth", "type": "Expression"}
          }
        }
      },
      {
        "name": "Process_Yellow_Taxis",
        "type": "DatabricksNotebook",
        "dependsOn": [{"activity": "Process_Dimensions", "dependencyConditions": ["Succeeded"]}],
        "typeProperties": {
          "notebookPath": "/ETL Production/Facts/ETL_Yellow_Taxis",
          "baseParameters": {
            "ProcessMonth": {"value": "@pipeline().parameters.ProcessMonth", "type": "Expression"},
            "Environment": {"value": "@pipeline().parameters.Environment", "type": "Expression"}
          }
        }
      }
    ]
  }
}

25. Automating ADF Pipelines with Triggers

25.1 Scheduled Trigger

// ADF Schedule Trigger
{
  "name": "DailyETLTrigger",
  "type": "ScheduleTrigger",
  "properties": {
    "recurrence": {
      "frequency": "Day",
      "interval": 1,
      "startTime": "2024-01-01T02:00:00Z",
      "timeZone": "Eastern Standard Time",
      "schedule": {
        "hours": [2],
        "minutes": [0]
      }
    },
    "pipelines": [
      {
        "pipelineReference": {"referenceName": "TaxiETLPipeline"},
        "parameters": {
          "ProcessMonth": {"value": "@formatDateTime(trigger().scheduledTime, 'yyyy-MM')"},
          "Environment": {"value": "prod"}
        }
      }
    ]
  }
}

25.2 Storage Event Trigger

// Storage Event Trigger — triggered by file arrival
{
  "name": "FileArrivalTrigger",
  "type": "BlobEventsTrigger",
  "properties": {
    "blobPathBeginsWith": "/taxidata/blobs/raw/YellowTaxis_",
    "blobPathEndsWith": ".csv",
    "events": ["Microsoft.Storage.BlobCreated"],
    "scope": "/subscriptions/{sub-id}/resourceGroups/ETL-RG/providers/Microsoft.Storage/storageAccounts/taxidatadls",
    "pipelines": [
      {
        "pipelineReference": {"referenceName": "TaxiETLPipeline"},
        "parameters": {
          "ProcessMonth": {
            "value": "@replace(replace(trigger().outputs.body.fileName, 'YellowTaxis_', ''), '.csv', '')"
          }
        }
      }
    ]
  }
}

26. Advanced ETL Architecture Patterns

26.1 Medallion Architecture (Bronze / Silver / Gold)

graph LR
    subgraph "Source"
        Raw[CSV Files\nSQL DB\nAPIs]
    end
    
    subgraph "Bronze Layer — Raw Data"
        B[Delta Table Bronze\n• Raw data\n• No transformation\n• All columns\n• Partitioned by ingestion date]
    end
    
    subgraph "Silver Layer — Cleaned Data"
        S[Delta Table Silver\n• Cleaned data\n• Correct types\n• Standardized columns\n• Invalid data filtered]
    end
    
    subgraph "Gold Layer — Business Data"
        G1[Gold Table: Revenue by Vendor]
        G2[Gold Table: Top Zones]
        G3[Aggregated Report]
    end
    
    Raw -->|Raw ingestion| B
    B -->|Cleaning + Validation| S
    S -->|Business aggregation| G1 & G2 & G3

26.2 Medallion Pipeline in Code

# ===== BRONZE LAYER =====
def ingest_to_bronze(source_path: str, target_table: str, process_month: str):
    """Raw ingestion without transformation — Bronze layer."""
    df = spark.read.csv(source_path, header=True)
    
    # Add ingestion metadata
    df_bronze = df \
        .withColumn("ingestion_date", F.current_date()) \
        .withColumn("ingestion_timestamp", F.current_timestamp()) \
        .withColumn("source_file", F.lit(source_path)) \
        .withColumn("process_month", F.lit(process_month))
    
    # Append write
    df_bronze.write \
        .mode("append") \
        .partitionBy("process_month") \
        .format("delta") \
        .saveAsTable(f"taxicatalog.bronze.{target_table}")
    
    return df_bronze.count()

# ===== SILVER LAYER =====
def transform_to_silver(source_table: str, target_table: str, process_month: str):
    """Cleaning and standardization — Silver layer."""
    df_bronze = spark.sql(f"""
        SELECT * FROM taxicatalog.bronze.{source_table}
        WHERE process_month = '{process_month}'
    """)
    
    df_silver = clean_taxi_data(df_bronze)
    df_silver = transform_taxi_data(df_silver)
    
    df_silver.write \
        .mode("append") \
        .partitionBy("VendorID", "PickupYear", "PickupMonth") \
        .format("delta") \
        .saveAsTable(f"taxicatalog.silver.{target_table}")
    
    return df_silver.count()

# ===== GOLD LAYER =====
def aggregate_to_gold(source_table: str):
    """Business aggregations — Gold layer."""
    
    # Revenue by vendor and zone
    revenue_df = spark.sql(f"""
        SELECT 
            VendorID,
            PickupYear,
            PickupMonth,
            PickupLocationId,
            COUNT(*) AS NumTrips,
            SUM(TotalAmount) AS TotalRevenue,
            AVG(TipPercentage) AS AvgTipPercent
        FROM taxicatalog.silver.{source_table}
        GROUP BY VendorID, PickupYear, PickupMonth, PickupLocationId
    """)
    
    revenue_df.write \
        .mode("overwrite") \
        .option("replaceWhere", f"PickupYear = {current_year} AND PickupMonth = {current_month}") \
        .format("delta") \
        .saveAsTable("taxicatalog.gold.revenue_by_vendor_zone")

# Orchestration in a notebook
process_month = dbutils.widgets.get("ProcessMonth")

print(f"=== Medallion Pipeline — {process_month} ===")
bronze_count = ingest_to_bronze(
    f"abfss://taxidata@storage.dfs.core.windows.net/raw/YellowTaxis_{process_month}.csv",
    "yellow_taxis_raw",
    process_month
)
print(f"Bronze: {bronze_count:,} rows ingested")

silver_count = transform_to_silver("yellow_taxis_raw", "yellow_taxis", process_month)
print(f"Silver: {silver_count:,} rows transformed")

aggregate_to_gold("yellow_taxis")
print("Gold: Aggregations recalculated")

27. Summary and Best Practices

27.1 Quality ETL Pipeline Checklist

mindmap
  root((ETL Best\nPractices))
    Code Quality
      Notebooks parameterized with Widgets
      Reusable functions in Utils
      Unit tests with pytest
      Mandatory Git integration
    Performance
      Manually defined schemas
      Well-chosen partitioning
      Regular OPTIMIZE + ZORDER
      AQE enabled
    Reliability
      Medallion Architecture
      Explicit error handling
      badRecordsPath configured
      Job monitoring
    Cost
      Job Clusters in production
      Serverless for short jobs
      Autotermination on all clusters
      Mandatory cost tags
    Security
      Unity Catalog enabled
      Service Principals for jobs
      Secrets in Azure Key Vault
      Principle of least privilege

27.2 Anti-Patterns to Avoid

28. Glossary

What is ETL?

• ETL = Extract, Transform, Load.
• Extract data from sources (customer databases, files, NoSQL).
• Apply business transformations (clean, combine, enrich).
• Load into a target repository (Data Lake, Data Warehouse).

Apache Spark Architecture

Driver (JVM)
  ↓ distributes the work
Executor 1 | Executor 2 | Executor 3 | ...
(JVM on different cluster machines)

• Driver: analyzes the code, determines how to execute it, distributes to executors.
• Executors: actually execute the code, return results.
• Scalable and fault-tolerant: can run on hundreds of machines.
• Supported languages: Scala, Python (PySpark), R, SQL.
• Use cases: batch processing, stream processing, ML, advanced analytics.

Azure Databricks

Unified platform for:

• Batch processing.
• Stream processing.
• Data Analytics.
• Machine Learning and AI.

Components

Databricks vs Azure Data Factory (ETL)

Unity Catalog

Problem Without Unity Catalog

• Each workspace is isolated.
• Separate management of users, metadata, governance.
• Difficult to share assets between workspaces.

Solution: Unity Catalog

• Shared layer above all workspaces.
• Shared Metastore: tables accessible by multiple workspaces.
• Centralized user management.
• Centralized governance, security, audit.

Unity Catalog Hierarchy

Metastore (1 per Azure region)
  └── Catalog (e.g. taxicatalog)
        └── Schema (e.g. rides)
              └── Tables / Views / Functions

Module 2 – Transforming Data with PySpark

Connecting to Azure Data Lake from Databricks

Without Unity Catalog (legacy method)

# With Access Key
storage_account = "mystorageaccount"
container = "taxidata"
access_key = "<key>"

spark.conf.set(
    f"fs.azure.account.key.{storage_account}.dfs.core.windows.net",
    access_key
)

# Read a CSV
df = spark.read.csv(f"abfss://{container}@{storage_account}.dfs.core.windows.net/raw/data.csv")

With Unity Catalog

• Define Credentials (Databricks Access Connector = Managed Identity).
• Create External Locations pointing to the Data Lake.
• Use Unity Catalog path in the code.

Apache Spark DataFrames

CSV/JSON/Parquet file in storage
  ↓ spark.read.csv/json/parquet(path)
DataFrame (tabular structure in memory)
  ↓ transformations (.filter, .select, .groupBy...)
New DataFrame
  ↓ .write.parquet/delta/saveAsTable(...)
Output to storage or table

Create a DataFrame

# Read a CSV with schema inference
df = spark.read.option("header", "true").option("inferSchema", "true").csv(file_path)

# Display data
df.show(5)
df.display()  # Databricks interface

# Display schema
df.printSchema()

# Statistics (count, mean, stddev, min, max)
df.describe("passenger_count", "trip_distance").show()

Define a manual schema (recommended in production)

from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType, TimestampType

schema = StructType([
    StructField("VendorID", IntegerType(), True),
    StructField("pickup_datetime", TimestampType(), True),
    StructField("passenger_count", IntegerType(), True),
    StructField("trip_distance", DoubleType(), True),
])

df = spark.read.schema(schema).csv(file_path)

Benefits of manual schema: no full file scan, fail fast if schema doesn’t match.

Data Quality Checks

# 1. Remove nulls
df_clean = df.dropna(subset=["pickup_datetime", "passenger_count"])

# 2. Filter invalid values
df_clean = df_clean.filter(
    (df_clean.passenger_count > 0) & (df_clean.passenger_count <= 9)
)

# 3. Remove duplicates
df_clean = df_clean.dropDuplicates()

# 4. Filter by date range
from pyspark.sql.functions import col
df_clean = df_clean.filter(col("pickup_datetime") > "2024-01-01")

PySpark Transformations

from pyspark.sql.functions import col, year, month, dayofmonth

# Select and rename columns
df_transformed = df_clean.select(
    col("VendorID"),
    col("passenger_count").cast("integer"),
    col("pickup_datetime").alias("PickupTime"),
    col("dropoff_datetime").alias("DropTime"),
    col("trip_distance").alias("TripDistance")
)

# Create derived columns
df_transformed = df_transformed \
    .withColumn("PickupYear", year(col("PickupTime"))) \
    .withColumn("PickupMonth", month(col("PickupTime"))) \
    .withColumn("PickupDay", dayofmonth(col("PickupTime")))

# Rename a column
df_transformed = df_transformed.withColumnRenamed("OldName", "NewName")

# Drop a column
df_transformed = df_transformed.drop("airport_fee")

SQL on DataFrames

# Create a temporary view (valid during the session)
df_transformed.createOrReplaceTempView("YellowTaxis")

# Execute a SQL query
result = spark.sql("SELECT VendorID, COUNT(*) as rides FROM YellowTaxis GROUP BY VendorID")
result.show()

# The result is a DataFrame

Handling Corrupted Data

# Permissive mode (default) - corrupted records → _corrupt_record column
df = spark.read.option("mode", "permissive").csv(file_path)

# Drop malformed mode - drops corrupted records
df = spark.read.option("mode", "dropMalformed").csv(file_path)

# Fail fast mode - error if a corrupted record is found
df = spark.read.option("mode", "failFast").csv(file_path)

# Databricks: store corrupted records at a path
df = spark.read.option("badRecordsPath", "/tmp/bad_records").csv(file_path)

Module 3 – Delta Lake

What is Delta Lake?

• Open-source layer above Parquet files in Data Lake.
• Brings reliability (ACID transactions) to Data Lakes.
• Pre-installed on Databricks clusters.

Comparison: Parquet vs Delta

Transaction Log (Delta)

• Each operation (insert, update, delete, optimize) → new JSON file in _delta_log/.
• Contains: commit info, metadata, files added/removed.
• Enables time travel (query past state of data).

Write in Delta

# Write a DataFrame in Delta (partitioned)
df_transformed.write \
    .mode("overwrite") \
    .partitionBy("VendorID") \
    .format("delta") \
    .save("abfss://taxidata@storage.dfs.core.windows.net/output/yellowtaxis.delta")

# Save as Delta Table (with registered metadata)
df_transformed.write \
    .mode("overwrite") \
    .partitionBy("VendorID") \
    .format("delta") \
    .saveAsTable("taxicatalog.rides.yellowtaxis")

DML on Delta Tables

-- INSERT
INSERT INTO taxicatalog.rides.yellowtaxis (VendorID, PickupTime, ...)
VALUES (3, '2024-01-15 08:00:00', ...);

-- UPDATE
UPDATE taxicatalog.rides.yellowtaxis
SET passenger_count = 2
WHERE trip_id = 123;

-- DELETE
DELETE FROM taxicatalog.rides.yellowtaxis
WHERE passenger_count = 0;

-- MERGE (upsert)
MERGE INTO target USING source ON target.id = source.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

History and Time Travel

-- View version history
DESCRIBE HISTORY taxicatalog.rides.yellowtaxis;

-- Query a past version
SELECT * FROM taxicatalog.rides.yellowtaxis VERSION AS OF 3;

-- Query by timestamp
SELECT * FROM taxicatalog.rides.yellowtaxis TIMESTAMP AS OF '2024-01-01 00:00:00';

Module 4 – Delta Performance Optimization

Bin Packing (OPTIMIZE)

Problem: many transactions create small files → degraded performance.

-- Compact small files into ~1 GB files
OPTIMIZE taxicatalog.rides.yellowtaxis;

-- With Z-ordering on a frequently filtered column
OPTIMIZE taxicatalog.rides.yellowtaxis ZORDER BY (PickupLocationId);

Z-Ordering

• Co-locates similar data in the same files.
• Reduces the number of files to read for filtered queries.
• Example: if always filtering by PickupLocationId, Z-ordering speeds up these queries.

VACUUM

• Physically removes files no longer referenced in the transaction log.
• Affects Time Travel (versions before VACUUM are no longer accessible).
• Default retention: 7 days.

-- Preview what will be deleted
VACUUM taxicatalog.rides.yellowtaxis DRY RUN;

-- Execute cleanup (keep 7 days of history)
VACUUM taxicatalog.rides.yellowtaxis RETAIN 168 HOURS;

Auto Optimization

# Enable on a table
ALTER TABLE taxicatalog.rides.yellowtaxis
SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true'
);

Module 5 – Automating Pipelines (Databricks Workflows)

Parameterizing Notebooks with Widgets

# Create a text widget
dbutils.widgets.text("ProcessMonth", "2024-01", "Month to process")

# Read the widget value
process_month = dbutils.widgets.get("ProcessMonth")

# Use in code
file_path = f"abfss://taxidata@storage.dfs.core.windows.net/raw/{process_month}/yellow_taxi.csv"

Databricks Jobs (Workflows)

Job Structure

Job
  ├── Tasks
  │     ├── Task A (Notebook) → depends on nothing
  │     ├── Task B (Notebook) → depends on A
  │     └── Task C (If/Else) → depends on A and B
  └── Triggers (schedule, file arrival, continuous)

Task Types

Compute Options for Tasks

Task Values (passing data between tasks)

# In the emitting task (TaxiZones notebook)
count = yellowtaxis_df.count()
dbutils.jobs.taskValues.set(key="TaxiZonesCount", value=count)

# In the If/Else task: condition = {{tasks.Dimension_Taxi_Zones.values.TaxiZonesCount}} > 100

Trigger Types

Git Integration with Databricks

Databricks Workspace → Settings → Linked Accounts → GitHub → Authorize

Databricks → Workspace → Create → Git Folder
→ Paste GitHub repo URL
→ Folder connected to repo
→ Push/Pull notebooks via Databricks interface

Module 6 – Orchestration with Azure Data Factory

ADF Components

Activity Types

Invoke Databricks from ADF

1. Create a Databricks Linked Service (with Personal Access Token).
2. Add a Databricks Notebook activity in the pipeline.
3. Configure: workspace, cluster (Job Cluster recommended), notebook path, parameters.

⚠️ Limitation: Databricks activities in ADF don’t yet support serverless compute. Each activity uses its own Job Cluster (not shared).

ADF Trigger Types

General Best Practices

• Manual schema in production (not inferSchema) → fail fast, better performance.
• Delta format by default for all data in Databricks.
• Run OPTIMIZE regularly on tables with many DML operations.
• Define a VACUUM retention adapted to your time travel needs.
• Use Unity Catalog for governance and multi-workspace sharing.
• Use Job Clusters (not All-Purpose) for production workloads.
• Widgets to parameterize notebooks and make them reusable.
• Integrate Git (GitHub/Azure Repos) to version notebooks.

Module 7 – ETL Architecture with Databricks: Medallion Architecture

Medallion Architecture Concept

The Medallion Architecture is a Lakehouse data design pattern that organizes data into progressive layers of increasing quality. Delta Lake is the foundation, guaranteeing ACID, time travel and performance at each layer.

Data Flow: Mermaid Diagram

flowchart LR
    subgraph Sources
        A1[Azure SQL DB]
        A2[CSV/JSON Files]
        A3[Kafka / Event Hub]
        A4[REST API]
    end

    subgraph Bronze["🥉 Bronze Layer (Raw)"]
        B1["(Delta Table\nbronze.rides_raw)"]
        B2["(Delta Table\nbronze.customers_raw)"]
    end

    subgraph Silver["🥈 Silver Layer (Cleaned)"]
        C1["(Delta Table\nsilver.rides_cleaned)"]
        C2["(Delta Table\nsilver.customers_cleaned)"]
    end

    subgraph Gold["🥇 Gold Layer (Aggregated)"]
        D1["(Delta Table\ngold.rides_daily_summary)"]
        D2["(Delta Table\ngold.customer_lifetime_value)"]
    end

    subgraph Consumption
        E1[Power BI / Tableau]
        E2[ML Models]
        E3[Data Science]
    end

    A1 --> B1
    A2 --> B1
    A3 --> B2
    A4 --> B2
    B1 --> C1
    B2 --> C2
    C1 --> D1
    C2 --> D2
    D1 --> E1
    D2 --> E2
    D1 --> E3

When to Use Each Layer?

Bronze – Raw Ingestion

• Keep data exactly as received (no transformation).
• Useful for reprocessing: if logic changes, Silver/Gold can be rebuilt from Bronze.
• Technical columns: _ingest_timestamp, _source_file, _batch_id.
• Write mode: append (never overwrite, except in special cases).

Silver – Cleaning and Validation

• Remove duplicates, nulls, out-of-range data.
• Apply correct types (cast string → integer, timestamp).
• Enrich with dimensions (lookup joins).
• Apply basic business rules.
• Write mode: merge (upsert) or append with deduplication.

Gold – Aggregation and Business Metrics

• Dimensional models (Facts + Dimensions).
• KPIs, metrics aggregated by day/week/month.
• Optimized for BI tools (Power BI, Tableau).
• Write mode: overwrite (recalculation) or incremental merge.

Bronze → Silver → Gold Implementation

# ── BRONZE: raw ingestion ──────────────────────────────────────────────────
from pyspark.sql.functions import current_timestamp, input_file_name

df_raw = spark.read \
    .option("header", "true") \
    .csv("abfss://raw@storage.dfs.core.windows.net/rides/*.csv")

df_bronze = df_raw \
    .withColumn("_ingest_timestamp", current_timestamp()) \
    .withColumn("_source_file", input_file_name())

df_bronze.write \
    .mode("append") \
    .format("delta") \
    .saveAsTable("bronze.rides_raw")

# ── SILVER: cleaning + validation ──────────────────────────────────────────
from pyspark.sql.functions import col, to_timestamp

df_silver = spark.table("bronze.rides_raw") \
    .filter(col("passenger_count").cast("int").between(1, 9)) \
    .filter(col("trip_distance").cast("double") > 0) \
    .withColumn("pickup_datetime", to_timestamp(col("tpep_pickup_datetime"))) \
    .withColumn("passenger_count", col("passenger_count").cast("int")) \
    .dropDuplicates(["vendor_id", "pickup_datetime", "trip_distance"]) \
    .select("vendor_id", "pickup_datetime", "passenger_count", "trip_distance",
            "fare_amount", "total_amount")

df_silver.write \
    .mode("overwrite") \
    .format("delta") \
    .saveAsTable("silver.rides_cleaned")

# ── GOLD: business aggregation ──────────────────────────────────────────────
from pyspark.sql.functions import date_trunc, sum, count, avg, round

df_gold = spark.table("silver.rides_cleaned") \
    .withColumn("ride_date", date_trunc("day", col("pickup_datetime"))) \
    .groupBy("ride_date", "vendor_id") \
    .agg(
        count("*").alias("total_rides"),
        sum("fare_amount").alias("total_fare"),
        avg("trip_distance").alias("avg_distance"),
        avg("passenger_count").alias("avg_passengers")
    ) \
    .withColumn("total_fare", round(col("total_fare"), 2))

df_gold.write \
    .mode("overwrite") \
    .format("delta") \
    .saveAsTable("gold.rides_daily_summary")

Delta Lake as Foundation

Delta Lake guarantees at each layer:

• ACID transactions: no reading of partially written data.
• Schema enforcement: automatic rejection of data not conforming to the schema.
• Time Travel: ability to replay transformations from a previous version.
• Scalability: Parquet + Photon Engine for ultra-fast reads.

Search Terms

etl · pipelines · azure · databricks · data · factory · spark · engineering · analytics · delta · architecture · catalog · lake · unity · trigger · types · adf · apache · tasks · components · connection · dataframes · git · medallion