# Install Databricks CLI
pip install databricks-cli

# Configure with your workspace
databricks configure --token
# Enter: Host = https://<your-workspace>.azuredatabricks.net
# Enter: Token = your_personal_access_token

# List repos
databricks repos list

# Create a repo (clone from GitHub)
databricks repos create \
  --url https://github.com/myorg/etl-project \
  --provider gitHub \
  --path /Repos/alice@company.com/etl-project

import requests

WORKSPACE = "https://<your-workspace>.azuredatabricks.net"
TOKEN = "your_pat_token"

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

# List all repos
resp = requests.get(
    f"{WORKSPACE}/api/2.0/repos",
    headers=headers
)
print(resp.json())

# Pull latest changes (update repo to latest branch HEAD)
repo_id = "12345"
resp = requests.patch(
    f"{WORKSPACE}/api/2.0/repos/{repo_id}",
    headers=headers,
    json={"branch": "main"}
)
print("Repo updated:", resp.json())

{
  "cluster_name": "dev-interactive",
  "spark_version": "14.3.x-scala2.12",
  "node_type_id": "Standard_DS3_v2",
  "num_workers": 2,
  "autotermination_minutes": 30,
  "spark_conf": {
    "spark.sql.shuffle.partitions": "200",
    "spark.databricks.io.cache.enabled": "true"
  }
}

# Format: {DBR_version}.x-scala{scala_version}
"14.3.x-scala2.12"
#  │    │  └─ Scala 2.12 (most common, Scala 2.13 also available)
#  │    └──── .x = latest patch within DBR 14.3
#  └───────── DBR major.minor version

# ML Runtime (includes ML libraries like TensorFlow, PyTorch)
"14.3.x-cpu-ml-scala2.12"

# GPU Runtime (for deep learning)
"14.3.x-gpu-ml-scala2.12"

# Long Term Support marker
# LTS versions: 12.2 LTS, 13.3 LTS, 14.3 LTS (check Databricks docs for current)

# %python — run Python (default)
%python
df = spark.read.csv("/data/sales.csv", header=True)
df.show(5)

---
# %sql — run Spark SQL
%sql
SELECT customer_id, SUM(amount) FROM sales GROUP BY customer_id

---
# %scala — run Scala
%scala
val df = spark.read.csv("/data/sales.csv")

---
# %sh — run shell commands on driver node
%sh
ls /dbfs/data/
pip show pandas

---
# %fs — Databricks File System (DBFS) commands
%fs
ls /data/
head /data/sample.csv

---
# %md — Markdown for documentation
%md
## Pipeline Step 1: Load Data
This cell loads raw CSV data from DBFS into a Spark DataFrame.

---
# %run — run another notebook (import its definitions)
%run ./utils/common_functions  # relative path
%run /Repos/myrepo/utils/helpers  # absolute path

# ─── File System (dbutils.fs) ─────────────────────────
# List files
files = dbutils.fs.ls("dbfs:/data/")
for f in files:
    print(f.name, f.size)

# Copy files
dbutils.fs.cp("dbfs:/raw/file.csv", "dbfs:/processed/file.csv")

# Move files
dbutils.fs.mv("dbfs:/staging/file.csv", "dbfs:/archive/file.csv")

# Delete files
dbutils.fs.rm("dbfs:/temp/file.csv", recurse=True)

# ─── Secrets (dbutils.secrets) ────────────────────────
# Read a secret from a Databricks secret scope
db_password = dbutils.secrets.get(scope="production", key="db_password")
api_key = dbutils.secrets.get(scope="production", key="api_key")

# Secrets are redacted in notebook output — [REDACTED]
print(db_password)  # prints: [REDACTED]

# ─── Notebook utilities (dbutils.notebook) ────────────
# Run another notebook and get its return value
result = dbutils.notebook.run(
    "/Repos/myrepo/etl/process_orders",
    timeout_seconds=3600,
    arguments={"date": "2024-01-15", "env": "prod"}
)
print(f"Child notebook returned: {result}")

# Exit the current notebook with a return value
dbutils.notebook.exit("SUCCESS: 1000 rows processed")

# ─── Widgets (dbutils.widgets) ────────────────────────
# Create input widgets (covered in next subtopic)
dbutils.widgets.text("date", "2024-01-01", "Processing Date")
date_param = dbutils.widgets.get("date")

{
  "name": "Daily Sales ETL",
  "job_clusters": [{
    "job_cluster_key": "etl-cluster",
    "new_cluster": {
      "spark_version": "14.3.x-scala2.12",
      "node_type_id": "Standard_DS4_v2",
      "num_workers": 4
    }
  }],
  "tasks": [
    {
      "task_key": "extract_raw_data",
      "notebook_task": {
        "notebook_path": "/Repos/myrepo/etl/01_extract",
        "base_parameters": {"date": "{{job.start_time.iso_date}}"}
      },
      "job_cluster_key": "etl-cluster"
    },
    {
      "task_key": "transform_to_silver",
      "depends_on": [{"task_key": "extract_raw_data"}],
      "notebook_task": {
        "notebook_path": "/Repos/myrepo/etl/02_transform"
      },
      "job_cluster_key": "etl-cluster"
    },
    {
      "task_key": "send_alert",
      "depends_on": [{"task_key": "transform_to_silver"}],
      "python_wheel_task": {
        "package_name": "my_etl_package",
        "entry_point": "send_completion_alert"
      },
      "job_cluster_key": "etl-cluster"
    }
  ],
  "schedule": {
    "quartz_cron_expression": "0 0 6 * * ?",  // Daily at 6 AM UTC
    "timezone_id": "UTC"
  },
  "email_notifications": {
    "on_failure": ["data-team@company.com"],
    "on_success": ["data-team@company.com"]
  },
  "max_concurrent_runs": 1
}

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import RunNowRequest

# Initialize Databricks SDK client
w = WorkspaceClient(
    host="https://<your-workspace>.azuredatabricks.net",
    token="your_pat_token"
)

# Trigger an existing job with parameters
run = w.jobs.run_now(
    job_id=123456,
    notebook_params={
        "date": "2024-01-15",
        "env": "prod"
    }
)
print(f"Job run started: run_id={run.run_id}")

# Wait for completion
result = w.jobs.wait_get_run_job_terminated_or_skipped(run_id=run.run_id)
print(f"Job finished with state: {result.state.result_state}")

# ─── Create catalog, schema, table ───────────────────
spark.sql("CREATE CATALOG IF NOT EXISTS production")
spark.sql("CREATE SCHEMA IF NOT EXISTS production.sales")

spark.sql("""
  CREATE TABLE IF NOT EXISTS production.sales.orders (
    order_id     BIGINT,
    customer_id  BIGINT,
    order_date   DATE,
    amount       DECIMAL(10,2),
    status       STRING
  )
  USING DELTA
  COMMENT 'Fact table for all customer orders'
""")

# ─── Set default catalog/schema for the session ──────
spark.sql("USE CATALOG production")
spark.sql("USE SCHEMA sales")

# Now you can use 1-part names (within sales schema)
df = spark.table("orders")         # = production.sales.orders
df2 = spark.table("customers")

# Always use 3-part names in production for clarity
df = spark.table("production.sales.orders")

# ─── Read/write with full 3-part names ───────────────
df.write.mode("append").saveAsTable("production.sales.orders")

# ─── List catalogs/schemas/tables ────────────────────
spark.sql("SHOW CATALOGS").show()
spark.sql("SHOW SCHEMAS IN production").show()
spark.sql("SHOW TABLES IN production.sales").show()

-- Create a managed volume (Databricks manages the storage location)
CREATE VOLUME production.sales.raw_files;

-- Create an external volume (you specify the storage path)
CREATE EXTERNAL VOLUME production.sales.landing_zone
  LOCATION 's3://my-bucket/landing/'
  COMMENT 'Landing zone for incoming CSV files';

-- Access volume files via /Volumes path
-- Path format: /Volumes/catalog/schema/volume_name/...

# Read CSV from a Unity Catalog Volume
df = spark.read.csv(
    "/Volumes/production/sales/landing_zone/2024-01-15/orders.csv",
    header=True,
    inferSchema=True
)

# Write processed data back to a volume
df.write.parquet(
    "/Volumes/production/sales/processed_files/2024-01-15/"
)

# List files in a volume using dbutils
files = dbutils.fs.ls("/Volumes/production/sales/landing_zone/")
for f in files:
    print(f.name)

import dlt
from pyspark.sql import functions as F

# ─── BRONZE LAYER: Raw ingestion ─────────────────────
@dlt.table(
    name="bronze_orders",
    comment="Raw orders loaded from CSV landing zone",
    table_properties={"quality": "bronze"}
)
def bronze_orders():
    return (
        spark.read.format("cloudFiles")  # Auto Loader
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .load("/Volumes/production/sales/landing_zone/")
    )

# ─── SILVER LAYER: Cleaned & validated ───────────────
@dlt.table(
    name="silver_orders",
    comment="Cleaned and validated orders"
)
@dlt.expect_or_drop("valid_order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("positive_amount", "amount > 0")
@dlt.expect("valid_status", "status IN ('pending','completed','cancelled')")
def silver_orders():
    return (
        dlt.read("bronze_orders")
        .withColumn("amount", F.col("amount").cast("decimal(10,2)"))
        .withColumn("order_date", F.to_date("order_date"))
        .withColumn("_loaded_at", F.current_timestamp())
        .dropDuplicates(["order_id"])
    )

# ─── GOLD LAYER: Aggregated business metrics ──────────
@dlt.table(
    name="gold_daily_sales",
    comment="Daily sales aggregations for reporting"
)
def gold_daily_sales():
    return (
        dlt.read("silver_orders")
        .filter(F.col("status") == "completed")
        .groupBy("order_date")
        .agg(
            F.sum("amount").alias("total_revenue"),
            F.count("order_id").alias("order_count"),
            F.avg("amount").alias("avg_order_value")
        )
    )

import dlt

@dlt.table()
# WARN: track violations, keep all rows
@dlt.expect("valid_phone", "phone_number RLIKE '^[0-9]{10}$'")
# DROP: drop bad rows silently
@dlt.expect_or_drop("not_null_email", "email IS NOT NULL")
# FAIL: stop the pipeline on any violation
@dlt.expect_or_fail("primary_key_unique", "COUNT(DISTINCT customer_id) = COUNT(*)")
# Multiple expectations on same decorator
@dlt.expect_all({
    "age_valid": "age BETWEEN 0 AND 120",
    "city_not_null": "city IS NOT NULL",
    "country_code_valid": "LENGTH(country_code) = 2"
})
def silver_customers():
    return dlt.read("bronze_customers")

import dlt

# 1. Define source — raw CDC events from Kafka
@dlt.table(name="cdc_events_raw")
def cdc_events_raw():
    return (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "orders-cdc")
        .load()
    )

# 2. Apply CDC changes to target table automatically
# DLT handles INSERT, UPDATE, DELETE from the CDC stream
dlt.apply_changes(
    target="silver_orders_cdc",     # target table to create/update
    source="cdc_events_raw",        # source CDC stream
    keys=["order_id"],              # primary key columns
    sequence_by="cdc_timestamp",   # ordering column (latest wins)
    apply_as_deletes=F.expr("op = 'd'"),   # 'd' events are deletes
    except_column_list=["op", "cdc_timestamp"]  # exclude CDC metadata cols
)
# DLT automatically handles SCD Type 1 (upsert) on silver_orders_cdc

from pyspark.sql import functions as F

# ─── Basic Auto Loader Ingestion ─────────────────────
df = (
    spark.readStream
    .format("cloudFiles")             # Auto Loader format
    .option("cloudFiles.format", "json")  # underlying file format
    .option("cloudFiles.schemaLocation", "dbfs:/checkpoints/orders_schema")
    .load("s3://my-bucket/landing/orders/")
)

# Write to Delta table (streaming)
(
    df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "dbfs:/checkpoints/orders_bronze")
    .trigger(availableNow=True)         # process all available, then stop
    .toTable("production.sales.bronze_orders")
)

# ─── Auto Loader with Schema Evolution ───────────────
df_evolved = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", "dbfs:/schemas/orders")
    # Schema evolution options:
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    # addNewColumns: new columns added automatically
    # rescue: unknown columns saved to _rescued_data
    # failOnNewColumns: pipeline fails if new column found
    .option("cloudFiles.inferColumnTypes", "true")
    .option("header", "true")
    .load("s3://my-bucket/landing/orders/")
    .withColumn("_source_file", F.input_file_name())
    .withColumn("_ingested_at", F.current_timestamp())
)

# ─── File Notification Mode (for scale) ──────────────
df_notify = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.useNotifications", "true")  # event-driven mode
    .option("cloudFiles.region", "us-east-1")
    .option("cloudFiles.schemaLocation", "dbfs:/schemas/events")
    .load("s3://my-bucket/events/")
)

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, precision_score
from sklearn.model_selection import train_test_split
import pandas as pd

# Set the experiment (creates it if doesn't exist)
mlflow.set_experiment("/Users/alice@company.com/churn_prediction")

# ─── Manual logging ────────────────────────────────────
with mlflow.start_run(run_name="rf_100_trees"):
    # Log parameters
    n_estimators = 100
    max_depth = 10
    mlflow.log_param("n_estimators", n_estimators)
    mlflow.log_param("max_depth", max_depth)
    mlflow.log_param("feature_count", X_train.shape[1])

    # Train model
    model = RandomForestClassifier(
        n_estimators=n_estimators, max_depth=max_depth, random_state=42
    )
    model.fit(X_train, y_train)

    # Log metrics
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    precision = precision_score(y_test, predictions)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("precision", precision)

    # Log the trained model
    mlflow.sklearn.log_model(model, artifact_path="model")

    # Log any file as an artifact
    mlflow.log_artifact("/tmp/feature_importance.png")

    print(f"Run ID: {mlflow.active_run().info.run_id}")
    print(f"Accuracy: {accuracy:.4f}")

# ─── Auto-logging (logs everything automatically!) ─────
mlflow.sklearn.autolog()  # enable autologging for sklearn

with mlflow.start_run():
    model = RandomForestClassifier(n_estimators=200)
    model.fit(X_train, y_train)
    # MLflow automatically logs: params, metrics, model, feature importance!

-- Query optimization: use ANALYZE TABLE for CBO
ANALYZE TABLE production.sales.orders
  COMPUTE STATISTICS FOR ALL COLUMNS;

-- Result caching: identical queries served from cache
-- Automatic in Databricks SQL — no action needed

-- Query history + explain
EXPLAIN SELECT * FROM production.sales.orders
WHERE order_date = '2024-01-15';

-- Create a named saved query (via UI or API)
-- Saved queries can be scheduled and used in dashboards

-- Parameterized query (for dashboards with filters)
SELECT *
FROM production.sales.orders
WHERE status = '{{status}}'          -- UI dropdown parameter
  AND order_date BETWEEN '{{start_date}}' AND '{{end_date}}';

-- Create an Alert: notify when a metric crosses a threshold
-- Example: alert when daily_orders < 100
SELECT COUNT(*) AS daily_orders
FROM production.sales.orders
WHERE order_date = CURRENT_DATE();

# 1. Install databricks-connect (match DBR version)
pip install databricks-connect==14.3.*

# 2. Configure connection
databricks configure --token

# OR set environment variables
export DATABRICKS_HOST="https://<workspace>.azuredatabricks.net"
export DATABRICKS_TOKEN="dapi..."
export DATABRICKS_CLUSTER_ID="0115-123456-abc123"

from databricks.connect import DatabricksSession

# Connect to remote Databricks cluster
spark = DatabricksSession.builder.getOrCreate()

# Now use spark exactly as you would in a notebook!
# This runs on the remote cluster, not your laptop
df = spark.table("production.sales.orders")
print(f"Row count: {df.count()}")   # executes on cluster
df.show(5)                          # results come back locally

# Transformations work the same
result = (
    df
    .filter("status = 'completed'")
    .groupBy("customer_id")
    .agg({"amount": "sum"})
    .sort("sum(amount)", ascending=False)
)
result.show()

# Explicitly connect with config
spark2 = (
    DatabricksSession.builder
    .remote(
        host="https://<workspace>.azuredatabricks.net",
        token="dapi...",
        cluster_id="0115-123456-abc123"
    )
    .getOrCreate()
)

bundle:
  name: daily-sales-etl

# Workspace configuration per environment
targets:
  dev:
    workspace:
      host: https://dev-workspace.azuredatabricks.net
    default: true

  staging:
    workspace:
      host: https://staging-workspace.azuredatabricks.net

  prod:
    workspace:
      host: https://prod-workspace.azuredatabricks.net
    mode: production   # adds "prod" prefix to all resource names

# Include resource files
include:
  - resources/*.yml

# Variables (overrideable per target)
variables:
  catalog:
    description: Target catalog name
    default: dev
  env:
    description: Environment name
    default: dev

resources:
  jobs:
    daily_etl_job:
      name: "Daily Sales ETL [${bundle.target}]"
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "UTC"
      
      job_clusters:
        - job_cluster_key: etl-cluster
          new_cluster:
            spark_version: 14.3.x-scala2.12
            node_type_id: Standard_DS4_v2
            num_workers: 4

      tasks:
        - task_key: extract
          job_cluster_key: etl-cluster
          notebook_task:
            notebook_path: ../src/notebooks/01_extract.py
            base_parameters:
              catalog: ${var.catalog}
              env: ${var.env}

        - task_key: transform
          depends_on:
            - task_key: extract
          job_cluster_key: etl-cluster
          notebook_task:
            notebook_path: ../src/notebooks/02_transform.py

      email_notifications:
        on_failure:
          - data-team@company.com

# Validate the bundle (check for errors)
databricks bundle validate

# Deploy to dev (default target)
databricks bundle deploy

# Deploy to specific target
databricks bundle deploy --target staging
databricks bundle deploy --target prod

# Run a specific job from the bundle
databricks bundle run daily_etl_job

# Run with variable overrides
databricks bundle run daily_etl_job --var="catalog=production"

# Destroy all deployed resources
databricks bundle destroy --target dev

# GitHub Actions CI/CD workflow
# .github/workflows/deploy.yml
# - on push to main: bundle deploy --target prod
# - on push to feature: bundle deploy --target dev

{
  "name": "Serverless ETL Job",
  "tasks": [{
    "task_key": "main_task",
    "notebook_task": {
      "notebook_path": "/Repos/myrepo/etl/main"
    },
    // No job_cluster_key! Use serverless instead:
    "environment_key": "serverless_env"
  }],
  "environments": [{
    "environment_key": "serverless_env",
    "spec": {
      "client": "1",
      "dependencies": [
        "pandas==2.0.0",
        "scikit-learn==1.3.0"
      ]
    }
  }]
}

import dlt
from pyspark.sql import functions as F

# STREAMING TABLE: continuously ingests new data
# Best for: source data that keeps arriving
@dlt.table()
def orders_streaming():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/Volumes/prod/landing/orders/")
    )

# MATERIALIZED VIEW: auto-refreshes on trigger
# Best for: aggregations over complete datasets
@dlt.view()
def daily_revenue_mv():
    return (
        dlt.read("orders_streaming")
        .groupBy(F.to_date("event_time").alias("date"))
        .agg(F.sum("amount").alias("revenue"))
    )

# PIPELINE configuration: set update mode
# continuous: always running (streaming)
# triggered: run on schedule (batch-like)
# Set in pipeline UI or databricks.yml:
# continuous: true   ← for streaming pipelines
# continuous: false  ← for scheduled batch pipelines