data-engineering-patterns-fabric-databricks

# Data Engineering Patterns - Fabric & Databricks > Skill by [ara.so](https://ara.so) — Data Skills collection. This skill provides access to 600+ field-tested data engineering patterns for Microsoft Fabric, Azure Databricks, and PySpark. These patterns cover everything from pipeline design and Delta Lake optimization to Unity Catalog governance and cost architecture. ## What This Project Provides A comprehensive collection of patterns organized into 12 books covering: **Microsoft Fabric (250 patterns):** - Pipelines and Data Factory - Lakehouse and PySpark - Warehouse and SQL - Power BI in Fabric - Architecture Patterns **Azure Databricks (350 patterns):** - Clusters and Compute - Delta Lake - Workflows and Orchestration - Structured Streaming and Auto Loader - Unity Catalog - Databricks SQL and Photon - Platform and Cost Architecture **PySpark:** - 88 concepts for production Spark across both platforms ## Installation Clone the repository to access all pattern PDFs: ```bash git clone https://github.com/ssanjaychandra123/data-engineering-patterns.git cd data-engineering-patterns ``` ## Repository Structure ``` data-engineering-patterns/ ├── Fabric Patterns/ │ ├── Fabric Engineering Patterns Book I - Pipelines and Data Factory.pdf │ ├── Fabric Engineering Patterns Book II - Lakehouse and PySpark.pdf │ ├── Fabric Engineering Patterns Book III - Warehouse and SQL.pdf │ ├── Fabric Engineering Patterns Book IV - Power BI in Fabric.pdf │ └── Fabric Engineering Patterns Book V - Architecture Patterns.pdf ├── Databricks Patterns/ │ ├── Azure Databricks Engineering Patterns Book I - Clusters and Compute.pdf │ ├── Azure Databricks Engineering Patterns Book II - Delta Lake.pdf │ ├── Azure Databricks Engineering Patterns Book III - Workflows and Orchestration.pdf │ ├── Azure Databricks Engineering Patterns Book IV - Structured Streaming and Auto Loader.pdf │ ├── Azure Databricks Engineering Patterns Book V - Unity Catalog.pdf │ ├── Azure Databricks Engineering Patterns Book VI - Databricks SQL and Photon.pdf │ └── Azure Databricks Engineering Patterns Book VII - Platform and Cost Architecture.pdf └── PySpark/ └── The PySpark Handbook for Fabric and Databricks.pdf ``` ## Key Pattern Categories ### Microsoft Fabric Patterns #### Pipeline and Data Factory Patterns Common patterns include: - Incremental data loading strategies - Pipeline retry and error handling - Parameter-driven pipeline design - Activity dependencies and control flow - Copy activity optimization - Metadata-driven frameworks Example incremental load pattern in Fabric Pipeline: ```python # Notebook activity in Fabric pipeline from datetime import datetime, timedelta # Get pipeline parameters watermark = spark.conf.get("pipeline.watermark") table_name = spark.conf.get("pipeline.tableName") # Read incremental data df = spark.read.format("delta") \ .load(f"abfss://source@storage.dfs.core.windows.net/{table_name}") \ .filter(f"modified_date > '{watermark}'") # Write to target df.write.format("delta") \ .mode("append") \ .option("mergeSchema", "true") \ .save(f"Tables/{table_name}") # Return new watermark new_watermark = df.agg({"modified_date": "max"}).collect()[0][0] mssparkutils.notebook.exit(str(new_watermark)) ``` #### Lakehouse and PySpark Patterns Key patterns for Fabric Lakehouse: ```python # Pattern: Upsert (merge) operation in Fabric Lakehouse from delta.tables import DeltaTable # Source data updates_df = spark.read.format("parquet").load("Files/updates/") # Target Delta table target_table = DeltaTable.forPath(spark, "Tables/customers") # Merge logic target_table.alias("target").merge( updates_df.alias("updates"), "target.customer_id = updates.customer_id" ).whenMatchedUpdate(set={ "name": "updates.name", "email": "updates.email", "updated_at": "updates.updated_at" }).whenNotMatchedInsert(values={ "customer_id": "updates.customer_id", "name": "updates.name", "email": "updates.email", "created_at": "updates.created_at", "updated_at": "updates.updated_at" }).execute() ``` Pattern: Optimize Delta tables in Fabric: ```python # Optimize with Z-ordering for common query patterns spark.sql(f""" OPTIMIZE lakehouse.customers ZORDER BY (customer_id, signup_date) """) # Vacuum old files (default 7 days retention) spark.sql(f""" VACUUM lakehouse.customers RETAIN 168 HOURS """) ``` #### Warehouse and SQL Patterns Pattern: Create warehouse tables with proper partitioning: ```sql -- Create partitioned warehouse table in Fabric CREATE TABLE dw.fact_sales ( sale_id BIGINT, customer_id BIGINT, product_id BIGINT, sale_amount DECIMAL(18,2), sale_date DATE, created_at TIMESTAMP ) USING DELTA PARTITIONED BY (sale_date); -- Insert with partition optimization INSERT INTO dw.fact_sales SELECT sale_id, customer_id, product_id, sale_amount, CAST(sale_date AS DATE) as sale_date, created_at FROM staging.sales WHERE sale_date >= CURRENT_DATE - INTERVAL 7 DAYS; ``` ### Azure Databricks Patterns #### Cluster and Compute Patterns Pattern: Configure autoscaling cluster for cost optimization: ```python # Databricks cluster configuration (JSON) { "cluster_name": "production-etl", "spark_version": "13.3.x-scala2.12", "node_type_id": "Standard_DS3_v2", "autoscale": { "min_workers": 2, "max_workers": 8 }, "autotermination_minutes": 30, "spark_conf": { "spark.databricks.delta.preview.enabled": "true", "spark.databricks.delta.properties.defaults.autoOptimize.optimizeWrite": "true", "spark.databricks.delta.properties.defaults.autoOptimize.autoCompact": "true" }, "aws_attributes": { "availability": "SPOT_WITH_FALLBACK", "spot_bid_price_percent": 100 } } ``` #### Delta Lake Patterns Pattern: Time travel and versioning: ```python # Read historical version of Delta table df_version_10 = spark.read.format("delta") \ .option("versionAsOf", 10) \ .load("/mnt/delta/customers") # Read table as of timestamp df_yesterday = spark.read.format("delta") \ .option("timestampAsOf", "2024-01-15 00:00:00") \ .load("/mnt/delta/customers") # Describe history history_df = spark.sql("DESCRIBE HISTORY delta.`/mnt/delta/customers`") history_df.select("version", "timestamp", "operation", "operationMetrics").show() ``` Pattern: Change Data Feed (CDF) for incremental processing: ```python # Enable CDF on table spark.sql(""" ALTER TABLE delta.customers SET TBLPROPERTIES (delta.enableChangeDataFeed = true) """) # Read changes between versions changes_df = spark.read.format("delta") \ .option("readChangeFeed", "true") \ .option("startingVersion", 10) \ .option("endingVersion", 20) \ .table("delta.customers") # Process different change types inserts = changes_df.filter("_change_type = 'insert'") updates = changes_df.filter("_change_type = 'update_postimage'") deletes = changes_df.filter("_change_type = 'delete'") ``` #### Structured Streaming Patterns Pattern: Auto Loader with schema evolution: ```python # Auto Loader with schema inference and evolution checkpoint_path = "/mnt/checkpoints/raw_files" target_path = "/mnt/delta/bronze/raw_data" df = spark.readStream.format("cloudFiles") \ .option("cloudFiles.format", "json") \ .option("cloudFiles.schemaLocation", checkpoint_path + "/schema") \ .option("cloudFiles.inferColumnTypes", "true") \ .option("cloudFiles.schemaEvolutionMode", "addNewColumns") \ .load("/mnt/landing/raw_files/") # Write to Delta with checkpointing query = df.writeStream \ .format("delta") \ .option("checkpointLocation", checkpoint_path) \ .option("mergeSchema", "true") \ .trigger(availableNow=True) \ .start(target_path) query.awaitTermination() ``` Pattern: Streaming aggregations with watermarking: ```python from pyspark.sql.functions import window, col # Read streaming data stream_df = spark.readStream.format("delta") \ .table("events") # Windowed aggregation with watermark aggregated = stream_df \ .withWatermark("event_time", "10 minutes") \ .groupBy( window(col("event_time"), "5 minutes"), col("user_id") ) \ .agg({ "event_id": "count", "amount": "sum" }) # Write to Delta table query = aggregated.writeStream \ .format("delta") \ .outputMode("append") \ .option("checkpointLocation", "/mnt/checkpoints/aggregations") \ .toTable("event_aggregations") ``` #### Unity Catalog Patterns Pattern: Create governed table with row-level security: ```python # Create schema with Unity Catalog spark.sql(""" CREATE SCHEMA IF NOT EXISTS main.finance COMMENT 'Finance department data' LOCATION 'abfss://data@storage.dfs.core.windows.net/finance' """) # Create managed table spark.sql(""" CREATE TABLE IF NOT EXISTS main.finance.transactions ( transaction_id BIGINT, account_id BIGINT, amount DECIMAL(18,2), region STRING, transaction_date DATE ) USING DELTA TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true') """) # Apply row filter for data access control spark.sql(""" CREATE FUNCTION main.finance.region_filter(region STRING) RETURN IF( IS_MEMBER('data_engineers'), TRUE, region = current_user() ) """) spark.sql(""" ALTER TABLE main.finance.transactions SET ROW FILTER main.finance.region_filter ON (region) """) ``` Pattern: Column masking with Unity Catalog: ```python # Create masking function spark.sql(""" CREATE FUNCTION main.finance.mask_ssn(ssn STRING) RETURN CASE WHEN IS_MEMBER('finance_managers') THEN ssn ELSE CONCAT('XXX-XX-', RIGHT(ssn, 4)) END """) # Apply column mask spark.sql(""" ALTER TABLE main.finance.customers ALTER COLUMN ssn SET MASK main.finance.mask_ssn """) ``` #### Workflows and Orchestration Patterns Pattern: Create parameterized Databricks job: ```python # In notebook: Get job parameters dbutils.widgets.text("date", "") dbutils.widgets.text("environment", "prod") processing_date = dbutils.widgets.get("date") env = dbutils.widgets.get("environment") # Use parameters in processing df = spark.read.format("delta") \ .load(f"/mnt/{env}/data") \ .filter(f"date = '{processing_date}'") # Process and write results result_df = df.groupBy("category").count() result_df.write.format("delta").mode("overwrite") \ .save(f"/mnt/{env}/results/{processing_date}") # Return status for orchestration dbutils.notebook.exit(f"Processed {result_df.count()} records") ``` Pattern: Job definition with retry logic: ```json { "name": "daily-etl-pipeline", "tasks": [ { "task_key": "extract", "notebook_task": { "notebook_path": "/Workflows/extract", "base_parameters": { "date": "{{job.start_time.date}}", "environment": "prod" } }, "existing_cluster_id": "{{cluster_id}}", "max_retries": 2, "timeout_seconds": 3600 }, { "task_key": "transform", "depends_on": [{"task_key": "extract"}], "notebook_task": { "notebook_path": "/Workflows/transform", "base_parameters": { "date": "{{job.start_time.date}}" } }, "existing_cluster_id": "{{cluster_id}}", "max_retries": 1 }, { "task_key": "load", "depends_on": [{"task_key": "transform"}], "notebook_task": { "notebook_path": "/Workflows/load" },