Platform & production, one governed system.

Working entirely on the running project's OWN Delta tables (no new raw data), you measure a baseline, tune silver_pricing and gold_price_integrity_mart by query pattern, resolve the small-files trap left by streaming ingestion, set VACUUM + retention, rehearse a time-travel recovery, and open the Performance Notes section with a business-led before/after. Every step is 'measure first, change one thing, measure again.'

1. BASELINE (measure first): on silver_pricing and gold_price_integrity_mart run DESCRIBE DETAIL (numFiles, sizeInBytes) and DESCRIBE HISTORY; run one representative Maria query (e.g. price integrity for a given payer_id + hospital_ccn) with a cold cache and record wall-clock, bytes scanned, and files read from the SQL query profile. Screenshot the slow stage in the Spark UI. Paste all numbers into a baseline block — you cannot claim a win without this.
2. Diagnose the small-files trap: quantify median file size and file count growth from streaming writes. Confirm from the query profile whether the slow stage is scan/IO (too many small files, weak data skipping) vs shuffle/skew — do NOT fix what isn't slow.
3. Compact with OPTIMIZE (bin-packing) on the worst table, then re-run DESCRIBE DETAIL and the benchmark query; record the new file count/size and runtime. Note the delta before adding clustering so you can attribute each gain.
4. Add liquid clustering keyed to the QUERY PATTERN: ALTER TABLE ... CLUSTER BY (payer_id, hospital_ccn, billing_code) chosen from Maria's actual filters, then OPTIMIZE to realize the layout. Re-run the benchmark; record files pruned and bytes scanned vs baseline. Contrast with what naive partitioning by billing_code would have done (compute the resulting partition/file count on paper).
5. Decide and document partition-vs-cluster per table using the >~1 GB-per-partition rule; write one paragraph each justifying why you clustered (or, for a genuinely low-cardinality column like ingest_date, partitioned). This written 'why' is an interview answer.
6. Set retention + VACUUM: choose a defensible retention window (default 7 days; justify any change), run VACUUM in DRY RUN first, then for real, and note reclaimed bytes. Explain in the runbook how shortening retention would break time travel / concurrent readers.
7. Rehearse recovery (build for failure): simulate a bad write (e.g. a MERGE that overwrites prices with nulls), detect it, then recover with RESTORE TABLE ... TO VERSION AS OF the last-good version; confirm row counts and a spot-checked price match pre-incident, and write an audit-log row (table, bad version, restored version, timestamp, why).
8. Sanity-check idempotency isn't broken: re-run the streaming/incremental load once after tuning to confirm OPTIMIZE/clustering didn't change results, only layout.
9. Write the Performance Notes section: a before/after table (files, bytes scanned, runtime, est. DBU) with a one-line headline LEADING WITH THE BUSINESS OUTCOME ('mart now lands ~4 min, comfortably before the 8am review; ~55% fewer bytes scanned'), then the technical changes underneath.

• Databricks — Compact data files with OPTIMIZE on Delta Lake
• Databricks — Use liquid clustering for Delta tables
• Databricks — Data skipping and Z-ordering for Delta Lake
• Databricks — When to partition tables on Databricks
• Databricks — Remove unused data files with VACUUM
• Databricks — Work with Delta Lake table history and time travel
• Databricks — Predictive optimization for Unity Catalog managed tables

# Databricks notebook source
# W31 - Delta optimization: MEASURE FIRST, change one thing, measure again.
# Runs against the project's OWN tables. No new raw data.

CATALOG = "ppi"          # TODO: your project catalog/schema from Phase 2
SCHEMA  = "silver"
TABLES  = ["silver_pricing", "gold_price_integrity_mart"]  # TODO: confirm names

# --- 1. BASELINE ------------------------------------------------------------
# TODO: for each table capture numFiles + sizeInBytes. You need these numbers
#       BEFORE any change or you cannot claim a win.
for t in TABLES:
    fqn = f"{CATALOG}.{SCHEMA}.{t}"
    display(spark.sql(f"DESCRIBE DETAIL {fqn}"))      # numFiles, sizeInBytes
    # TODO: DESCRIBE HISTORY {fqn} -> note current version (your 'last good')

# TODO: define ONE representative Maria query (filter by payer_id + hospital_ccn).
#       Run it cold, then open the SQL query profile and RECORD BY HAND:
#       wall-clock, bytes scanned, files read, and the slowest stage.
BENCHMARK_SQL = """
  -- TODO: paste the real mart/price-integrity query Maria runs
"""
# spark.sql(BENCHMARK_SQL).collect()   # force execution; then read the profile

# --- 2. DIAGNOSE ------------------------------------------------------------
# TODO: is the slow stage scan/IO (small files, weak skipping) or shuffle/skew?
#       Only fix what the profile proves is slow.

# --- 3. OPTIMIZE (compaction) ----------------------------------------------
# TODO: OPTIMIZE the worst table, then re-DESCRIBE DETAIL and re-run BENCHMARK.
#       Record the delta BEFORE adding clustering (attribute each gain).

# --- 4. LIQUID CLUSTERING BY QUERY PATTERN ---------------------------------
# Keys come from Maria's FILTERS, not the data model. Justify your choice.
CLUSTER_KEYS = []  # TODO e.g. ["payer_id", "hospital_ccn", "billing_code"]
# TODO: ALTER TABLE ... CLUSTER BY (...) ; then OPTIMIZE to realize layout;
#       re-run BENCHMARK and record files pruned + bytes scanned vs baseline.

# --- 5. VACUUM + RETENTION --------------------------------------------------
# TODO: VACUUM ... DRY RUN first; justify retention (default 7 days) in runbook;
#       then VACUUM for real and note reclaimed bytes.

# --- 6. RECOVERY REHEARSAL (build for failure) -----------------------------
# TODO: simulate a bad MERGE (null out prices), detect it, then
#       RESTORE TABLE {fqn} TO VERSION AS OF <last_good>; confirm row counts;
#       append an audit-log row: table, bad_version, restored_version, ts, why.

# --- 7. PERFORMANCE NOTES ---------------------------------------------------
# TODO: write before/after to the README Performance Notes section.
#       Headline the BUSINESS OUTCOME first (lands before 8am / DBU down ~X%),
#       technical change second.

• Managed vs External Tables in Databricks: The $15K Mistake

• Delta Lake: The Definitive Guide O'Reilly · Book
• Databricks Certified Data Engineer Professional - Preparation (Derar Alhussein) Udemy · Course

Retrofit the entire Phase 2 Provider Pricing Integrity project into Unity Catalog: design and stand up the catalog/schema structure per medallion layer, make and document a managed-vs-external decision per table, migrate the tables, lock them down with persona-based GRANTs, tag PHI-adjacent columns, and capture built-in lineage as audit evidence. No new raw data — you are governing the project you already built.

1. DESIGN FIRST (on paper, before any SQL): sketch the target namespace. Decide one project catalog (e.g. ppi_prod) with schemas bronze/silver/gold/ops, OR per-layer catalogs — and write 3-4 sentences on why, tied to how Maria's analysts vs engineers query. Draft a managed-vs-external decision table with one row per project table and a reason column.
2. Stand up the structure with a parameterized notebook: CREATE CATALOG IF NOT EXISTS, then CREATE SCHEMA per layer, plus MANAGED LOCATION / an external location + storage credential for any layer you decided to keep external. Set COMMENT and OWNER (a group, not you) on each object.
3. Migrate the Phase 2 tables into UC. For managed targets use CREATE TABLE ... DEEP CLONE or CTAS from the old hive_metastore tables; for external, register with CREATE TABLE ... LOCATION. Verify with SELECT * FROM information_schema.tables that every project table resolves under the new namespace and count = expected.
4. Implement RBAC for 3 personas via groups: GRANT USE CATALOG/USE SCHEMA + SELECT on gold to an analysts group (Maria), MODIFY on bronze/silver to an engineers group, and read-only + BROWSE to an auditors group. Keep it least-privilege — no blanket ALL PRIVILEGES.
5. Prove enforcement: as (or simulating) the analyst persona, run a SELECT on a silver table you did NOT grant and capture the permission-denied error; then GRANT the minimum needed and re-run to success. Save both outputs — this denied-then-granted pair is your interview evidence.
6. Tag for governance: ALTER TABLE/ALTER COLUMN ... SET TAGS for phi_adjacent, pii, medallion_layer, data_owner, and cost_center on the sensitive provider/pricing columns. Then query them back from system.information_schema tag views to prove 'show me every PHI-adjacent column' is one query.
7. Capture lineage: run the silver->gold KPI transformation once under UC so lineage populates, open the lineage graph for a gold KPI table, screenshot the column-level path from CMS pricing fields, and run a system.access lineage query to export the same as a table.
8. DOCUMENT: update the Architecture Doc with the final namespace diagram, the managed-vs-external decision table, the GRANT matrix, the tag taxonomy, and the lineage screenshot. Write a short 'why' note: why this catalog boundary, why groups over users, why external for the layer(s) you chose. Idempotency check: re-run the setup notebook end-to-end and confirm it is safe (IF NOT EXISTS / no errors) on the 2nd and 3rd run.

• What is Unity Catalog? (overview & object model)
• Manage privileges in Unity Catalog (GRANT / REVOKE, privilege types)
• Managed tables vs external tables
• Unity Catalog best practices (catalog/schema organization)
• Apply tags to Unity Catalog securable objects
• Capture and view data lineage using Unity Catalog
• Monitor usage with system tables

-- Provider Pricing Integrity :: Unity Catalog retrofit (SCAFFOLD)
-- Parameterize the environment so this notebook is idempotent + reusable.
-- TODO: set these widgets/vars for your workspace
-- SET catalog_name = 'ppi_prod';           -- TODO: your project catalog
-- SET analysts_group = 'ppi_analysts';       -- TODO: Maria's group
-- SET engineers_group = 'ppi_engineers';     -- TODO
-- SET auditors_group  = 'ppi_auditors';      -- TODO

-- 1) NAMESPACE ---------------------------------------------------------
-- TODO: justify one-catalog-many-schemas vs per-layer-catalog in the Arch Doc first.
CREATE CATALOG IF NOT EXISTS ${catalog_name}
  COMMENT 'Provider Pricing Integrity — governed project catalog';
-- TODO: create one schema per medallion layer (+ an ops schema for audit/quarantine)
-- CREATE SCHEMA IF NOT EXISTS ${catalog_name}.bronze COMMENT '...';
-- ... silver / gold / ops

-- 2) OWNERSHIP ---------------------------------------------------------
-- TODO: transfer ownership to a GROUP, never your personal user
-- ALTER SCHEMA ${catalog_name}.silver OWNER TO `${engineers_group}`;

-- 3) MANAGED vs EXTERNAL ----------------------------------------------
-- TODO: for each Phase 2 table, choose ONE path and record why in the decision table.
-- Managed (let UC own the bytes):
--   CREATE TABLE ${catalog_name}.silver.provider_clean
--   DEEP CLONE hive_metastore.<old_schema>.provider_clean;   -- TODO
-- External (you own the storage lifecycle):
--   CREATE TABLE ${catalog_name}.bronze.pricing_raw
--   LOCATION 'abfss://.../bronze/pricing_raw';               -- TODO cloud path

-- 4) LEAST-PRIVILEGE RBAC ---------------------------------------------
-- TODO: grant the MINIMUM. Gate with USE, then SELECT/MODIFY per persona.
-- GRANT USE CATALOG ON CATALOG ${catalog_name} TO `${analysts_group}`;
-- GRANT USE SCHEMA, SELECT ON SCHEMA ${catalog_name}.gold TO `${analysts_group}`;
-- GRANT MODIFY ON SCHEMA ${catalog_name}.silver TO `${engineers_group}`;
-- GRANT BROWSE ON CATALOG ${catalog_name} TO `${auditors_group}`;
-- Do NOT: GRANT ALL PRIVILEGES ... (explain in your why-note why not)

-- 5) GOVERNANCE TAGS ---------------------------------------------------
-- TODO: tag PHI-adjacent + PII columns so 'find sensitive data' is a query.
-- ALTER TABLE ${catalog_name}.silver.provider_clean
--   ALTER COLUMN provider_name SET TAGS ('pii' = 'true', 'phi_adjacent' = 'true');
-- ALTER SCHEMA ${catalog_name}.gold SET TAGS ('data_owner'='maria', 'cost_center'='rcm');

-- 6) VERIFY (evidence for the Arch Doc) --------------------------------
-- SELECT table_schema, table_name, table_type
-- FROM ${catalog_name}.information_schema.tables ORDER BY 1,2;   -- TODO screenshot
-- TODO: query column tags back from information_schema.column_tags

• End-to-End Unity Catalog Setup
• $180k Databricks Engineers Mastered This One Method

• Data Governance with Unity Catalog on Databricks O'Reilly · Book
• Databricks Certified Data Engineer Associate Study Guide O'Reilly · Book
• Databricks Certified Data Engineer Associate - Preparation (Derar Alhussein) Udemy · Course

Convert the Phase 2 bronze→silver stage of the Provider Pricing Integrity pipeline into a declarative Lakeflow pipeline in your Unity Catalog (from W32), with the W28 DQ rules re-expressed as named, tiered expectations. You will define streaming tables for CMS pricing ingest and silver validated providers, a materialized view for a conformed KPI-ready output, run a full and then an incremental update, and then read lineage plus per-expectation quality metrics from the event log — recording pass/drop/fail counts and writing the 'why this tier' justification for each rule.

1. SETUP & FRAME: In your project repo, create a new pipeline source notebook/file (e.g. pipelines/silver_pricing.py). Open it with a 1–2 sentence business-framing comment naming Maria and the contract-decision risk. Confirm your W32 Unity Catalog + schema (catalog.schema) and the UC Volume landing path where raw CMS pricing files arrive — you target UC, not hive_metastore.
2. DECLARE BRONZE STREAMING TABLE: Using the starter scaffold, define an Auto Loader (cloud_files) streaming table that ingests the raw CMS Hospital Price Transparency files into bronze as append-only with ingest metadata (source file, ingest timestamp) — no business logic. Run a first pipeline full-refresh update and confirm bronze populates.
3. TRANSLATE THE W28 RULES: In a markdown cell, list your existing W28 DQ rules and, next to each, write the boolean constraint expression and the chosen tier. Minimum 6: pricing-completeness (rate present), NPI 10-digit format, non-null grain composite key (provider+code+payer), negotiated-rate plausibility bounds (>0 and < a sane ceiling), payer/plan not-null, and a stale-file freshness flag. This is a design table you will defend.
4. DEFINE SILVER STREAMING TABLE WITH EXPECTATIONS: Build the silver validated_pricing streaming table reading from bronze (dlt.read_stream) and apply your expectations: @dlt.expect_or_fail on the non-null grain key (unrecoverable — halt), @dlt.expect_or_drop on bad NPI and implausible rates (quarantine, not golden path), and @dlt.expect (warn/track) on soft-completeness signals. Type/conform columns as in Phase 2.
5. ADD THE MATERIALIZED VIEW: Define a materialized view that conforms silver into a KPI-ready shape (one clean row per grain with cleaned rate + provider attributes) — the input your Phase 2 gold marts will later consume. Comment WHY this is a materialized view (recomputed conformed output) while the two silver tables are streaming (incremental append ingest).
6. RUN FULL + INCREMENTAL: Trigger a full pipeline update; confirm the lineage/DAG renders and all three tables build in inferred dependency order. Then drop ONE new/changed CMS file into the landing path and run an incremental update — confirm only new data flows through and the streaming tables append without reprocessing everything.
7. READ QUALITY METRICS + LINEAGE: Query the pipeline event log (event_log TVF or the pipeline system table) to pull per-expectation results for your latest update; record passing/dropped/failing counts per named expectation into a results table in the notebook/Runbook. Then open the UC lineage for the materialized view and confirm column-level lineage traces back through silver to bronze to the source files (ties to W32); capture it as governance evidence.
8. FORCE A FAILURE ON PURPOSE (break-safely proof): Introduce a row with a null grain key (or point at a deliberately corrupt test file) and run an update; confirm the expect_or_fail expectation HALTS the update loudly rather than passing. Then confirm a bad-NPI row is DROPPED (not failed) and shows in the drop count. Revert. This proves loud-not-silent behavior for the interview.
9. WRITE THE 'WHY' NOTE: In the architecture doc, add a 'Data Contract' subsection: for each expectation, one line stating the rule, its tier, and the Maria-facing consequence that justifies the tier (e.g. 'missing grain key → fail: a row with no key can't be deduped or trusted in a contract review'). Add the event-log pass/drop/fail numbers as the observability evidence.

• Lakeflow Declarative Pipelines — overview and concepts (Databricks docs)
• Manage data quality with pipeline expectations (Databricks docs)
• Load data with streaming tables and Auto Loader in Lakeflow (Databricks docs)
• Materialized views and streaming tables — concepts (Databricks docs)
• Monitor pipelines with the event log (Databricks docs)
• Lakeflow Declarative Pipelines Python reference — @dlt decorators and expectations API

# Provider Pricing Integrity — Silver stage as a declarative Lakeflow pipeline
# BUSINESS FRAMING: Maria's team makes payer-contract decisions on this data.
# A silent bad load = wrong negotiated rates in a contract review. This pipeline
# must ENFORCE and REPORT its data contract on every row, not just "run green."
#
# NOTE: configure catalog/schema in the pipeline settings (target your W32 UC),
# and set the CMS landing path as a pipeline parameter. Do NOT hardcode secrets.

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

# CMS raw file landing location (set via pipeline config / spark conf)
# TODO: read this from a pipeline parameter, not a literal
CMS_LANDING_PATH = spark.conf.get("pricing.landing_path", "/Volumes/<catalog>/<schema>/landing/cms_pricing")


# ---------------------------------------------------------------------------
# 1) BRONZE — streaming ingest of raw CMS pricing files via Auto Loader
#    Append-only. NO business logic here. Keep it thin.
# ---------------------------------------------------------------------------
@dlt.table(
    name="bronze_cms_pricing",
    comment="Raw CMS Hospital Price Transparency files, append-only via Auto Loader.",
)
def bronze_cms_pricing():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "csv")  # TODO: confirm real format per file drop
        # TODO: add schema hints / schemaLocation; CMS files are inconsistent
        .load(CMS_LANDING_PATH)
        .withColumn("_ingest_ts", current_timestamp())
        .withColumn("_source_file", input_file_name())
    )


# ---------------------------------------------------------------------------
# 2) SILVER — validated, typed, conformed pricing as a STREAMING table
#    This is where the W28 rules become tiered expectations.
#    Tier guide: expect = warn/track | expect_or_drop = quarantine
#                expect_or_fail = halt the update (unrecoverable)
# ---------------------------------------------------------------------------
@dlt.table(
    name="silver_validated_pricing",
    comment="Typed, conformed pricing with enforced data contract.",
)
# HARD FAIL — a row with no grain key cannot be deduped or trusted downstream.
# TODO: replace with your real composite-grain null check (provider+code+payer)
@dlt.expect_or_fail("grain_key_present", "provider_npi IS NOT NULL AND code IS NOT NULL AND payer IS NOT NULL")
# QUARANTINE — malformed NPI: drop from golden path, count it, don't kill the run.
# TODO: tighten to your real 10-digit NPI shape rule from W28
@dlt.expect_or_drop("npi_10_digit", "provider_npi RLIKE '^[0-9]{10}$'")
# QUARANTINE — implausible negotiated rate.
# TODO: set the sane ceiling you justified in W28
@dlt.expect_or_drop("rate_in_bounds", "negotiated_rate > 0 AND negotiated_rate < 1000000")
# WARN/TRACK — soft completeness signal you measure but do not enforce.
@dlt.expect("payer_plan_present", "payer IS NOT NULL AND plan IS NOT NULL")
# TODO: add at least 2 more named expectations to reach your >=6 rule contract
def silver_validated_pricing():
    df = dlt.read_stream("bronze_cms_pricing")
    # TODO: cast/clean columns to typed schema (rate -> decimal, trim codes, etc.)
    # TODO: conform CMS column-name variants to your canonical silver schema
    # TODO: derive provider_npi / code / payer / plan / negotiated_rate
    return df


# ---------------------------------------------------------------------------
# 3) MATERIALIZED VIEW — conformed, KPI-ready output for the (later) gold marts.
#    Materialized view because it is a RECOMPUTED conformed output, not an
#    incremental append. Downstream Phase 2 gold marts will read this.
# ---------------------------------------------------------------------------
@dlt.table(
    name="silver_pricing_conformed_mv",
    comment="One clean row per grain, KPI-ready. Feeds gold marts.",
)
def silver_pricing_conformed_mv():
    # TODO: dedupe to one row per composite grain + select KPI-facing columns
    return dlt.read("silver_validated_pricing")


# AFTER RUNNING: query the event log for per-expectation pass/drop/fail counts
# and paste them into the Runbook. TODO: add your event_log TVF query here or in
# a companion notebook.

• Your Orchestration Isn't Ready for AI
• Databricks Notebooks in Production: Director's Verdict
• Hybrid Notebook/Script Architecture Map (free)

• Databricks Certified Data Engineer Associate - Preparation (Derar Alhussein) Udemy · Course
• Data Engineering with Databricks Cookbook O'Reilly · Book

Take the Phase 2 medallion pipeline (Lakeflow declarative pipeline from W33 + MDM golden refresh + gold marts/KPIs) and orchestrate it as one production multi-task Databricks Job DAG. Add task dependencies, retries, timeouts, parameters, a cron schedule, and both failure and freshness-SLA notifications — then deliberately break it and recover with Repair Run, and prove idempotency across 10 reruns.

1. Inventory the DAG on paper first: list the 3 stages (Lakeflow pipeline → MDM golden refresh → gold marts/KPI refresh) and draw the dependency edges + which KPIs each stage feeds. Write the intended 8am freshness SLA in your RUNBOOK as the target you'll alert on.
2. Create the Job with a **pipeline task** that runs your W33 Lakeflow declarative pipeline (select the existing pipeline; full-refresh=false). Confirm it runs green on its own before adding anything downstream.
3. Add task 2 (MDM golden-record refresh notebook/SQL) with `depends_on` = the pipeline task, and task 3 (gold marts + KPI refresh) depending on task 2. Set task 3's `run_if = ALL_SUCCESS` so it never publishes off a half-built silver/MDM layer.
4. Add reliability controls to each task: `max_retries` (2) with a `min_retry_interval_millis`, a `timeout_seconds` sized to ~1.5x normal runtime, and set the job `max_concurrent_runs = 1`. Add a job-level `run_date` and `env` parameter and read them inside the MDM/gold tasks (TODO markers in the scaffold) instead of hard-coding.
5. Pass state between tasks with `taskValues`: have the MDM task set `providers_collapsed` and `records_quarantined`, and have the gold/KPI task read them to stamp the run's KPI row (so lineage of counts flows through the DAG, not a shared side table).
6. Wire notifications: on-failure to your email + a webhook system destination, AND add an on-duration-threshold-exceeded alert set to your 8am-SLA budget. Set a cron schedule (nightly, correct timezone) but leave it PAUSED for the lab.
7. Break it on purpose: add a temporary `assert False` (or a bad cast) in the MDM task, trigger the run, and watch (a) the automatic retries exhaust, (b) the on-failure alert fire, (c) task 3 stay skipped because ALL_SUCCESS wasn't met. Screenshot the run timeline.
8. Recover cleanly: remove the injected failure, use **Repair Run** to resume from the failed MDM task (NOT a full rerun), and confirm only the failed + downstream tasks re-execute. Note in the RUNBOOK the exact repair procedure.
9. Prove idempotency: trigger the full Job 10x and record gold row counts + the 4 KPIs after each run; they must be stable (MERGE/overwrite-by-key, not append). Read the Gantt to find the single longest task and note it as next week's optimization candidate in Performance Notes.

• Databricks — Create and run Databricks Jobs (multi-task orchestration)
• Databricks — Configure task dependencies and control flow (depends_on, run_if)
• Databricks — Retries, timeouts, and max concurrent runs for job tasks
• Databricks — Add a pipeline (Lakeflow declarative pipeline) task to a job
• Databricks — Parameterize jobs and pass values between tasks (job/task parameters, taskValues)
• Databricks — Notifications and system notification destinations for jobs
• Databricks — Repair an unsuccessful job run

# resources/pricing_integrity_job.yml
# Multi-task Databricks Job that orchestrates the Provider Pricing Integrity DAG.
# Deploy via the UI (paste equivalents), Jobs API, or later as a DAB resource (W41).
resources:
  jobs:
    pricing_integrity_nightly:
      name: pricing_integrity_nightly
      max_concurrent_runs: 1        # TODO: confirm 1 so a slow run never stacks

      parameters:
        - name: run_date            # TODO: read this in MDM + gold tasks; do NOT hard-code dates
          default: "{{job.trigger.time.iso_date}}"
        - name: env
          default: dev              # TODO: dev|prod -> resolves catalog/schema

      # --- SCHEDULE: stay on a triggered/cron schedule this week (streaming is W35) ---
      schedule:
        quartz_cron_expression: "TODO"   # TODO: nightly, well before the 8am SLA
        timezone_id: "TODO"              # TODO: your real timezone, not UTC by accident
        pause_status: PAUSED             # keep PAUSED during the lab

      # --- NOTIFICATIONS: break LOUD, and trip on freshness ---
      email_notifications:
        on_failure:
          - "TODO@your.email"           # TODO
      # webhook_notifications: {}         # TODO: add a system destination (Slack/PagerDuty)
      health:
        rules:
          - metric: RUN_DURATION_SECONDS # freshness-SLA tripwire
            op: GREATER_THAN
            value: 0                      # TODO: set to your 8am-SLA budget in seconds

      tasks:
        - task_key: lakeflow_pipeline
          pipeline_task:
            pipeline_id: "TODO"           # TODO: your W33 Lakeflow declarative pipeline id
            full_refresh: false
          timeout_seconds: 0              # TODO: ~1.5x normal runtime
          max_retries: 0                  # TODO: 2
          # min_retry_interval_millis: TODO

        - task_key: mdm_golden_refresh
          depends_on:
            - task_key: lakeflow_pipeline
          # notebook_task / sql_task:      # TODO: point at your MDM refresh
          #   base_parameters: { run_date: "{{job.parameters.run_date}}", env: "{{job.parameters.env}}" }
          # TODO: set taskValues providers_collapsed, records_quarantined for the gold task
          max_retries: 0                  # TODO
          timeout_seconds: 0              # TODO

        - task_key: gold_marts_kpis
          depends_on:
            - task_key: mdm_golden_refresh
          run_if: ALL_SUCCESS            # never publish off a half-built layer
          # TODO: read taskValues from mdm_golden_refresh to stamp the KPI row
          # TODO: point at your gold marts + KPI refresh; keep it idempotent (MERGE/overwrite-by-key)

• Your Orchestration Isn't Ready for AI
• Databricks Notebooks in Production: Director's Verdict
• Hybrid Notebook/Script Architecture Map (free)

• Introduction to Apache Airflow in Python DataCamp · Course
• Databricks Certified Data Engineer Professional - Preparation (Derar Alhussein) Udemy · Course

Refactor the Phase-2 bronze CMS-pricing ingest from a batch spark.read into an Auto Loader stream (cloudFiles) writing to the existing bronze Delta table under Unity Catalog. You will configure schemaLocation + a checkpoint, choose Trigger.AvailableNow, prove new-files-only processing and exactly-once recovery by killing and restarting mid-stream, and force a schema-evolution event to see a new column land in _rescued_data instead of silently dropping. You end with a short written 'streaming ≠ real-time' note and a freshness-SLA statement wired into the RUNBOOK.

1. Baseline & framing: open your Phase-2 bronze batch notebook, record its current input-file count and run cost/time, and write one sentence naming the freshness SLA the stream must meet for Maria (e.g., 'corrected pricing files visible within 30 min of landing'). Stage a landing volume in Unity Catalog (a UC managed Volume or external location path) and drop 3–4 real CMS pricing files into it to start.
2. Build the Auto Loader readStream: use format('cloudFiles') with cloudFiles.format ('csv' or 'json' matching your CMS files), a dedicated cloudFiles.schemaLocation, and rescuedDataColumn enabled. Read from the landing volume path. Do NOT yet write — display the stream's schema and confirm inference picked up the pricing columns plus _rescued_data.
3. Wire the sink with a checkpoint: writeStream to your existing bronze UC table (toTable or Delta path) with a dedicated, per-table checkpointLocation, outputMode('append'), and .trigger(availableNow=True). Add lineage/audit columns (_ingest_file = input_file_name(), _ingest_ts = current_timestamp()) so every row is traceable — carrying forward the Phase-2 'timestamp + reason' audit habit.
4. Prove new-files-only (idempotent reruns): run the stream (it catches up and stops). Record rows written. Re-run with NO new files and confirm 0 rows processed. Drop 1 new CMS file, re-run, and confirm ONLY that file's rows land. Screenshot the input-rows metric across the 3 runs for your Performance Notes — this is the batch-relist-vs-incremental win.
5. Prove exactly-once recovery: drop several files, start the stream, and kill the cluster/cancel the query mid-batch. Restart from the SAME checkpoint. Verify the bronze row count equals the expected total with zero duplicates (group-by _ingest_file counts, or compare to a batch count of the same files). Write 2–3 sentences in the RUNBOOK explaining WHY the checkpoint + Delta idempotent commit gave you this.
6. Force a schema-evolution event: craft one CMS file with a new/renamed column (e.g., an extra 'payer_specific_negotiated_charge_pct' field). Run with schemaEvolutionMode='rescue' first — find the surprise value in _rescued_data. Then switch to 'addNewColumns' (the stream stops with the new schema; restart adopts it) and confirm the column is now first-class. Record what each mode did.
7. Trigger decision write-up: in the architecture doc, add a short table comparing availableNow vs. processingTime='10 minutes' vs. continuous for THIS project, and state which you chose and why (cost posture + freshness SLA). Do not implement always-on unless your SLA demands it — measure first, don't over-engineer.
8. Freshness SLA + 'not real-time' note: write the one-paragraph stakeholder explainer for Maria (streaming ≠ live), define the numeric freshness SLA, and add a RUNBOOK entry: how to restart the stream, where the checkpoint lives, and the one thing you must never do (delete/share/edit the checkpoint or silently change query logic).

• Databricks — What is Auto Loader?
• Databricks — Configure schema inference and evolution in Auto Loader
• Databricks — Configure Structured Streaming triggers (incl. Trigger.AvailableNow)
• Databricks — Structured Streaming checkpoints
• Databricks — Production considerations for Structured Streaming
• Databricks — Use Auto Loader with Unity Catalog

# bronze_cms_pricing_autoloader.py
# Week 35 - Structured Streaming I: batch bronze ingest -> Auto Loader stream.
# STARTER SCAFFOLD. Fill every TODO. Do NOT turn this into a full solution.

from pyspark.sql import functions as F

# --- Config (edit for your Unity Catalog + landing path) -------------------
CATALOG = "ppi"                 # TODO: your Phase-2 catalog
SCHEMA = "bronze"
TARGET_TABLE = f"{CATALOG}.{SCHEMA}.cms_pricing_raw"   # existing bronze table

# TODO: a UC Volume / external location where CMS pricing files LAND
LANDING_PATH = "/Volumes/ppi/landing/cms_pricing/"

# TODO: dedicated, PER-TABLE dirs. Never share a checkpoint between queries.
SCHEMA_LOCATION = "/Volumes/ppi/_schemas/cms_pricing/"
CHECKPOINT_LOCATION = "/Volumes/ppi/_checkpoints/cms_pricing/"

# --- Read: Auto Loader (cloudFiles) ---------------------------------------
src = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")            # TODO: match your CMS files
    .option("cloudFiles.schemaLocation", SCHEMA_LOCATION)
    # TODO: choose an evolution mode and DEFEND it in your write-up.
    #       Try 'rescue' first, then 'addNewColumns'.
    .option("cloudFiles.schemaEvolutionMode", "rescue")
    .option("rescuedDataColumn", "_rescued_data")  # break loud, not silent
    # TODO: header / delimiter / inference options for your files
    .load(LANDING_PATH)
)

# --- Lineage/audit columns (carry the Phase-2 audit habit) ----------------
enriched = (
    src
    .withColumn("_ingest_file", F.col("_metadata.file_path"))
    .withColumn("_ingest_ts", F.current_timestamp())
    # TODO: add any conforming/casting you already do in batch bronze
)

# --- Write: Delta sink + checkpoint + trigger -----------------------------
query = (
    enriched.writeStream
    .option("checkpointLocation", CHECKPOINT_LOCATION)
    .outputMode("append")
    # TODO: JUSTIFY this trigger vs processingTime vs continuous in the arch doc.
    .trigger(availableNow=True)
    .toTable(TARGET_TABLE)
)

query.awaitTermination()

# TODO (do NOT script yet — run by hand and record results):
#  1) Run twice with no new files -> prove 0 rows the 2nd time.
#  2) Kill mid-run, restart SAME checkpoint -> prove 0 duplicates.
#  3) Drop a file with a NEW column -> find it in _rescued_data, then evolve.
#  Capture input-row metrics for your Performance Notes.

• Build a Real-Time Streaming Project | Live Demo

• Stream Processing with Apache Spark O'Reilly · Book
• Streaming Concepts DataCamp · Course
• Learning Spark, 2nd Ed. O'Reilly · Book

Turn the provider-change event feed you began ingesting in W35 into a production-grade stateful stream: attach a watermark to bound state, dedup on a composite business key, quarantine beyond-watermark late arrivals, and idempotently upsert the survivors into the silver Delta table via foreachBatch + MERGE — then prove idempotency with a 10x replay and inspect the state metrics.

1. Open the runbook stub and write the business framing (2 sentences) at the top of the notebook: what a mis-applied or duplicated provider change costs Maria. This is recognition-before-evaluation, not a canned intro.
2. Generate the event feed: run the provider-change generator (Faker over real NPIs pulled from the NPPES bulk file) to emit JSON events into the landing Volume, DELIBERATELY injecting (a) exact duplicate retries, (b) out-of-order event_ts, and (c) a handful of events 30h+ late. Inspect a sample and note the mess in the why note.
3. Ingest with readStream + Auto Loader (reuse the W35 cloudFiles config), applying an EXPLICIT schema (no inference in production) and casting event_ts from string to a real timestamp so watermarking can work.
4. Apply .withWatermark('event_ts', '24 hours') then .dropDuplicatesWithinWatermark(['npi','change_type','event_ts']); write one sentence justifying the 24h threshold from the lateness you observed in step 2.
5. Write the foreachBatch function: inside the batch, tag rows whose event_ts falls beyond the late threshold and write them to the quarantine table with reason='beyond_watermark' and quarantined_at=current_timestamp(); keep the good rows.
6. MERGE the good rows into the silver Delta table on the composite key — WHEN MATCHED AND source.event_ts > target.event_ts THEN UPDATE, WHEN NOT MATCHED THEN INSERT — so a newer change wins and a stale duplicate is a no-op.
7. Set checkpointLocation and run the stream with trigger(availableNow=True); confirm the silver duplicate-provider count is 0 and quarantine count > 0.
8. Idempotency test: re-run the same availableNow batch 10x. Assert the silver row count is unchanged and use Delta time travel (DESCRIBE HISTORY / VERSION AS OF) to show no duplicate rows were appended across versions.
9. Read query.lastProgress['stateOperators'] and record numRowsTotal across several micro-batches into a Performance/State Notes section; finish the why note (threshold, dedup grain, late-handling decision) as interview-ready answers.

• Apply watermarks to control data processing thresholds (Databricks)
• Use foreachBatch to write to arbitrary data sinks (Databricks)
• Upsert into a Delta Lake table using merge (MERGE INTO) (Databricks)
• Idempotent table writes in foreachBatch — txnAppId/txnVersion (Databricks Delta + Structured Streaming)
• Structured Streaming in production — state store, RocksDB, and stateful query tuning (Databricks)

# Week 36 - Stateful dedup + watermark for the provider-change event stream.
# SCAFFOLD ONLY. Fill every TODO. Do not paste a full solution here.

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType

# --- Config (edit for your workspace) ----------------------------------
CATALOG, SCHEMA = "ppi", "silver"
EVENTS_VOLUME    = "/Volumes/ppi/landing/provider_change_events"
CHECKPOINT       = "/Volumes/ppi/_checkpoints/provider_change_dedup"
SILVER_TABLE     = f"{CATALOG}.{SCHEMA}.provider_change"
QUARANTINE_TABLE = f"{CATALOG}.{SCHEMA}.provider_change_quarantine"

WATERMARK_DELAY = "24 hours"   # TODO: justify from OBSERVED lateness, not a guess
DEDUP_KEYS = ["npi", "change_type", "event_ts"]  # TODO: confirm the true business grain

# --- Explicit schema (never infer a production stream) -----------------
event_schema = StructType([
    StructField("npi", StringType()),
    StructField("change_type", StringType()),
    StructField("event_ts", StringType()),   # parse to timestamp below
    # TODO: add payload fields (new_address, new_specialty, deactivation_flag, ...)
])

# --- Read stream (reuse the W35 Auto Loader config) --------------------
def read_stream(spark):
    df = (spark.readStream.format("cloudFiles")
          .option("cloudFiles.format", "json")
          .schema(event_schema)
          .load(EVENTS_VOLUME))
    # TODO: df = df.withColumn("event_ts", F.to_timestamp("event_ts"))
    return df

# --- Watermark + stateful dedup ----------------------------------------
def dedup(df):
    # TODO: df.withWatermark("event_ts", WATERMARK_DELAY)
    # TODO:   .dropDuplicatesWithinWatermark(DEDUP_KEYS)
    return df

# --- foreachBatch: quarantine the too-late, MERGE the rest -------------
def upsert_batch(batch_df, batch_id):
    # 1) TODO: tag beyond-threshold rows -> write to QUARANTINE_TABLE with
    #    reason='beyond_watermark', quarantined_at=current_timestamp()
    #    (build for failure: quarantine, never silently drop)
    # 2) TODO: MERGE good rows into SILVER_TABLE on DEDUP_KEYS:
    #    WHEN MATCHED AND s.event_ts > t.event_ts THEN UPDATE
    #    WHEN NOT MATCHED THEN INSERT
    # 3) MERGE keyed on the dedup grain makes a batch replay a no-op (idempotent).
    pass

def main(spark):
    q = (dedup(read_stream(spark)).writeStream
         .foreachBatch(upsert_batch)
         .option("checkpointLocation", CHECKPOINT)
         # TODO: .trigger(availableNow=True)  # enables the 10x rerun idempotency test
         .start())
    # TODO: after it runs, print q.lastProgress['stateOperators'] -> numRowsTotal
    return q

• Build a Real-Time Streaming Project | Live Demo

• Streaming Systems O'Reilly · Book
• Stream Processing with Apache Spark O'Reilly · Book

Instrument the Phase 2 medallion pipeline for production observability: compute freshness/quality/quarantine SLAs into a metrics Delta table, write a strictly one-row-per-run audit log via idempotent MERGE, wire Databricks SQL alerts and Jobs failure notifications to thresholds, capture Unity Catalog column lineage for a PHI-adjacent field, build a small SLA/KPI dashboard, and write the RUNBOOK that closes the loop with a rehearsed time-travel RESTORE. This lab operates on your OWN project tables - you are not ingesting new raw data, you are making the existing platform observable and recoverable.

1. 1. Create an observability schema in Unity Catalog (e.g. ops.audit_log and ops.sla_metrics) with explicit typed schemas. Define your SLA thresholds as a small config table or dict: freshness_max_minutes, min_pricing_completeness_pass_rate, max_quarantine_rate, and a duplicate-collapse target. Write the numeric targets down - they are the contract the rest of the lab tests against.
2. 2. At the end of the silver/gold job, capture run context (run_id, job_id, task, start/end timestamps) and compute this run's metrics: rows_in, rows_out, rows_quarantined, pricing-completeness pass rate, duplicates collapsed, and freshness = now minus max source-file arrival watermark. Emit one metric-set for the run.
3. 3. Write the audit row with an idempotent MERGE keyed on run_id (never a blind append) so a retried or 10x-replayed job produces zero duplicate audit rows. Re-run the job 3-5 times and query the table to prove the count is stable - this is your break-safely receipt.
4. 4. Cross-check your hand-built audit numbers against the Lakeflow pipeline event_log (row counts, expectation pass/fail) and the Unity Catalog system tables (system.lakeflow.pipelines / system.access) so your audit log agrees with the platform's own telemetry; note any discrepancy in the architecture doc.
5. 5. In Catalog Explorer, open the lineage graph for the gold KPI table and drill to column lineage on a PHI-adjacent field (e.g. provider/NPI or organization name). Screenshot the upstream-to-downstream path and list every downstream consumer; save this as the governance evidence for the masking story.
6. 6. Build a Databricks SQL dashboard with a query over ops.sla_metrics: current freshness, pass rate, quarantine rate vs their thresholds (green/red), plus a 30-run trend. This is the single pane Maria's lead would glance at each morning.
7. 7. Create a Databricks SQL Alert on the quarantine-rate query (trigger when max quarantine_rate over the last run > threshold) and add Jobs notifications on_failure / on_duration_warning routed to email or a webhook. Force a breach (feed a deliberately dirty batch) and confirm the alert actually fires; capture the alert message.
8. 8. Rehearse recovery: run DESCRIBE HISTORY on a gold table, pick a known-good version, RESTORE TABLE ... VERSION AS OF, and write a NEW audit-log row recording the restore (who/when/from-version/why). Confirm the KPI returns to correct values.
9. 9. Write the RUNBOOK: SLA table, alert-to-action mapping, severity levels, the operate checklist, and the numbered recovery playbook (quarantine spike, late/failed load, bad-data rollback via time travel) - including the VACUUM retention caution. Add a 'Performance/Ops Notes' entry: what you now detect that was previously silent, with a before/after.

• Databricks - Capture and view data lineage using Unity Catalog
• Databricks SQL - Create alerts
• Databricks - Monitor Lakeflow Declarative Pipelines (event log / observability)
• Databricks Jobs - Add notifications to a job
• Databricks - Work with Delta table history and RESTORE (time travel)
• Databricks - Monitor data quality with Lakehouse Monitoring
• Databricks - Monitor usage with system tables

# ops/emit_run_audit.py
# STARTER SCAFFOLD - one audit row per run, written idempotently. Fill the TODOs.
from pyspark.sql import functions as F
from delta.tables import DeltaTable

# --- config (TODO: move thresholds to a real config table, not literals) ---
AUDIT_TABLE = "ops.audit_log"
SLA = {
    "freshness_max_minutes": 180,        # TODO: set from Maria's SLA
    "min_pass_rate": 0.95,               # pricing-completeness
    "max_quarantine_rate": 0.02,
}

# --- run context: prefer job task values / widgets over hardcoding ---
# TODO: read run_id/job_id/task from dbutils job context or task values
run_id = dbutils.widgets.get("run_id")   # TODO: wire real value


def compute_metrics(spark):
    # TODO: read THIS run's silver/gold + quarantine tables and compute:
    #   rows_in, rows_out, rows_quarantined, pass_rate, dupes_collapsed
    #   freshness_minutes = now - max(source_file_arrival_ts)   # watermark, not wall clock
    raise NotImplementedError("compute from your own tables")


def build_audit_row(spark, m):
    return spark.createDataFrame([{
        "run_id": run_id,
        "ended_at": None,                 # TODO: F.current_timestamp() at write
        # TODO: add started_at, rows_in/out, rows_quarantined, pass_rate,
        #       source_watermark, and a status derived from SLA checks below
        "status": None,                   # TODO: OK / SLA_BREACH / FAILED
    }])


def evaluate_sla(m):
    # TODO: return breaches list by comparing m against SLA dict
    #   (this is what your alert query will also key on)
    return []


def upsert_audit(spark, row_df):
    # IDEMPOTENT: MERGE on run_id so retries/replays never duplicate a row.
    tgt = DeltaTable.forName(spark, AUDIT_TABLE)
    (tgt.alias("t")
        .merge(row_df.alias("s"), "t.run_id = s.run_id")
        .whenMatchedUpdateAll()
        .whenNotMatchedInsertAll()
        .execute())
    # TODO: after wiring, re-run 3-5x and assert COUNT stays constant.


# TODO: also emit the same metrics into ops.sla_metrics for the dashboard/alert.
# TODO: on RESTORE recovery, call upsert_audit with a distinct run_id + reason.

• Mature Data Teams Don't Look Like Yours
• AIOps Is About to Explode—Here's Why
• Data-Maturity Tracker (free)

• Data Observability for Data Engineering O'Reilly · Book
• Data Quality Fundamentals O'Reilly · Book

Put the entire Phase 2 'Provider Pricing Integrity' pipeline under a real Git workflow: connect it as a Databricks Git folder backed by a GitHub repo, make a schema-touching change to silver validation on a feature branch, open a decision-log PR, run a data-specific review that catches three planted risks, then deliberately reproduce and remediate the 'manual change lost on deploy' trap. You end with branch protection + CODEOWNERS guarding the critical paths and a written root-cause note.

1. Create a GitHub repo for the project and connect it as a Databricks Git folder (Repos): commit the existing Phase 2 notebooks + config as notebooks-as-code on main. Verify the '# Databricks notebook source' .py serialization by viewing the raw files in GitHub -- confirm nothing lives only in the workspace UI.
2. Establish the branching + environment map: protect main, create a feature branch (feature/silver-npi-normalization), and confirm the branch runs against the DEV Unity Catalog (dev.silver) not prod. Document the branch->catalog mapping in the Architecture Doc.
3. Make one real, grain-touching change on the branch: tighten the silver NPPES validation (e.g., normalize/validate NPI as the composite key and add a Lakeflow expectation). Commit in small, well-messaged commits that reference the change's 'why'.
4. Plant three review targets to exercise the checklist: (a) a commit that changes the primary/merge key, (b) a commit that switches a Delta write from MERGE-by-key to append (non-idempotent), (c) a commit that loosens a data-quality expectation / silences a quarantine path. These simulate what a reviewer must catch.
5. Open a Pull Request whose body is a decision log: what changed, why, grain/schema impact, idempotency proof (rerun 10x -> row counts stable), quality/quarantine impact, and rollback plan (Delta time travel version). Link the PR to the Charter/Architecture Doc.
6. Run the data code review AS the reviewer using the checklist from lesson 4: comment on each of the three planted risks, request changes, and resolve them on the branch. Capture the review thread as evidence a human reviewed it.
7. Reproduce the trap: make a 'hotfix' edit directly in the workspace UI (outside the repo), then pull/redeploy from Git and watch the workspace edit get overwritten and lost. Screenshot before/after and write the 2-4 sentence root-cause note.
8. Remediate: add branch protection (require PR + 1 approval, no direct pushes to main) and a CODEOWNERS file mapping silver-validation, masking, and MDM-merge paths to a required reviewer. Re-attempt a direct push to main and confirm it is now blocked. Update the Runbook with the 'no workspace hotfixes -- branch + PR only' rule.
9. Merge the reviewed PR to main and confirm the change lands in dev; leave promotion-to-prod for the W39 CI/CD week (do not build the pipeline yet).

• Databricks docs — Git folders (Repos) for version control
• Databricks docs — Git operations with Git folders (branch, commit, PR, merge)
• Databricks docs — CI/CD techniques with Git folders and notebooks-as-code
• Databricks docs — Notebook source format (.py '# Databricks notebook source')
• GitHub docs — About code owners (CODEOWNERS)
• GitHub docs — About protected branches (require PR + review)

#!/usr/bin/env bash
# Scaffold: a short-lived feature-branch workflow for the running project.
# Run from the root of your Databricks Git folder / local clone.
set -euo pipefail

# --- config (edit these) -------------------------------------------
BRANCH="feature/silver-npi-normalization"   # TODO: name the branch after the change, not the person
DEV_CATALOG="dev"                            # TODO: confirm this maps to your dev Unity Catalog

# --- 1. start from an up-to-date main ------------------------------
git checkout main
git pull --ff-only

# --- 2. cut a short-lived feature branch ---------------------------
git checkout -b "$BRANCH"
# TODO: make ONE focused change (silver NPI validation/expectation).
#       Commit in small steps; each message says WHAT and WHY.

# --- 3. drop governance scaffolds (fill these in) ------------------
mkdir -p .github

cat > .github/CODEOWNERS <<'EOF'
# Require review on the critical, PHI-adjacent / grain-defining paths.
# TODO: replace @your-handle with the required reviewer(s).
# /src/silver/validation/**   @your-handle
# /src/silver/masking/**       @your-handle
# /src/mdm/merge/**            @your-handle
EOF

cat > .github/pull_request_template.md <<'EOF'
## What changed
<!-- TODO: one-line summary -->

## Why (decision log)
<!-- TODO: the business/technical reason — this becomes your interview answer -->

## Grain / schema impact
<!-- TODO: did the primary/merge key or schema change? -->

## Idempotency proof
<!-- TODO: MERGE/overwrite-by-key? paste 10x rerun row counts -->

## Quality / quarantine impact
<!-- TODO: expectations tightened or loosened? quarantine still loud? -->

## Rollback
<!-- TODO: Delta time-travel version to restore + audit-log note -->
EOF

# --- 4. push and open the PR ---------------------------------------
git add -A
git commit -m "scaffold: governance files for feature branch"  # TODO: better message
git push -u origin "$BRANCH"
# TODO: open the PR (gh pr create ...) and fill the template as a real decision log.

# --- 5. AFTER MERGE: turn on the guardrails ------------------------
# TODO: in GitHub settings, protect main: require PR + 1 approval, block direct pushes,
#       require CODEOWNERS review. Then verify a direct push to main is rejected.
echo "Branch $BRANCH ready. Now fill every TODO before requesting review."

• Head First Git O'Reilly · Book
• Introduction to Git and GitHub (Google) Coursera · Course
• Intermediate Git DataCamp · Course

Wire the running Provider Pricing Integrity repo (from W38) with a real CI/CD pipeline: a PR-triggered GitHub Actions job that runs pytest + DQ contract checks and blocks merge on red, followed by a merge-triggered CD job that promotes across three Unity Catalog environments (dev -> staging -> prod) using service-principal secrets and a manual approval gate before prod. You lead with the business outcome ('a broken transform can no longer reach the compliance mart') and prove the path is idempotent and rollback-ready.

1. Set the contract: on the repo's main branch enable branch protection with 'Require a pull request before merging' + 'Require status checks to pass' (you will select the CI job once it has run once). Create GitHub Environments named 'staging' and 'prod'; add a required-reviewer protection rule to 'prod'. Add per-environment secrets DATABRICKS_HOST and DATABRICKS_TOKEN backed by a Databricks SERVICE PRINCIPAL (not your PAT).
2. Author .github/workflows/ci.yml triggered on pull_request: checkout, setup-python, cache pip, install requirements, then run `pytest` over the transform unit tests using tiny committed CSV/JSON fixtures (a handful of NPPES + CMS rows) so no cluster is needed. Add a second step that runs the DQ/contract checks (schema match, silver grain uniqueness, expectation pass-rate threshold) and fails the job on any breach.
3. Open a PR that intentionally corrupts a silver transform (e.g., drops the dedup key) and confirm the CI check goes red and the Merge button is blocked; fix it, watch it go green, then merge — capture both screenshots for the deliverable.
4. Add a deploy-staging job that runs on push to main (needs: build): use the databricks/setup-cli action + service-principal secrets to sync the repo to a staging workspace path and run a validation job against the staging_pricing catalog, asserting KPIs (pricing-completeness pass rate, duplicate providers collapsed) meet thresholds; fail the job if any KPI misses.
5. Add a promote-prod job that declares `environment: prod` so it PAUSES for the required-reviewer approval, then runs the same validation job against prod_pricing; on validation failure it must NOT promote (leave prod on the last good version).
6. Prove idempotency + recovery: re-run the staging deploy twice and confirm no duplicate rows / side effects (MERGE-based, not append), then record in the RUNBOOK the exact Delta time-travel rollback command (`RESTORE TABLE ... TO VERSION AS OF n`) and how to read the audit log to pick the version.
7. Measure & tune: note the CI wall-clock time before caching, add pip + fixture caching, re-run, and record the before/after in the Performance Notes README section, leading with the outcome ('PR feedback in ~3 min keeps reviewers in flow').
8. Document: write the CI/CD section of the RUNBOOK (what green vs red means, who approves prod, how to roll back) and a short 'why' note defending your secrets model and 3-catalog promotion choice as interview answers.

• Databricks — CI/CD on Databricks (overview)
• Databricks — Continuous integration and delivery using GitHub Actions
• databricks/setup-cli — official GitHub Action to install the Databricks CLI
• Databricks — Manage service principals (for CI/CD auth)
• Databricks — What is Unity Catalog (catalogs as dev/staging/prod isolation)
• GitHub Docs — Using environments for deployment (environment secrets + required-reviewer protection rules)
• GitHub Docs — About protected branches (required status checks)

name: ci-cd

# TODO: run CI on every PR; run CD only on merge to main
on:
  pull_request:
    branches: [ main ]
  push:
    branches: [ main ]

jobs:
  build:                      # ---- CI: must be GREEN before merge ----
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
          cache: "pip"        # TODO: measure CI time before/after enabling cache -> Performance Notes
      - name: Install deps
        run: pip install -r requirements.txt
      - name: Unit tests (transform logic on committed fixtures)
        run: echo "TODO: pytest tests/unit -q  # pure PySpark logic, no cluster"
      - name: DQ / data-contract checks
        run: echo "TODO: run schema + grain-uniqueness + expectation pass-rate checks; exit non-zero on any breach"

  deploy-staging:             # ---- CD: promote on green merge ----
    needs: build
    if: github.event_name == 'push'
    runs-on: ubuntu-latest
    environment: staging       # pulls staging secrets
    steps:
      - uses: actions/checkout@v4
      - uses: databricks/setup-cli@main
      - name: Validate against staging catalog
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}   # service principal, NEVER echoed
        run: echo "TODO: databricks bundle/CLI -> run validation job on staging_pricing; assert KPIs meet thresholds"

  promote-prod:
    needs: deploy-staging
    if: github.event_name == 'push'
    runs-on: ubuntu-latest
    environment: prod          # protection rule PAUSES here for required-reviewer approval
    steps:
      - uses: actions/checkout@v4
      - uses: databricks/setup-cli@main
      - name: Validate + promote to prod catalog
        env:
          DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
          DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
        run: echo "TODO: run validation on prod_pricing; if KPIs fail, DO NOT promote (leave last good version)"

• Your Production and Test Environments Are Drifting Apart

• GitHub Actions - The Complete Guide (Maximilian Schwarzmuller) Udemy · Course
• Learning GitHub Actions O'Reilly · Book
• Continuous Integration and Continuous Delivery (CI/CD) (IBM) Coursera · Course

Take the ENTIRE Provider Pricing Integrity project — the Lakeflow declarative pipeline, the Databricks Job that orchestrates ingest→bronze→silver→MDM→gold, its cluster/serverless config, and its Unity Catalog grants — and express all of it as a single Databricks Asset Bundle. Define dev and prod targets with per-environment config, deploy to dev with zero manual drift, prove two developers can deploy in parallel without collision, and hand the prod deploy off to the W39 GitHub Actions workflow running as a service principal so a merge to main promotes the change by code with an auditable run link. No new raw data this week — you are IaC-ifying the platform you already built.

1. Inventory first (measure before you change): install/authenticate the Databricks CLI, then snapshot what exists today — `databricks jobs list`, `databricks pipelines list`, and the UC grants on your project schema — into a `pre_deploy_inventory.md`. This is your drift baseline; the acceptance test compares against it.
2. Scaffold the bundle: run `databricks bundle init` (default-python or an empty template), then replace the generated resources so `databricks.yml` declares your project's `bundle.name`, an `include` for resource files, and a `resources` block referencing your existing Lakeflow pipeline definition and Job. Keep source notebooks/py files where they are; the bundle points at them, it does not rewrite them.
3. Parametrize with variables and substitutions: add a `variables` block (catalog, schema/prefix, warehouse or cluster size, notification email) and replace every hard-coded catalog/schema/path in the resource definitions with `${var.*}` and `${workspace.*}` references. Nothing environment-specific should be a literal.
4. Define targets: create a `dev` target with `mode: development` (auto name-prefixing, paused schedule, your user as run-as) and a `prod` target with `mode: production`, a service-principal `run_as`, the prod catalog/schema, production cluster/serverless sizing, and an active schedule. Set `default: true` on dev.
5. Validate and diff environments: run `databricks bundle validate` until it returns zero errors, then run `databricks bundle validate -o json --target dev` vs. `--target prod` and diff the resolved configs to PROVE the same source yields two correct, isolated environments (different catalog, run-as, schedule state, cluster size).
6. Deploy to dev and confirm zero drift: `databricks bundle deploy --target dev`, then `databricks bundle summary`. Re-list jobs/pipelines/grants and diff against `pre_deploy_inventory.md` — every project object in the workspace must now be bundle-managed (dev-prefixed), with nothing orphaned or hand-created. Trigger `databricks bundle run <job_key>` and confirm the pipeline executes.
7. Prove collision-free parallel dev: from a second identity or with `DATABRICKS_BUNDLE_ENV`/a second `--target`, deploy again and show the dev-mode name prefix isolates the two deployments (two `[dev you] provider_pricing_job` vs. `[dev teammate] ...`) with no shared-object clobbering.
8. Wire prod into W39 CI/CD: extend the existing GitHub Actions workflow so PRs run `databricks bundle validate` (fail the check on errors) and merge-to-main runs `databricks bundle deploy --target prod` authenticated as a service principal via OIDC or a stored secret. Open a trivial PR (e.g., bump a variable default), let it deploy prod, and capture the Actions run URL.
9. Break-and-recover drill + write the 'why': manually edit the deployed dev job in the UI (change its schedule), re-run `databricks bundle deploy`, and observe the bundle reconcile the drift back to code. Record before/after and write a 1-paragraph decision note on why Git-is-source-of-truth beats click-ops for a compliance-sensitive pipeline.

• What are Databricks Asset Bundles? (official docs)
• Databricks Asset Bundle configuration reference (databricks.yml, resources, targets)
• Bundle deployment modes (development vs. production)
• Substitutions and variables in Databricks Asset Bundles
• Run a CI/CD workflow with a Databricks Asset Bundle and GitHub Actions
• Databricks Asset Bundles resources (jobs, pipelines, schemas, grants)

# databricks.yml — Provider Pricing Integrity bundle (STARTER SCAFFOLD)
# Fill every TODO from YOUR existing project. Run `databricks bundle validate` after each edit.

bundle:
  name: provider-pricing-integrity   # TODO: keep stable; this names deployed resources

# Split resource defs into files under ./resources/*.yml and glob them in.
include:
  - resources/*.yml                  # TODO: create resources/pipeline.yml and resources/job.yml

variables:
  catalog:
    description: Unity Catalog for this environment
    # TODO: no default that leaks prod; set per-target below
  schema:
    description: Target schema / dataset
  notify_email:
    description: Alert recipient for job/pipeline failures
  # TODO: add variables for cluster size / warehouse id / landing volume path

# resources/*.yml will define these — referenced here only to show wiring.
# resources:
#   pipelines:
#     provider_pricing_pipeline:
#       name: ${bundle.name}-lakeflow
#       catalog: ${var.catalog}
#       target: ${var.schema}
#       # TODO: point libraries at your EXISTING silver/MDM/gold notebooks — do not rewrite them
#   jobs:
#     provider_pricing_job:
#       name: ${bundle.name}-orchestrator
#       # TODO: tasks = ingest -> pipeline update -> gold KPIs, as in Phase 2
#       email_notifications:
#         on_failure: [ "${var.notify_email}" ]
#   grants: []                       # TODO: declare UC grants here so access is reviewed in Git

targets:
  dev:
    mode: development                # prefixes names, pauses schedules, run-as = you
    default: true
    workspace:
      host: ${workspace.host}         # TODO: or set DATABRICKS_HOST via profile
    variables:
      catalog: dev_provider_pricing   # TODO: your dev catalog
      schema: maria_dev
      notify_email: you@example.com

  prod:
    mode: production                 # locked run-as, no prefixing, enforced deploy path
    workspace:
      host: ${workspace.host}
      root_path: /Workspace/Shared/.bundle/${bundle.name}/prod  # TODO: confirm prod path
    run_as:
      service_principal_name: TODO-PROD-SP-APPLICATION-ID       # TODO: prod SP, not your token
    variables:
      catalog: prod_provider_pricing  # TODO: your prod catalog
      schema: maria
      notify_email: TODO-oncall@example.com
    # TODO: override cluster/serverless sizing and schedule (active in prod, paused in dev)

• Databricks Certified Data Engineer Professional - Preparation (Derar Alhussein) Udemy · Course
• Terraform: Up and Running, 3rd Ed. O'Reilly · Book
• DevOps Concepts DataCamp · Course

Harden the running Provider Pricing Integrity catalog with Unity Catalog governance: build a data-classification table, stand up analyst vs engineer groups with least-privilege grants, apply column masks (hashed NPI/TIN, redacted address) and a row filter (analysts see only their assigned hospitals), swap the nightly Job/DAB from a personal PAT to a service principal, then pull system-table audit evidence proving the controls work. Every control is verified by querying as each persona and diffing — a control you cannot demonstrate does not exist.

1. Classify first. Create `governance.data_classification` (column, table, sensitivity_level, rule, rationale) and populate it for every silver/gold column touching NPI, provider_name, tin, address, hospital_id. This table is the single source of truth that drives every mask below — recognition before evaluation, and your written 'why'.
2. Stand up personas. Create two account groups, `ppi_analysts` and `ppi_engineers`, and add yourself to each via separate test principals (or use is_account_group_member() with a swappable test group). Document in the runbook which real-world role each maps to.
3. Grant least privilege. Write the three-level GRANT chain so analysts get USE CATALOG + USE SCHEMA + SELECT on the gold marts ONLY; engineers additionally get SELECT on silver and bronze/quarantine. Prove least privilege: run a SELECT on `bronze.raw_pricing` as an analyst and capture the PERMISSION_DENIED error as evidence.
4. Mask sensitive columns. Create UC SQL UDF masks: sha2(npi,256) and sha2(tin,256) for analysts, cleartext for engineers via is_account_group_member('ppi_engineers'); redact street address to ZIP-only. ALTER TABLE ... ALTER COLUMN ... SET MASK on the gold provider mart. Verify by SELECTing the same rows as each persona and diffing the two result sets.
5. Filter rows. Create a row-filter UDF so `ppi_analysts` see only rows whose hospital_id is in an assignment table (governance.analyst_scope), while engineers see all rows. Apply with ALTER TABLE ... SET ROW FILTER and verify row counts differ by persona. Confirm masks + filter compose (an analyst gets fewer rows AND hashed IDs).
6. Idempotency + recovery check. Re-run the entire policy notebook a second and third time; confirm CREATE OR REPLACE FUNCTION and SET MASK are idempotent (no duplicate policies, no drift). Note in the runbook how to remove a mask (UNSET MASK) for an emergency break-glass, and who authorizes it.
7. Kill the personal token. Create a service principal, grant it exactly the pipeline's needs (USE on catalog/schema, SELECT/MODIFY on the tables it writes, no ALL PRIVILEGES), set the Job/DAB run-as to the SP, and configure OAuth M2M so no PAT is stored. Trigger the job and confirm it runs as the SP identity.
8. Prove it with audit. Query system.access.audit filtered to your catalog for the last day: show the analyst's masked SELECT, the denied bronze read, and the SP-run job. Join to column lineage to answer 'who read npi and were they masked'. Export a one-page 'Access Evidence' summary into the RUNBOOK.
9. Write the why. Add a 'Security & Governance Decisions' section to the ARCHITECTURE DOC: why hashing over tokenization here, why masks over separate restricted tables, why an SP over a PAT, and the one control you deliberately did NOT add and why (scope discipline — avoid over-engineering).

• Databricks — Filter sensitive table data using row filters and column masks
• Databricks — Manage privileges in Unity Catalog (GRANT, the three-level privilege chain, securable hierarchy)
• Databricks — Unity Catalog privileges and securable objects reference
• Databricks — Manage service principals
• Databricks — Authorize unattended access with OAuth machine-to-machine (M2M) service principals
• Databricks — Run a job as a service principal (run-as identity)
• Databricks — Monitor access with the audit log system table (system.access.audit)
• Databricks — Manage groups (account groups for RBAC)

-- Provider Pricing Integrity :: W41 governance scaffold
-- Idempotent: safe to re-run. CREATE OR REPLACE + SET MASK do not stack.
-- Prereq: catalog/schema from Phase 2 already exist under Unity Catalog.

-- 0) Config -----------------------------------------------------------------
-- TODO: set to YOUR account group names created in step 2
-- e.g. ppi_engineers (cleartext), ppi_analysts (masked, scoped)

-- 1) Column mask: cleartext for engineers, hash for everyone else ----------
CREATE OR REPLACE FUNCTION governance.mask_identifier(id STRING)
RETURN CASE
  -- TODO: replace group literal with your engineer group name
  WHEN is_account_group_member('ppi_engineers') THEN id
  -- TODO: decide hash vs redaction vs tokenization and justify in ARCH doc
  ELSE sha2(id, 256)
END;

-- TODO: create a second mask for address -> ZIP-only (redaction pattern)
-- CREATE OR REPLACE FUNCTION governance.mask_address(addr STRING) RETURN ...;

-- 2) Row filter: analysts limited to assigned hospitals --------------------
-- Assumes governance.analyst_scope(hospital_id) lists in-scope hospitals.
CREATE OR REPLACE FUNCTION governance.filter_by_scope(hospital_id STRING)
RETURN
  is_account_group_member('ppi_engineers')  -- engineers: all rows
  -- TODO: analysts -> only rows in governance.analyst_scope
  OR hospital_id IN (SELECT s.hospital_id FROM governance.analyst_scope s);

-- 3) Bind policies to the gold provider mart -------------------------------
-- TODO: point at YOUR gold table; add every column from data_classification
ALTER TABLE gold.provider_mart
  ALTER COLUMN npi SET MASK governance.mask_identifier;
-- TODO: ALTER COLUMN tin SET MASK governance.mask_identifier;
-- TODO: ALTER COLUMN address SET MASK governance.mask_address;
ALTER TABLE gold.provider_mart
  SET ROW FILTER governance.filter_by_scope ON (hospital_id);

-- 4) Verify (run interactively as each persona; do NOT skip) ---------------
-- As analyst:  SELECT npi, tin, address, hospital_id FROM gold.provider_mart LIMIT 20;
-- As engineer: SELECT npi, tin, address, hospital_id FROM gold.provider_mart LIMIT 20;
-- TODO: capture both outputs + row counts as your control evidence.

-- Break-glass (document who authorizes; do not leave applied):
-- ALTER TABLE gold.provider_mart ALTER COLUMN npi UNSET MASK;

• Data Governance: The Definitive Guide O'Reilly · Book
• Data Governance Concepts DataCamp · Course

Fork your Provider Pricing Integrity project onto ONE cloud (Azure or AWS). Provision cloud-native storage for the medallion layers, connect it to Unity Catalog through a keyless storage credential + external location, land the CMS/NPPES raw files there, verify the pipeline reads/writes the external path, then produce a per-run DBU + dollar cost baseline from system tables. You end with a governed, keyless, cost-measured cloud foundation and an updated architecture doc that defends the choice.

1. CHOOSE YOUR TRACK and record why (data residency, employer's cloud, cost) in one paragraph of the architecture doc. Everything below has an Azure branch and an AWS branch — do exactly one.
2. Provision storage: Azure — create an ADLS Gen2 storage account with hierarchical namespace ON, add a container (e.g. ppi-lakehouse) with bronze/silver/gold folders. AWS — create an S3 bucket (e.g. ppi-lakehouse-<acct>) with the same prefixes, versioning ON, public access fully blocked, default encryption ON.
3. Create the keyless identity: Azure — create/assign a user-assigned managed identity (or Access Connector for Databricks) and grant it Storage Blob Data Contributor on the container only. AWS — create an IAM role with a trust policy for the Databricks account and an inline policy scoped to just that bucket/prefixes; attach it as an instance profile or self-assuming role. Copy NO keys anywhere.
4. Wire Unity Catalog to storage: create a STORAGE CREDENTIAL that references the managed identity / IAM role, then an EXTERNAL LOCATION pointing at the abfss:// or s3:// path. GRANT READ FILES / WRITE FILES to your project group only. (Use the starter scaffold's TODOs.)
5. Prove access + least privilege: from a notebook, LIST FILES on the external location and write a tiny Delta test table to bronze/_probe, then try to read a path OUTSIDE the granted container/bucket and capture the PERMISSION DENIED — screenshot both; the denial is the compliance evidence.
6. Land the real data: pull the NPPES bulk file and a couple of CMS Hospital Price Transparency machine-readable files into the bronze external path (upload or a notebook download-then-write). Point your existing bronze table's LOCATION at this external path so the Phase 2 pipeline now runs on cloud-native, UC-governed storage.
7. Re-run the bronze->silver step once end to end on a job cluster to confirm the fork works, and TAG the cluster/job with project=ppi and env=prod so the run is cost-attributable.
8. Measure the bill: query system.billing.usage joined to system.billing.list_prices, filtered to your tag and run window, to compute DBUs and estimated dollars for that one run. Record the number and a target (e.g. 'cut ~15-20% in the W44 tuning week') in a new PERFORMANCE NOTES > Cost Baseline subsection.
9. Document the topology: add a networking paragraph to the architecture doc (control plane vs your data plane, VNet-injection/SCC or VPC/PrivateLink intent, storage firewall posture) and note it as the compliance answer for PHI-adjacent data.

• Databricks — Connect to cloud object storage using Unity Catalog (storage credentials & external locations)
• Azure Databricks — Create a storage credential for connecting to ADLS Gen2 (managed identity / access connector)
• Databricks on AWS — Configure a storage credential and external location with an IAM role
• Databricks — Monitor usage with system tables (system.billing.usage and list_prices)
• Azure Databricks — Secure cluster connectivity and VNet injection (networking)
• Databricks on AWS — Customer-managed VPC and PrivateLink (networking)

-- setup_cloud_foundation.sql
-- PPI project: keyless UC access to cloud storage + a cost baseline.
-- Run as a Unity Catalog admin / metastore admin on a UC-enabled cluster.

-- ============ 1. STORAGE CREDENTIAL (pick ONE branch) ============
-- Azure branch: reference a user-assigned managed identity / Access Connector.
-- CREATE STORAGE CREDENTIAL ppi_cred
--   WITH AZURE_MANAGED_IDENTITY (
--     ACCESS_CONNECTOR_ID = '/subscriptions/<TODO>/resourceGroups/<TODO>/providers/Microsoft.Databricks/accessConnectors/<TODO>'
--   )
--   COMMENT 'Keyless access to ppi-lakehouse';

-- AWS branch: reference an IAM role ARN (trust policy must allow Databricks).
-- CREATE STORAGE CREDENTIAL ppi_cred
--   WITH IAM_ROLE ( 'arn:aws:iam::<TODO-acct>:role/<TODO-role>' )
--   COMMENT 'Keyless access to ppi-lakehouse';

-- NOTE: no account key / access key appears anywhere. If you typed a secret, stop and redo.

-- ============ 2. EXTERNAL LOCATION -> your storage path ============
-- Azure path form:  abfss://<container>@<account>.dfs.core.windows.net/<prefix>
-- AWS path form:    s3://<bucket>/<prefix>
-- CREATE EXTERNAL LOCATION ppi_bronze
--   URL '<TODO storage url>'
--   WITH (STORAGE CREDENTIAL ppi_cred)
--   COMMENT 'Bronze landing zone';

-- ============ 3. LEAST-PRIVILEGE GRANTS ============
-- GRANT READ FILES, WRITE FILES ON EXTERNAL LOCATION ppi_bronze TO `<TODO project group>`;
-- TODO: do NOT grant to account users; scope to the project group only.

-- ============ 4. PROVE ACCESS (run, screenshot) ============
-- LIST '<TODO storage url>';
-- TODO: then attempt a path OUTSIDE the granted location and capture PERMISSION DENIED.

-- ============ 5. COST BASELINE (measure the bill) ============
-- TODO: set the run window + your tag values, then compute DBUs and est. dollars.
-- SELECT u.sku_name,
--        SUM(u.usage_quantity)                         AS dbus,
--        SUM(u.usage_quantity * p.pricing.default)     AS est_usd
-- FROM system.billing.usage u
-- JOIN system.billing.list_prices p
--   ON u.sku_name = p.sku_name
--  AND u.usage_end_time BETWEEN p.price_start_time
--                           AND COALESCE(p.price_end_time, current_timestamp())
-- WHERE u.custom_tags['project'] = 'ppi'      -- TODO confirm your tag key
--   AND u.usage_date >= '<TODO run date>'
-- GROUP BY u.sku_name
-- ORDER BY est_usd DESC;
-- TODO: paste the total into architecture doc > Performance Notes > Cost Baseline, with a reduction target.

• Pick the Right Data Platform (Before It's Too Late)
• Watch This Before Choosing a Data Platform

• Understanding Microsoft Azure DataCamp · Course
• AWS Cloud Technology and Services DataCamp · Course
• Azure Databricks & Spark for Data Engineers (Ramesh Retnasamy) Udemy · Course

This is the release week: no new sources, no new features — you integrate everything Phase 3 built and prove it operates and recovers as one governed, observable platform. You do a cold-start end-to-end run on a clean prod-target, inject a failure to prove the pipeline breaks loudly (quarantine + reason + timestamp + alert) and reruns idempotently, run a recovery drill with Delta time travel, finalize Performance Notes and the runbook, then cut a tagged release through GitHub Actions and assemble the portfolio so Project 1 reads as senior on a 30-second skim.

1. FREEZE & VALIDATE: cut a `release/v1.0.0` branch. From a clean checkout run `databricks bundle validate --target prod` and confirm every bundle resource resolves — the Lakeflow pipeline, the orchestrating Job, UC catalog/schema and grants, and the service principal run-as. Record the git SHA; do not deploy yet.
2. COLD-START DEPLOY: `databricks bundle deploy --target prod` into an empty prod-like workspace/schema so you prove nothing depends on hand-created state. Seed the Auto Loader landing volume with one fresh CMS Hospital Price Transparency file and emit one batch of provider-change events, then trigger the Job.
3. OBSERVE & PROVE QUALITY: watch the Lakeflow pipeline event log and expectation metrics; confirm the KPI mart lands inside the freshness SLA. Inject one deliberately malformed row and confirm it lands in quarantine WITH reason + timestamp and fires the alert — loud, not silent.
4. PROVE IDEMPOTENCY: re-drop the exact same CMS file and re-run the Job 3× (aim for your standing 10× discipline). Assert zero net change in silver/golden via the watermarked, composite-key MERGE — no duplicate providers, quarantine counts stable.
5. MEASURE FIRST, THEN TUNE: open the Spark UI / query profile for the slowest gold aggregation, identify the ONE bottleneck stage (skew, wide scan, or a big-vs-small join), apply exactly one fix (liquid clustering by query pattern / broadcast / repartition), and record before/after timing in Performance Notes led by the business outcome ("Maria's KPI mart now lands before 8am").
6. GOVERNANCE AUDIT: review UC grants against least-privilege, confirm PHI-adjacent columns are masked (column mask / row filter) for the analyst role, and capture a Catalog Explorer lineage view end-to-end for the architecture doc's governance section.
7. RECOVERY DRILL: simulate a bad load into the golden table, then restore it using ONLY runbook commands — `DESCRIBE HISTORY`, Delta time travel (`VERSION AS OF` / `RESTORE`) cross-checked against the audit log — and note the VACUUM retention that keeps the drill possible. Time the recovery and write it down.
8. CUT THE RELEASE: open the PR to `main`; GitHub Actions runs `bundle validate` → deploy to prod → smoke run as the service principal. On green, tag `v1.0.0` and confirm the Actions badge and job run history show the automated, run-as-SP execution.
9. ASSEMBLE & FRESH-EYES DRY RUN: finalize the README (architecture diagram, KPI table with real numbers, CI badge, links to charter / 4-part architecture doc / Performance Notes / runbook / decision log). Then have a peer (or yourself, cold) deploy and recover using only the runbook; fix any gap they hit and capture final SLA evidence.

• Databricks Asset Bundles (development, targets, and deployment)
• CI/CD for Databricks with Asset Bundles and GitHub Actions
• Lakeflow Declarative Pipelines — manage data quality with pipeline expectations
• Unity Catalog — filter sensitive data using row filters and column masks
• Delta Lake — work with table history and time travel (RESTORE)
• Optimize with predictive optimization and liquid clustering; VACUUM
• Databricks Jobs — run jobs as a service principal

# STARTER SCAFFOLD — structure + config only. Fill every TODO yourself.
# Purpose: promote the Provider Pricing Integrity bundle to prod and tag a release.
# This is NOT a full pipeline — it just wires validate -> deploy -> smoke.

name: release

on:
  push:
    tags:
      - "v*"          # TODO: confirm your tag convention (e.g. v1.0.0)

# TODO: set these as GitHub repo secrets — never inline them here
env:
  DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
  # Auth as a SERVICE PRINCIPAL (OAuth M2M), not your personal PAT:
  DATABRICKS_CLIENT_ID: ${{ secrets.DATABRICKS_CLIENT_ID }}
  DATABRICKS_CLIENT_SECRET: ${{ secrets.DATABRICKS_CLIENT_SECRET }}

jobs:
  release:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      # TODO: install the Databricks CLI (pin a version for reproducibility)
      - name: Install Databricks CLI
        run: |
          echo "TODO: curl the pinned CLI installer and verify version"

      # Gate 1 — never deploy an invalid bundle
      - name: Validate bundle (prod target)
        run: |
          echo "TODO: databricks bundle validate --target prod"

      # Gate 2 — deploy the frozen release to prod
      - name: Deploy bundle (prod target)
        run: |
          echo "TODO: databricks bundle deploy --target prod"

      # Gate 3 — smoke run: trigger the job, fail the release if it fails
      - name: Smoke run
        run: |
          echo "TODO: databricks bundle run <job_key> --target prod"
          echo "TODO: assert freshness SLA + expectations pass, else exit 1"

      # TODO: on failure, post to the same alert channel your pipeline uses
      # TODO: on success, this tagged, green run IS the release record

• Managed vs External Tables in Databricks: The $15K Mistake

• Data Engineering Design Patterns O'Reilly · Book
• Databricks Certified Data Engineer Professional - Preparation (Derar Alhussein) Udemy · Course