┌─────────────────────────────────────────────────────────────────┐
│                    DP-750 KNOWLEDGE PILLARS                     │
│                                                                 │
│  ┌─────────────────┐  ┌─────────────────┐  ┌───────────────┐  │
│  │  Azure          │  │  Unity Catalog  │  │  Apache       │  │
│  │  Databricks     │  │  Governance     │  │  Spark        │  │
│  │  Platform       │  │                 │  │  Engine       │  │
│  │                 │  │  • Catalogs     │  │               │  │
│  │  • Workspaces   │  │  • Schemas      │  │  • RDDs       │  │
│  │  • Clusters     │  │  • Tables       │  │  • DataFrames │  │
│  │  • Notebooks    │  │  • Volumes      │  │  • Spark SQL  │  │
│  │  • Delta Lake   │  │  • Lineage      │  │  • Streaming  │  │
│  │  • Lakeflow     │  │  • ABAC/RBAC    │  │  • MLlib      │  │
│  └─────────────────┘  └─────────────────┘  └───────────────┘  │
└─────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐
│                    COMPUTE DECISION TREE                     │
│                                                              │
│  Interactive work?                                           │
│    └─► All-Purpose Cluster (shared or classic)              │
│                                                              │
│  Automated pipeline?                                         │
│    └─► Job Compute (isolated, cost-optimized)               │
│                                                              │
│  SQL analytics / BI dashboards?                              │
│    └─► SQL Warehouse (serverless or pro)                    │
│                                                              │
│  Need maximum performance?                                   │
│    └─► Enable Photon Acceleration                           │
│                                                              │
│  Variable workloads?                                         │
│    └─► Enable Autoscaling + Auto Termination                │
└──────────────────────────────────────────────────────────────┘

# Create a managed Delta table
spark.sql("""
  CREATE TABLE catalog.schema.sales_data (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      DECIMAL(10,2),
    order_date  DATE
  )
  USING DELTA
  PARTITIONED BY (order_date)
""")

# Time Travel — query historical versions
spark.sql("""
  SELECT * FROM catalog.schema.sales_data
  VERSION AS OF 5
""")

# Optimize and vacuum
spark.sql("OPTIMIZE catalog.schema.sales_data ZORDER BY (customer_id)")
spark.sql("VACUUM catalog.schema.sales_data RETAIN 168 HOURS")

Unity Catalog Hierarchy
│
├── Metastore (one per region — account level)
│   │
│   ├── Catalog  (top-level container — e.g., "prod", "dev", "raw")
│   │   │
│   │   ├── Schema  (logical grouping — e.g., "sales", "finance")
│   │   │   │
│   │   │   ├── Tables (managed or external)
│   │   │   ├── Views
│   │   │   ├── Materialized Views
│   │   │   ├── Volumes (unstructured file storage)
│   │   │   └── Functions
│   │   │
│   │   └── Schema ...
│   │
│   └── Catalog ...
│
└── External Locations (S3, ADLS Gen2, GCS)

-- Grant read access to a group
GRANT SELECT ON TABLE catalog.schema.customers TO `data-analysts`;

-- Grant write access to a service principal
GRANT MODIFY ON SCHEMA catalog.schema TO `etl-service-principal`;

-- Grant full catalog access to a team
GRANT USE CATALOG, USE SCHEMA, SELECT, MODIFY
  ON CATALOG prod_catalog
  TO `data-engineering-team`;

-- Revoke access
REVOKE SELECT ON TABLE catalog.schema.pii_data FROM `external-user`;

-- Row filter: users only see their own region's data
CREATE FUNCTION catalog.schema.region_filter(region_col STRING)
  RETURN IF(
    IS_ACCOUNT_GROUP_MEMBER('admin'),
    TRUE,
    region_col = CURRENT_USER()
  );

ALTER TABLE catalog.schema.sales_data
  SET ROW FILTER catalog.schema.region_filter ON (region);

-- Column mask: hide PII for non-privileged users
CREATE FUNCTION catalog.schema.mask_email(email STRING)
  RETURN IF(
    IS_ACCOUNT_GROUP_MEMBER('pii-access'),
    email,
    CONCAT(LEFT(email, 2), '***@***.com')
  );

ALTER TABLE catalog.schema.customers
  ALTER COLUMN email
  SET MASK catalog.schema.mask_email;

Data Lineage Flow Example:
raw_orders (Bronze)
    │
    ▼ (notebook: transform_orders.py)
cleaned_orders (Silver)
    │
    ▼ (DLT pipeline: aggregate_pipeline)
daily_revenue (Gold)
    │
    ▼ (Power BI / SQL Warehouse)
Executive Dashboard

# Share data with an external organization — no data movement
spark.sql("""
  CREATE SHARE partner_share
  COMMENT 'Shared dataset for Partner Corp'
""")

spark.sql("""
  ALTER SHARE partner_share
  ADD TABLE catalog.schema.public_products
""")

spark.sql("""
  CREATE RECIPIENT partner_corp
  COMMENT 'External partner access'
""")

spark.sql("""
  GRANT SELECT ON SHARE partner_share TO RECIPIENT partner_corp
""")

Spark Cluster Architecture
│
├── Driver Node (Master)
│   ├── SparkSession / SparkContext
│   ├── DAG Scheduler
│   ├── Task Scheduler
│   └── Spark UI (port 4040)
│
└── Worker Nodes (Executors)
    ├── Executor 1
    │   ├── Task A  ← processes partition 1
    │   ├── Task B  ← processes partition 2
    │   └── Cache (in-memory storage)
    │
    ├── Executor 2
    │   ├── Task C  ← processes partition 3
    │   └── Task D  ← processes partition 4
    │
    └── Executor N ...

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, avg, count

spark = SparkSession.builder.appName("DP750-Study").getOrCreate()

# TRANSFORMATIONS — lazy, build the DAG, nothing executes yet
df = spark.read.format("delta").load("/mnt/data/sales")

filtered = df.filter(col("amount") > 100)           # lazy
grouped  = filtered.groupBy("region")               # lazy
result   = grouped.agg(
    sum("amount").alias("total_sales"),
    avg("amount").alias("avg_sale"),
    count("*").alias("num_orders")
)                                                    # lazy

# ACTION — triggers actual execution
result.show()        # triggers the full DAG
result.write.format("delta").mode("overwrite").saveAsTable("catalog.schema.regional_summary")

# Read from Azure Event Hubs as a stream
stream_df = (
    spark.readStream
    .format("eventhubs")
    .options(**eventhubs_conf)
    .load()
)

# Transform the stream
processed = (
    stream_df
    .select(
        col("body").cast("string").alias("payload"),
        col("enqueuedTime").alias("event_time")
    )
    .filter(col("payload").isNotNull())
)

# Write to Delta table with checkpointing
query = (
    processed.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/events")
    .trigger(processingTime="30 seconds")
    .toTable("catalog.schema.streaming_events")
)

query.awaitTermination()

# Adaptive Query Execution (AQE) — enabled by default in DBR 7.3+
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Dynamic Partition Pruning
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")

# Broadcast join for small tables (avoids shuffle)
from pyspark.sql.functions import broadcast

result = large_df.join(broadcast(small_lookup_df), "key_column")

# Cache frequently accessed DataFrames
hot_data = spark.table("catalog.schema.dimension_table").cache()
hot_data.count()  # materialize the cache

# Auto Loader — incrementally ingests new files from cloud storage
df = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/mnt/schema/orders")
    .option("cloudFiles.inferColumnTypes", "true")
    .load("/mnt/raw/orders/")
)

df.writeStream \
  .format("delta") \
  .option("checkpointLocation", "/mnt/checkpoints/orders") \
  .option("mergeSchema", "true") \
  .trigger(availableNow=True) \
  .toTable("catalog.bronze.raw_orders")

# CDC pattern — upsert incoming changes into a Delta table
from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "catalog.silver.customers")

(
    target.alias("target")
    .merge(
        source=cdc_df.alias("source"),
        condition="target.customer_id = source.customer_id"
    )
    .whenMatchedUpdate(
        condition="source.operation = 'UPDATE'",
        set={
            "name":       "source.name",
            "email":      "source.email",
            "updated_at": "source.updated_at"
        }
    )
    .whenNotMatchedInsert(
        condition="source.operation = 'INSERT'",
        values={
            "customer_id": "source.customer_id",
            "name":        "source.name",
            "email":       "source.email",
            "created_at":  "source.created_at"
        }
    )
    .whenMatchedDelete(
        condition="source.operation = 'DELETE'"
    )
    .execute()
)

import dlt
from pyspark.sql.functions import col, current_timestamp

# Bronze layer — raw ingestion
@dlt.table(
    name="raw_orders",
    comment="Raw orders ingested from cloud storage",
    table_properties={"quality": "bronze"}
)
def raw_orders():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .load("/mnt/raw/orders/")
    )

# Silver layer — cleansed with quality expectations
@dlt.table(
    name="cleaned_orders",
    comment="Validated and cleansed orders",
    table_properties={"quality": "silver"}
)
@dlt.expect_or_drop("valid_amount", "amount > 0")
@dlt.expect_or_drop("valid_customer", "customer_id IS NOT NULL")
@dlt.expect("valid_date", "order_date >= '2020-01-01'")
def cleaned_orders():
    return (
        dlt.read_stream("raw_orders")
        .select(
            col("order_id").cast("bigint"),
            col("customer_id").cast("bigint"),
            col("amount").cast("decimal(10,2)"),
            col("order_date").cast("date"),
            current_timestamp().alias("processed_at")
        )
    )

# Gold layer — business aggregation
@dlt.table(
    name="daily_revenue",
    comment="Daily revenue aggregated by region",
    table_properties={"quality": "gold"}
)
def daily_revenue():
    return (
        dlt.read("cleaned_orders")
        .groupBy("order_date", "region")
        .agg(
            sum("amount").alias("total_revenue"),
            count("*").alias("order_count")
        )
    )

my-databricks-project/
├── databricks.yml          ← bundle configuration
├── src/
│   ├── notebooks/
│   │   ├── ingest.py
│   │   └── transform.py
│   └── pipelines/
│       └── dlt_pipeline.py
├── resources/
│   ├── jobs/
│   │   └── daily_etl_job.yml
│   └── pipelines/
│       └── revenue_pipeline.yml
└── tests/
    ├── unit/
    └── integration/

bundle:
  name: dp750-data-pipeline

workspace:
  host: https://adb-xxxx.azuredatabricks.net

targets:
  dev:
    mode: development
    workspace:
      root_path: /Users/${workspace.current_user.userName}/.bundle/dev

  staging:
    workspace:
      root_path: /Shared/bundles/staging

  prod:
    mode: production
    workspace:
      root_path: /Shared/bundles/prod

resources:
  jobs:
    daily_etl:
      name: "Daily ETL Pipeline"
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "UTC"
      tasks:
        - task_key: ingest
          notebook_task:
            notebook_path: ./src/notebooks/ingest.py
        - task_key: transform
          depends_on:
            - task_key: ingest
          notebook_task:
            notebook_path: ./src/notebooks/transform.py

# Validate the bundle
databricks bundle validate --target dev

# Deploy to development
databricks bundle deploy --target dev

# Run a specific job
databricks bundle run daily_etl --target dev

# Deploy to production
databricks bundle deploy --target prod

# Configure diagnostic settings to send logs to Log Analytics
az monitor diagnostic-settings create \
  --name "databricks-diagnostics" \
  --resource "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.Databricks/workspaces/{ws}" \
  --workspace "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.OperationalInsights/workspaces/{law}" \
  --logs '[{"category":"dbfs","enabled":true},{"category":"clusters","enabled":true},{"category":"jobs","enabled":true}]'

Spark UI Navigation for Troubleshooting:
│
├── Jobs tab
│   └── Find failed or slow jobs → click to see stages
│
├── Stages tab
│   └── Look for:
│       ├── Skewed tasks (one task >> others in duration)
│       ├── Spill to disk (shuffle read/write metrics)
│       └── GC time (memory pressure indicator)
│
├── SQL / DataFrame tab
│   └── Query plan → look for:
│       ├── BroadcastHashJoin (good — no shuffle)
│       ├── SortMergeJoin (expensive — involves shuffle)
│       └── Exchange (shuffle boundary — minimize these)
│
└── Storage tab
    └── Cached RDDs/DataFrames — verify cache is being used

# Programmatically repair a failed job run
import requests

headers = {"Authorization": f"Bearer {token}"}

# Get failed run details
run = requests.get(
    f"{workspace_url}/api/2.1/jobs/runs/get",
    headers=headers,
    params={"run_id": failed_run_id}
).json()

# Repair — re-run only failed tasks
repair = requests.post(
    f"{workspace_url}/api/2.1/jobs/runs/repair",
    headers=headers,
    json={
        "run_id": failed_run_id,
        "rerun_all_failed_tasks": True
    }
)

Week 1 — Foundation
├── Azure Databricks workspace setup
├── Compute types and configuration
├── Delta Lake fundamentals
└── Unity Catalog hierarchy and objects

Week 2 — Governance Deep Dive
├── Unity Catalog privileges and RBAC
├── Row-level security and column masking
├── Data lineage and audit logging
└── Delta Sharing

Week 3 — Data Processing
├── Apache Spark architecture and internals
├── DataFrame API and Spark SQL
├── Structured Streaming
├── Auto Loader and CDC patterns
└── DLT / Lakeflow Declarative Pipelines

Week 4 — Pipelines and Operations
├── Lakeflow Jobs and scheduling
├── Databricks Asset Bundles and CI/CD
├── Git integration and branching strategies
├── Monitoring with Spark UI and Azure Monitor
└── Performance tuning and optimization

Week 5 — Practice and Review
├── Microsoft Learn free practice assessment
├── Review weak areas from practice tests
├── Hands-on labs in a real Databricks workspace
└── Review official study guide one more time

ACCOUNT ADMIN
    └── Metastore Admin
            └── Catalog Owner / USE CATALOG
                    └── Schema Owner / USE SCHEMA
                            └── Table Owner / SELECT / MODIFY
                                    └── Row Filters / Column Masks

Data Size < 1TB and query patterns change frequently?
    └─► Liquid Clustering (adaptive, no static partitions)

Data Size > 1TB with predictable filter columns?
    └─► Static Partitioning + ZORDER

Frequent point lookups on high-cardinality columns?
    └─► Bloom Filter Index

Many small files accumulating?
    └─► OPTIMIZE command + Auto Compaction

Soft deletes without full rewrites?
    └─► Deletion Vectors (enabled by default in DBR 14+)