databricks-data-engineering

# 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: ```python 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: ```python 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:** ```python # 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:** ```python # 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 ----------