databricks-data-engineering

name	databricks-data-engineering
description	Production data engineering pipelines following medallion architecture (Bronze/Silver/Gold layers) with data ingestion, transformation, quality checks, Delta Lake optimization, and orchestration. Use when building ETL pipelines, medallion architecture, data lakes, or data transformation workflows.
allowed-tools	["Bash","Read","Write","Edit","Grep","Glob"]
model	claude-sonnet-4-5-20250929
user-invocable	true

Databricks Data Engineering Pipelines

Build production-grade data pipelines following medallion architecture (Bronze/Silver/Gold) with data quality checks, Delta Lake optimization, and multi-layer transformations.

When to Use This Skill

• Building ETL/ELT data pipelines
• Implementing medallion architecture
• Data lake transformations
• Data quality and validation workflows
• Incremental data processing
• Batch data pipelines
• Real-time streaming (structured streaming)

Medallion Architecture

The medallion architecture organizes data into three layers of increasing quality:

Bronze Layer (Raw/Landing)
    ↓
Silver Layer (Cleaned/Validated)
    ↓
Gold Layer (Business/Aggregated)

Bronze Layer (Raw Ingestion)

• Purpose: Ingest raw data with minimal transformation
• Pattern: Append-only, preserve source format
• Transformations: Type casting, timestamp addition
• Storage: Delta Lake tables
• Schema: Flexible, can evolve

Silver Layer (Cleaned Data)

• Purpose: Cleaned, validated, and conformed data
• Pattern: Deduplication, quality checks, schema enforcement
• Transformations: Data quality rules, null handling, type validation
• Storage: Delta Lake tables (optimized)
• Schema: Strict, well-defined

Gold Layer (Business-Ready)

• Purpose: Aggregated, business-ready datasets
• Pattern: Joins, aggregations, business logic
• Transformations: Metrics, KPIs, analytics-ready views
• Storage: Delta Lake tables (highly optimized)
• Schema: Denormalized for analytics

Complete Data Pipeline Workflow

Phase 1: Schema Setup

Use databricks-unity-catalog skill to create medallion schemas:

catalog = "de_prod"

# Bronze - raw data
create_schema(
    catalog_name=catalog,
    schema_name="bronze",
    comment="Raw ingested data. Minimal transformation. Append-only. 90-day retention."
)

# Silver - cleaned data
create_schema(
    catalog_name=catalog,
    schema_name="silver",
    comment="Cleaned and validated data. Deduplicated, quality-checked. 1-year retention."
)

# Gold - business aggregates
create_schema(
    catalog_name=catalog,
    schema_name="gold",
    comment="Business-ready aggregates. Optimized for analytics. 3-year retention."
)

Phase 2: Bronze Layer Development

Use databricks-testing skill to test ingestion logic:

databricks_command(
    cluster_id="0123-456789-abc123",
    language="python",
    code="""
from pyspark.sql import functions as F

# Ingest raw data
raw_df = (
    spark.read
    .format("json")  # or csv, parquet, etc.
    .option("inferSchema", "true")
    .load("/mnt/source/events/*.json")
)

# Add ingestion metadata
bronze_df = (
    raw_df
    .withColumn("ingestion_timestamp", F.current_timestamp())
    .withColumn("ingestion_date", F.current_date())
    .withColumn("source_file", F.input_file_name())
)

print(f"Ingested {bronze_df.count()} records")

# Save to bronze
bronze_df.write \\
    .format("delta") \\
    .mode("append") \\
    .saveAsTable("de_prod.bronze.raw_events")

print("Bronze ingestion complete")
"""
)

Phase 3: Silver Layer Development

Use databricks-testing skill to test cleaning logic:

Complete Silver Layer Notebook:

# Databricks notebook source
# MAGIC %md
# MAGIC # Silver Layer - Data Cleaning & Validation
# MAGIC
# MAGIC Transforms bronze data into cleaned, validated silver tables

# COMMAND ----------
# Widget parameterization
try:
    catalog = dbutils.widgets.get("catalog")
except:
    catalog = "de_dev"

try:
    bronze_schema = dbutils.widgets.get("bronze_schema")
except:
    bronze_schema = "bronze"

try:
    silver_schema = dbutils.widgets.get("silver_schema")
except:
    silver_schema = "silver"

try:
    batch_date = dbutils.widgets.get("batch_date")
except:
    from datetime import date
    batch_date = str(date.today())

print(f"Processing silver layer:")
print(f"  Catalog: {catalog}")
print(f"  Bronze schema: {bronze_schema}")
print(f"  Silver schema: {silver_schema}")
print(f"  Batch date: {batch_date}")

# COMMAND ----------
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Read from bronze (incremental)
bronze_df = spark.read.table(f"{catalog}.{bronze_schema}.raw_events") \\
    .filter(F.col("ingestion_date") == batch_date)

print(f"Bronze records for {batch_date}: {bronze_df.count()}")

# COMMAND ----------
# Data quality checks BEFORE cleaning
print("Data Quality Report - Before Cleaning:")
print(f"  Total records: {bronze_df.count()}")
print(f"  Null event_id: {bronze_df.filter(F.col('event_id').isNull()).count()}")
print(f"  Null timestamp: {bronze_df.filter(F.col('timestamp').isNull()).count()}")
print(f"  Duplicate event_id: {bronze_df.groupBy('event_id').count().filter(F.col('count') > 1).count()}")

# Record initial count for tracking
initial_count = bronze_df.count()

# COMMAND ----------
# Cleaning transformations
silver_df = (
    bronze_df

    # Remove records with null in critical fields
    .filter(F.col("event_id").isNotNull())
    .filter(F.col("timestamp").isNotNull())
    .filter(F.col("user_id").isNotNull())

    # Deduplicate - keep latest by timestamp
    .withColumn("row_num", F.row_number().over(
        Window.partitionBy("event_id").orderBy(F.col("timestamp").desc())
    ))
    .filter(F.col("row_num") == 1)
    .drop("row_num")

    # Type casting and validation
    .withColumn("amount", F.col("amount").cast("double"))
    .withColumn("quantity", F.col("quantity").cast("int"))
    .withColumn("timestamp", F.col("timestamp").cast("timestamp"))

    # Remove invalid values
    .filter(F.col("amount") >= 0)  # No negative amounts
    .filter(F.col("quantity") > 0)  # Quantity must be positive

    # Add silver layer metadata
    .withColumn("silver_processed_at", F.current_timestamp())
    .withColumn("silver_processing_date", F.current_date())

    # Drop ingestion metadata (not needed in silver)
    .drop("source_file", "ingestion_timestamp", "ingestion_date")
)

# COMMAND ----------
# Data quality checks AFTER cleaning
final_count = silver_df.count()
removed_count = initial_count - final_count

print(f"\\nData Quality Report - After Cleaning:")
print(f"  Clean records: {final_count}")
print(f"  Removed records: {removed_count} ({removed_count/initial_count*100:.2f}%)")

# Validation assertions
assert silver_df.filter(F.col("event_id").isNull()).count() == 0, "Null event_ids remain"
assert silver_df.filter(F.col("timestamp").isNull()).count() == 0, "Null timestamps remain"
assert silver_df.groupBy("event_id").count().filter(F.col("count") > 1).count() == 0, "Duplicates remain"

print("✓ All quality checks passed")

# COMMAND ----------
# Write to silver (merge for idempotency)
from delta.tables import DeltaTable

silver_table = f"{catalog}.{silver_schema}.clean_events"

# Check if table exists
if spark.catalog.tableExists(silver_table):
    # Merge into existing table (upsert pattern)
    deltaTable = DeltaTable.forName(spark, silver_table)

    deltaTable.alias("target").merge(
        silver_df.alias("source"),
        "target.event_id = source.event_id"
    ).whenMatchedUpdateAll() \\
     .whenNotMatchedInsertAll() \\
     .execute()

    print(f"Merged {final_count} records into {silver_table}")
else:
    # Create table
    silver_df.write \\
        .format("delta") \\
        .mode("overwrite") \\
        .saveAsTable(silver_table)

    print(f"Created {silver_table} with {final_count} records")

# COMMAND ----------
# Optimize Delta table
spark.sql(f"OPTIMIZE {silver_table}")
print(f"✓ Optimized {silver_table}")

# Update statistics
spark.sql(f"ANALYZE TABLE {silver_table} COMPUTE STATISTICS")
print(f"✓ Updated statistics for {silver_table}")

# COMMAND ----------
# Z-order for query optimization (optional)
spark.sql(f"OPTIMIZE {silver_table} ZORDER BY (event_id, timestamp)")
print(f"✓ Z-ordered {silver_table}")

# COMMAND ----------
print(f"\\n{'='*50}")
print("Silver layer processing complete!")
print(f"  Input (Bronze): {initial_count} records")
print(f"  Output (Silver): {final_count} records")
print(f"  Quality: {final_count/initial_count*100:.2f}% retention")
print(f"{'='*50}")

Phase 4: Gold Layer Development

Use databricks-testing skill to test aggregation logic:

Complete Gold Layer Notebook:

# Databricks notebook source
# MAGIC %md
# MAGIC # Gold Layer - Business Aggregates
# MAGIC
# MAGIC Creates business-ready aggregated datasets for analytics

# COMMAND ----------
# Widget parameterization
try:
    catalog = dbutils.widgets.get("catalog")
except:
    catalog = "de_dev"

try:
    silver_schema = dbutils.widgets.get("silver_schema")
except:
    silver_schema = "silver"

try:
    gold_schema = dbutils.widgets.get("gold_schema")
except:
    gold_schema = "gold"

print(f"Processing gold layer:")
print(f"  Catalog: {catalog}")
print(f"  Silver schema: {silver_schema}")
print(f"  Gold schema: {gold_schema}")

# COMMAND ----------
from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Read from silver
silver_df = spark.table(f"{catalog}.{silver_schema}.clean_events")
print(f"Silver records: {silver_df.count()}")

# COMMAND ----------
# Daily aggregations by user
daily_user_metrics = (
    silver_df
    .withColumn("event_date", F.to_date("timestamp"))
    .groupBy("event_date", "user_id")
    .agg(
        F.count("*").alias("event_count"),
        F.sum("amount").alias("total_amount"),
        F.avg("amount").alias("avg_amount"),
        F.sum("quantity").alias("total_quantity"),
        F.countDistinct("event_id").alias("unique_events"),
        F.min("timestamp").alias("first_event_time"),
        F.max("timestamp").alias("last_event_time")
    )
    .withColumn("gold_created_at", F.current_timestamp())
)

print(f"Daily user metrics: {daily_user_metrics.count()} records")

# COMMAND ----------