The Ultimate Guide to Databricks Data Engineer Associate Exam: Everything You Need to Know

Application
└── Job (triggered by action)
    └── Stage (divided by shuffles)
        └── Task (one per partition)

# In Databricks, SparkSession is pre-created as 'spark'
# But you can access it like this:
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MyApplication") \
    .config("spark.some.config.option", "some-value") \
    .getOrCreate()

# Check Spark version
print(spark.version)

# Access SparkContext
sc = spark.sparkContext
print(sc.appName)

# From a Python list
data = [("Alice", 30), ("Bob", 25), ("Charlie", 35)]
columns = ["name", "age"]
df = spark.createDataFrame(data, columns)

# From a CSV file
df_csv = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("/path/to/file.csv")

# From a JSON file
df_json = spark.read.json("/path/to/file.json")

# From a Parquet file
df_parquet = spark.read.parquet("/path/to/file.parquet")

# From a Delta table
df_delta = spark.read.format("delta").load("/path/to/delta/table")

# Or using SQL
df_sql = spark.read.table("database.tablename")

# From a JDBC source
df_jdbc = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:5432/database") \
    .option("dbtable", "schema.tablename") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

from pyspark.sql import functions as F
from pyspark.sql.types import *

# Select specific columns
df.select("name", "age")
df.select(F.col("name"), F.col("age"))

# Filter/Where
df.filter(F.col("age") > 25)
df.where("age > 25")

# Add new columns
df.withColumn("birth_year", 2024 - F.col("age"))

# Rename columns
df.withColumnRenamed("name", "full_name")

# Drop columns
df.drop("unnecessary_column")

# Sort/OrderBy
df.sort("age")
df.orderBy(F.col("age").desc())

# Distinct values
df.distinct()
df.dropDuplicates(["name"])

# Limit rows
df.limit(100)

# Group By and Aggregations
df.groupBy("department") \
  .agg(
      F.count("*").alias("employee_count"),
      F.avg("salary").alias("avg_salary"),
      F.max("salary").alias("max_salary"),
      F.min("salary").alias("min_salary"),
      F.sum("salary").alias("total_salary")
  )

# Joins
df1.join(df2, df1["id"] == df2["id"], "inner")
df1.join(df2, df1["id"] == df2["id"], "left")
df1.join(df2, df1["id"] == df2["id"], "right")
df1.join(df2, df1["id"] == df2["id"], "full")
df1.join(df2, df1["id"] == df2["id"], "left_semi")
df1.join(df2, df1["id"] == df2["id"], "left_anti")
df1.join(df2, df1["id"] == df2["id"], "cross")

# Union
df1.union(df2)           # Combines by position
df1.unionByName(df2)     # Combines by column name

# Collect all rows to driver (careful with large datasets!)
rows = df.collect()

# Show first N rows
df.show(20)
df.show(20, truncate=False)

# Count rows
count = df.count()

# Get first row
first_row = df.first()
first_5 = df.take(5)

# Describe statistics
df.describe().show()
df.summary().show()

# Write to storage
df.write.parquet("/path/to/output")
df.write.format("delta").save("/path/to/delta/table")
df.write.saveAsTable("database.tablename")

# Register DataFrame as temporary view
df.createOrReplaceTempView("employees")

# Run SQL query
result = spark.sql("""
    SELECT department, 
           COUNT(*) as emp_count,
           AVG(salary) as avg_salary
    FROM employees
    WHERE salary > 50000
    GROUP BY department
    ORDER BY avg_salary DESC
""")

# Create global temporary view (accessible across sessions)
df.createOrReplaceGlobalTempView("global_employees")
spark.sql("SELECT * FROM global_temp.global_employees")

from pyspark.sql.types import *

# Define explicit schema
schema = StructType([
    StructField("id", IntegerType(), nullable=False),
    StructField("name", StringType(), nullable=True),
    StructField("salary", DoubleType(), nullable=True),
    StructField("hire_date", DateType(), nullable=True),
    StructField("is_active", BooleanType(), nullable=True),
    StructField("address", StructType([
        StructField("street", StringType(), nullable=True),
        StructField("city", StringType(), nullable=True),
        StructField("zip", StringType(), nullable=True)
    ]), nullable=True),
    StructField("skills", ArrayType(StringType()), nullable=True),
    StructField("metadata", MapType(StringType(), StringType()), nullable=True)
])

# Create DataFrame with explicit schema
df = spark.createDataFrame(data, schema)

# Check schema
df.printSchema()
df.schema

# Cast column types
df.withColumn("salary", F.col("salary").cast(LongType()))
df.withColumn("hire_date", F.col("hire_date").cast("date"))

# Working with Arrays
df = spark.createDataFrame([
    (1, ["Python", "SQL", "Spark"]),
    (2, ["Java", "Scala"])
], ["id", "skills"])

# Explode array into rows
df.select("id", F.explode("skills").alias("skill"))

# Array functions
df.select(
    "id",
    F.size("skills").alias("skill_count"),
    F.array_contains("skills", "Python").alias("knows_python"),
    F.array_distinct("skills"),
    F.sort_array("skills")
)

# Working with Structs
df = spark.createDataFrame([
    (1, ("John", "Doe", 30)),
], ["id", "person"])

df.select("id", "person.first_name", "person.last_name")

# Working with Maps
df = spark.createDataFrame([
    (1, {"key1": "value1", "key2": "value2"})
], ["id", "metadata"])

df.select(
    "id",
    F.map_keys("metadata").alias("keys"),
    F.map_values("metadata").alias("values"),
    df["metadata"]["key1"].alias("key1_value")
)

from pyspark.sql.window import Window

# Define window specification
window_spec = Window.partitionBy("department").orderBy("salary")

# Ranking functions
df.withColumn("rank", F.rank().over(window_spec))
df.withColumn("dense_rank", F.dense_rank().over(window_spec))
df.withColumn("row_number", F.row_number().over(window_spec))
df.withColumn("percent_rank", F.percent_rank().over(window_spec))
df.withColumn("ntile", F.ntile(4).over(window_spec))

# Analytic functions
df.withColumn("lag_salary", F.lag("salary", 1).over(window_spec))
df.withColumn("lead_salary", F.lead("salary", 1).over(window_spec))

# Aggregate functions with window
window_agg = Window.partitionBy("department")
df.withColumn("dept_avg_salary", F.avg("salary").over(window_agg))
df.withColumn("dept_total_salary", F.sum("salary").over(window_agg))

# Running totals
window_running = Window.partitionBy("department") \
    .orderBy("hire_date") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
df.withColumn("running_salary_total", F.sum("salary").over(window_running))

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType

# Python UDF (not optimized by Catalyst)
def categorize_salary(salary):
    if salary < 50000:
        return "Low"
    elif salary < 100000:
        return "Medium"
    else:
        return "High"

# Register UDF
categorize_udf = udf(categorize_salary, StringType())

# Use UDF
df.withColumn("salary_category", categorize_udf(F.col("salary")))

# Lambda UDF
double_salary = udf(lambda x: x * 2, IntegerType())

# Pandas UDF (vectorized - much faster!)
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf(StringType())
def categorize_salary_pandas(salary: pd.Series) -> pd.Series:
    return salary.apply(lambda x: "Low" if x < 50000 else "Medium" if x < 100000 else "High")

df.withColumn("salary_category", categorize_salary_pandas(F.col("salary")))

# Method 1: Write DataFrame as Delta
df.write.format("delta").save("/path/to/delta/table")

# Method 2: Write with partitioning
df.write \
  .format("delta") \
  .partitionBy("year", "month") \
  .save("/path/to/delta/table")

# Method 3: Save as table (creates metadata in metastore)
df.write.format("delta").saveAsTable("database.table_name")

# Method 4: Create table using SQL
spark.sql("""
    CREATE TABLE IF NOT EXISTS sales (
        id BIGINT,
        customer_id BIGINT,
        product_id BIGINT,
        amount DOUBLE,
        sale_date DATE
    )
    USING DELTA
    PARTITIONED BY (sale_date)
    LOCATION '/path/to/delta/sales'
    COMMENT 'Sales transactions table'
    TBLPROPERTIES (
        'delta.autoOptimize.optimizeWrite' = 'true',
        'delta.autoOptimize.autoCompact' = 'true'
    )
""")

# Method 5: CREATE TABLE AS SELECT (CTAS)
spark.sql("""
    CREATE TABLE new_sales
    USING DELTA
    AS SELECT * FROM old_sales WHERE year = 2024
""")

# Read Delta table from path
df = spark.read.format("delta").load("/path/to/delta/table")

# Read Delta table from catalog
df = spark.read.table("database.table_name")

# Read a specific version (Time Travel)
df_v2 = spark.read.format("delta") \
    .option("versionAsOf", 2) \
    .load("/path/to/delta/table")

# Read at a specific timestamp
df_ts = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-01") \
    .load("/path/to/delta/table")

# Using SQL for time travel
spark.sql("""
    SELECT * FROM sales VERSION AS OF 3
""")

spark.sql("""
    SELECT * FROM sales TIMESTAMP AS OF '2024-01-01 00:00:00'
""")

# Append data (default mode)
new_data.write.mode("append").format("delta").save("/path/to/delta/table")

# Overwrite entire table
new_data.write.mode("overwrite").format("delta").save("/path/to/delta/table")

# Insert using SQL
spark.sql("""
    INSERT INTO sales VALUES (1, 100, 200, 99.99, '2024-01-15')
""")

spark.sql("""
    INSERT INTO sales SELECT * FROM staging_sales
""")

# Insert Overwrite (replaces data matching partition)
spark.sql("""
    INSERT OVERWRITE sales
    SELECT * FROM staging_sales WHERE sale_date = '2024-01-15'
""")

from delta.tables import DeltaTable

# Load the Delta table
delta_table = DeltaTable.forPath(spark, "/path/to/delta/table")

# Update with condition
delta_table.update(
    condition = F.col("customer_id") == 100,
    set = {"amount": F.col("amount") * 1.1}
)

# Update using SQL
spark.sql("""
    UPDATE sales
    SET amount = amount * 1.1
    WHERE customer_id = 100
""")

# Delete with condition
delta_table.delete(condition = F.col("sale_date") < "2020-01-01")

# Delete using SQL
spark.sql("""
    DELETE FROM sales
    WHERE sale_date < '2020-01-01'
""")

from delta.tables import DeltaTable

# Target table
target = DeltaTable.forPath(spark, "/path/to/delta/table")

# Source DataFrame
source = spark.read.table("staging_updates")

# Perform MERGE
target.alias("target") \
    .merge(
        source.alias("source"),
        "target.id = source.id"
    ) \
    .whenMatchedUpdate(set={
        "amount": F.col("source.amount"),
        "updated_at": F.current_timestamp()
    }) \
    .whenNotMatchedInsert(values={
        "id": F.col("source.id"),
        "customer_id": F.col("source.customer_id"),
        "amount": F.col("source.amount"),
        "sale_date": F.col("source.sale_date"),
        "created_at": F.current_timestamp()
    }) \
    .whenNotMatchedBySourceDelete() \
    .execute()

-- MERGE using SQL
MERGE INTO sales AS target
USING staging_updates AS source
ON target.id = source.id
WHEN MATCHED AND source.action = 'UPDATE' THEN
    UPDATE SET target.amount = source.amount,
               target.updated_at = current_timestamp()
WHEN MATCHED AND source.action = 'DELETE' THEN
    DELETE
WHEN NOT MATCHED THEN
    INSERT (id, customer_id, amount, sale_date)
    VALUES (source.id, source.customer_id, source.amount, source.sale_date)

# This will FAIL if 'new_column' doesn't exist in the schema
new_data_with_extra_column.write \
    .mode("append") \
    .format("delta") \
    .save("/path/to/delta/table")

# Error: A schema mismatch detected when writing to the Delta table

# Enable schema evolution (mergeSchema)
new_data_with_extra_column.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .format("delta") \
    .save("/path/to/delta/table")

# Overwrite and update schema
new_data.write \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .format("delta") \
    .save("/path/to/delta/table")

# Add column
spark.sql("ALTER TABLE sales ADD COLUMN discount DOUBLE")

# Rename column (requires column mapping)
spark.sql("ALTER TABLE sales RENAME COLUMN amount TO total_amount")

# Drop column (requires column mapping)
spark.sql("ALTER TABLE sales DROP COLUMN old_column")

# Change column type
spark.sql("ALTER TABLE sales ALTER COLUMN amount TYPE DECIMAL(10,2)")

# Add table comment
spark.sql("COMMENT ON TABLE sales IS 'Main sales transactions'")

# Add column comment
spark.sql("ALTER TABLE sales ALTER COLUMN amount COMMENT 'Transaction amount in USD'")

# Query version 0 (initial version)
spark.read.format("delta") \
    .option("versionAsOf", 0) \
    .load("/path/to/delta/table")

# Query at timestamp
spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-01T00:00:00.000Z") \
    .load("/path/to/delta/table")

# View table history
delta_table = DeltaTable.forPath(spark, "/path/to/delta/table")
delta_table.history().show()
delta_table.history(10).show()  # Last 10 operations

# SQL equivalent
spark.sql("DESCRIBE HISTORY sales").show()
spark.sql("DESCRIBE HISTORY sales LIMIT 5").show()

# Optimize entire table
spark.sql("OPTIMIZE sales")

# Optimize with Z-ORDER (co-locate related data)
spark.sql("OPTIMIZE sales ZORDER BY (customer_id, sale_date)")

# Optimize specific partition
spark.sql("OPTIMIZE sales WHERE sale_date = '2024-01-15'")

# Vacuum with default retention (7 days)
spark.sql("VACUUM sales")

# Vacuum with custom retention (in hours)
spark.sql("VACUUM sales RETAIN 168 HOURS")  # 7 days

# DRY RUN - shows files to be deleted without actually deleting
spark.sql("VACUUM sales DRY RUN")

# WARNING: This disables time travel beyond the retention period
# To vacuum with less than 7 days (use carefully!):
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
spark.sql("VACUUM sales RETAIN 0 HOURS")
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "true")

# View table properties
spark.sql("DESCRIBE EXTENDED sales")
spark.sql("SHOW TBLPROPERTIES sales")

# Set table properties
spark.sql("""
    ALTER TABLE sales 
    SET TBLPROPERTIES (
        'delta.logRetentionDuration' = 'interval 30 days',
        'delta.deletedFileRetentionDuration' = 'interval 7 days',
        'delta.autoOptimize.optimizeWrite' = 'true',
        'delta.autoOptimize.autoCompact' = 'true'
    )
""")

# Enable column mapping (required for renaming/dropping columns)
spark.sql("""
    ALTER TABLE sales 
    SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name')
""")

# Enable Change Data Feed
spark.sql("""
    ALTER TABLE sales 
    SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true')
""")

# Enable CDF on table creation
spark.sql("""
    CREATE TABLE sales (
        id BIGINT,
        amount DOUBLE,
        updated_at TIMESTAMP
    )
    USING DELTA
    TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true')
""")

# Read changes from a specific version
changes = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", 1) \
    .table("sales")

# Read changes from a specific timestamp
changes = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingTimestamp", "2024-01-01") \
    .table("sales")

# CDF adds these special columns:
# _change_type: 'insert', 'update_preimage', 'update_postimage', 'delete'
# _commit_version: The version of the commit
# _commit_timestamp: The timestamp of the commit

changes.filter("_change_type = 'insert'").show()
changes.filter("_change_type IN ('update_preimage', 'update_postimage')").show()

# NOT NULL constraint
spark.sql("""
    ALTER TABLE sales 
    ALTER COLUMN id SET NOT NULL
""")

# CHECK constraint
spark.sql("""
    ALTER TABLE sales 
    ADD CONSTRAINT valid_amount CHECK (amount > 0)
""")

# View constraints
spark.sql("DESCRIBE EXTENDED sales")

# Drop constraint
spark.sql("ALTER TABLE sales DROP CONSTRAINT valid_amount")

Source Systems → [Bronze] → [Silver] → [Gold] → Analytics/ML

# Ingesting raw JSON data into Bronze
raw_data = spark.read \
    .format("json") \
    .option("multiLine", "true") \
    .load("/raw/incoming/orders/*.json")

# Add ingestion metadata
bronze_data = raw_data.withColumn("_ingested_at", F.current_timestamp()) \
                      .withColumn("_source_file", F.input_file_name())

bronze_data.write \
    .mode("append") \
    .format("delta") \
    .save("/bronze/orders")

# Read from Bronze
bronze_df = spark.read.format("delta").load("/bronze/orders")

# Clean and validate
silver_df = bronze_df \
    .filter(F.col("order_id").isNotNull()) \
    .filter(F.col("amount") > 0) \
    .dropDuplicates(["order_id"]) \
    .withColumn("order_date", F.to_date("order_date", "yyyy-MM-dd")) \
    .withColumn("amount", F.col("amount").cast("decimal(10,2)")) \
    .withColumn("status", F.upper(F.col("status"))) \
    .withColumn("_processed_at", F.current_timestamp())

# Upsert into Silver (handle late-arriving data)
from delta.tables import DeltaTable

if DeltaTable.isDeltaTable(spark, "/silver/orders"):
    silver_table = DeltaTable.forPath(spark, "/silver/orders")
    silver_table.alias("target") \
        .merge(silver_df.alias("source"), "target.order_id = source.order_id") \
        .whenMatchedUpdateAll() \
        .whenNotMatchedInsertAll() \
        .execute()
else:
    silver_df.write.format("delta").save("/silver/orders")

# Create Gold table: Daily Sales Summary
gold_daily_sales = spark.read.format("delta").load("/silver/orders") \
    .filter("status = 'COMPLETED'") \
    .groupBy(
        F.to_date("order_date").alias("date"),
        "product_category",
        "region"
    ) \
    .agg(
        F.count("*").alias("order_count"),
        F.sum("amount").alias("total_revenue"),
        F.avg("amount").alias("avg_order_value"),
        F.countDistinct("customer_id").alias("unique_customers")
    )

gold_daily_sales.write \
    .mode("overwrite") \
    .format("delta") \
    .partitionBy("date") \
    .save("/gold/daily_sales_summary")

# String functions
df.select(
    F.upper("name"),
    F.lower("name"),
    F.trim("name"),
    F.ltrim("name"),
    F.rtrim("name"),
    F.length("name"),
    F.substring("name", 1, 3),
    F.concat("first_name", F.lit(" "), "last_name"),
    F.concat_ws(" ", "first_name", "last_name"),
    F.split("full_name", " "),
    F.regexp_replace("phone", "[^0-9]", ""),
    F.regexp_extract("email", "(@.*)", 1),
    F.like("name", "A%"),
    F.lpad("id", 10, "0"),
    F.rpad("name", 20, " ")
)

# Date/Time functions
df.select(
    F.current_date(),
    F.current_timestamp(),
    F.to_date("date_str", "yyyy-MM-dd"),
    F.to_timestamp("ts_str", "yyyy-MM-dd HH:mm:ss"),
    F.date_format("date_col", "MM/dd/yyyy"),
    F.year("date_col"),
    F.month("date_col"),
    F.dayofmonth("date_col"),
    F.dayofweek("date_col"),
    F.dayofyear("date_col"),
    F.hour("timestamp_col"),
    F.minute("timestamp_col"),
    F.second("timestamp_col"),
    F.date_add("date_col", 7),
    F.date_sub("date_col", 7),
    F.datediff("end_date", "start_date"),
    F.months_between("end_date", "start_date"),
    F.add_months("date_col", 3),
    F.last_day("date_col"),
    F.next_day("date_col", "Monday"),
    F.trunc("date_col", "month"),
    F.date_trunc("month", "timestamp_col"),
    F.unix_timestamp("timestamp_col"),
    F.from_unixtime("unix_ts"),
    F.from_utc_timestamp("timestamp_col", "America/New_York")
)

# Null handling
df.select(
    F.isnull("column"),
    F.isnan("column"),
    F.coalesce("col1", "col2", F.lit("default")),
    F.nvl("col1", "col2"),          # Returns col1 if not null, else col2
    F.nvl2("col1", "col2", "col3"), # If col1 not null return col2, else col3
    F.nullif("col1", "col2"),       # Returns null if col1 == col2
    F.ifnull("col1", "col2"),       # Same as NVL
    F.nanvl("col1", "col2")         # Returns col1 if not NaN, else col2
)

# Fill null values
df.fillna(0, subset=["salary"])
df.fillna({"salary": 0, "name": "Unknown"})
df.na.fill(0)
df.na.drop()  # Drop rows with any null
df.na.drop(how="all")  # Drop rows where all values are null
df.na.drop(subset=["critical_column"])

# CASE WHEN using when/otherwise
df.withColumn("salary_tier",
    F.when(F.col("salary") < 50000, "Junior")
     .when(F.col("salary") < 100000, "Mid")
     .when(F.col("salary") < 150000, "Senior")
     .otherwise("Executive")
)

# Using expr for complex SQL expressions
df.withColumn("status",
    F.expr("CASE WHEN age < 18 THEN 'minor' WHEN age < 65 THEN 'adult' ELSE 'senior' END")
)

# CSV
df = spark.read \
    .format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .option("delimiter", ",") \
    .option("quote", '"') \
    .option("escape", "\\") \
    .option("nullValue", "NULL") \
    .option("dateFormat", "yyyy-MM-dd") \
    .option("timestampFormat", "yyyy-MM-dd HH:mm:ss") \
    .option("multiLine", "true") \
    .option("encoding", "UTF-8") \
    .load("/path/to/*.csv")

# JSON
df = spark.read \
    .format("json") \
    .option("multiLine", "true") \
    .option("inferSchema", "true") \
    .option("dateFormat", "yyyy-MM-dd") \
    .load("/path/to/*.json")

# Parquet
df = spark.read.parquet("/path/to/parquet/")

# Avro
df = spark.read.format("avro").load("/path/to/avro/")

# ORC
df = spark.read.orc("/path/to/orc/")

# Text files
df = spark.read.text("/path/to/text/")
df = spark.read.format("text").load("/path/to/text/*.txt")

# Write modes
# "append" - Append to existing data
# "overwrite" - Overwrite all existing data
# "error"/"errorifexists" - Throw error if data exists (default)
# "ignore" - Silently skip write if data exists

# Write as Parquet
df.write \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet("/path/to/output/")

# Write as CSV
df.write \
    .mode("overwrite") \
    .option("header", "true") \
    .csv("/path/to/output.csv")

# Write as Delta
df.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .partitionBy("date") \
    .save("/path/to/delta/table")

# Coalesce before writing (reduce output files)
df.coalesce(1).write.mode("overwrite").csv("/path/to/single/file.csv")

# Repartition for balanced output
df.repartition(10).write.format("delta").save("/path/to/delta/")

# Kafka source
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "topic1,topic2") \
    .option("startingOffsets", "earliest") \
    .option("maxOffsetsPerTrigger", 10000) \
    .load()

# Parse Kafka value
from pyspark.sql.types import *
schema = StructType([
    StructField("id", IntegerType()),
    StructField("value", StringType())
])

parsed_df = kafka_df.select(
    F.from_json(F.col("value").cast("string"), schema).alias("data")
).select("data.*")

# Auto Loader (cloud file source)
autoloader_df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("cloudFiles.inferColumnTypes", "true") \
    .option("cloudFiles.schemaLocation", "/path/to/schema/") \
    .load("/path/to/landing/zone/")

# Delta table as stream source
delta_stream = spark.readStream \
    .format("delta") \
    .option("maxFilesPerTrigger", 100) \
    .table("bronze.orders")

# Rate source (for testing)
test_stream = spark.readStream \
    .format("rate") \
    .option("rowsPerSecond", 100) \
    .load()

# Write to Delta (most common)
query = df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/path/to/checkpoint/") \
    .trigger(processingTime="1 minute") \
    .start("/path/to/output/delta/")

# Write to Delta table by name
query = df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/path/to/checkpoint/") \
    .toTable("silver.orders")

# Write to Kafka
query = df.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("topic", "output-topic") \
    .option("checkpointLocation", "/path/to/checkpoint/") \
    .start()

# Write to memory (for debugging/testing)
query = df.writeStream \
    .format("memory") \
    .queryName("temp_table") \
    .outputMode("complete") \
    .start()

spark.sql("SELECT * FROM temp_table").show()

# Console (for debugging)
query = df.writeStream \
    .format("console") \
    .outputMode("append") \
    .start()

# Manage streaming query
query.status     # Current status
query.isActive   # True if running
query.stop()     # Stop the query
query.awaitTermination()  # Block until complete
query.awaitTermination(timeout=60)  # Block with timeout

# Always specify a unique checkpoint location per query
query = df.writeStream \
    .option("checkpointLocation", "/checkpoint/orders_to_silver/") \
    .format("delta") \
    .start("/silver/orders/")

# Basic Auto Loader setup
df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("cloudFiles.schemaLocation", "/schema/orders/") \
    .load("/landing/orders/")

# Write to Bronze Delta table
query = df.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/checkpoint/orders_landing/") \
    .option("mergeSchema", "true") \
    .trigger(availableNow=True) \
    .toTable("bronze.orders")

query.awaitTermination()

df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "csv") \
    .option("cloudFiles.schemaLocation", "/schema/events/") \
    .option("cloudFiles.inferColumnTypes", "true") \
    .option("cloudFiles.backfillInterval", "1 day") \
    .option("cloudFiles.maxBytesPerTrigger", "10g") \
    .option("cloudFiles.maxFilesPerTrigger", 1000) \
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns") \
    .option("header", "true") \
    .load("/landing/events/")

# Define watermark (tolerate up to 10 minutes of late data)
windowed_counts = df \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        F.window("event_time", "5 minutes"),
        "user_id"
    ) \
    .count()

# Sliding window
sliding_window = df \
    .withWatermark("event_time", "1 hour") \
    .groupBy(
        F.window("event_time", "1 hour", "15 minutes"),  # 1-hour window, 15-min slide
        "product_id"
    ) \
    .agg(F.sum("amount").alias("total"))

-- Basic COPY INTO
COPY INTO sales
FROM '/path/to/source/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true')

-- COPY INTO with JSON
COPY INTO events
FROM '/landing/events/'
FILEFORMAT = JSON
FORMAT_OPTIONS ('multiLine' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true')

-- COPY INTO with explicit schema
COPY INTO sales
FROM (
    SELECT 
        cast(id as BIGINT) as id,
        cast(amount as DECIMAL(10,2)) as amount,
        to_date(sale_date, 'yyyy-MM-dd') as sale_date
    FROM read_files('/landing/sales/', format => 'csv', header => true)
)

import dlt
from pyspark.sql import functions as F
from pyspark.sql.types import *

# === BRONZE LAYER ===

# Create a streaming table from Auto Loader
@dlt.table(
    name="raw_orders",
    comment="Raw orders data ingested from landing zone",
    table_properties={
        "quality": "bronze",
        "pipelines.reset.allowed": "true"
    }
)
def raw_orders():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .option("cloudFiles.inferColumnTypes", "true")
            .option("cloudFiles.schemaLocation", 
                    "/pipelines/schemas/raw_orders/")
            .load("/landing/orders/")
    )

# === SILVER LAYER ===

# Create a streaming table with expectations (data quality)
@dlt.table(
    name="clean_orders",
    comment="Cleaned and validated orders",
    table_properties={"quality": "silver"}
)
@dlt.expect("valid_order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("positive_amount", "amount > 0")
@dlt.expect_or_fail("valid_status", "status IN ('PENDING', 'PROCESSING', 'COMPLETED', 'CANCELLED')")
def clean_orders():
    return (
        dlt.read_stream("raw_orders")
            .filter(F.col("order_id").isNotNull())
            .withColumn("order_date", F.to_date("order_date"))
            .withColumn("amount", F.col("amount").cast("decimal(10,2)"))
            .withColumn("status", F.upper("status"))
            .withColumn("_processed_at", F.current_timestamp())
    )

# === GOLD LAYER ===

# Create a Materialized View for aggregations
@dlt.table(
    name="daily_sales_summary",
    comment="Daily sales aggregated by product category",
    table_properties={"quality": "gold"}
)
def daily_sales_summary():
    return (
        dlt.read("clean_orders")
            .filter("status = 'COMPLETED'")
            .groupBy(
                F.to_date("order_date").alias("date"),
                "product_category"
            )
            .agg(
                F.count("*").alias("order_count"),
                F.sum("amount").alias("total_revenue"),
                F.avg("amount").alias("avg_order_value")
            )
    )

# @dlt.expect: Tracks violations but doesn't drop or fail
@dlt.expect("valid_email", "email LIKE '%@%.%'")
def my_table():
    ...

# @dlt.expect_or_drop: Drops rows that violate the constraint
@dlt.expect_or_drop("non_null_id", "id IS NOT NULL")
def my_table():
    ...

# @dlt.expect_or_fail: Fails the pipeline if any row violates
@dlt.expect_or_fail("critical_check", "amount >= 0")
def my_table():
    ...

# Multiple expectations
@dlt.expect("valid_order_id", "order_id IS NOT NULL")
@dlt.expect_or_drop("positive_amount", "amount > 0")
@dlt.expect_or_fail("valid_customer", "customer_id IS NOT NULL")
def my_table():
    ...

# expect_all: Apply multiple constraints at once
@dlt.expect_all({
    "valid_id": "id IS NOT NULL",
    "valid_amount": "amount > 0",
    "valid_date": "order_date >= '2020-01-01'"
})
def my_table():
    ...

# expect_all_or_drop
@dlt.expect_all_or_drop({
    "valid_id": "id IS NOT NULL",
    "valid_amount": "amount > 0"
})
def my_table():
    ...

# expect_all_or_fail
@dlt.expect_all_or_fail({
    "critical_id": "id IS NOT NULL",
    "critical_amount": "amount > 0"
})
def my_table():
    ...

-- Bronze: Streaming table from Auto Loader
CREATE OR REFRESH STREAMING TABLE raw_customers
COMMENT "Raw customer data from source systems"
AS SELECT * FROM cloud_files("/landing/customers/", "json",
    map("cloudFiles.inferColumnTypes", "true"))

-- Silver: Streaming table with expectations
CREATE OR REFRESH STREAMING TABLE clean_customers (
    CONSTRAINT valid_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION FAIL UPDATE,
    CONSTRAINT valid_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW,
    CONSTRAINT valid_age EXPECT (age BETWEEN 18 AND 120) ON VIOLATION WARN
)
COMMENT "Cleaned customer data"
AS
SELECT 
    customer_id,
    upper(trim(first_name)) as first_name,
    upper(trim(last_name)) as last_name,
    lower(email) as email,
    cast(age as INT) as age,
    to_date(signup_date, 'yyyy-MM-dd') as signup_date,
    current_timestamp() as _processed_at
FROM STREAM(LIVE.raw_customers)
WHERE customer_id IS NOT NULL

-- Gold: Materialized view
CREATE OR REFRESH MATERIALIZED VIEW customer_summary
COMMENT "Customer segment summary"
AS
SELECT 
    CASE 
        WHEN age < 25 THEN 'Gen Z'
        WHEN age < 40 THEN 'Millennial'
        WHEN age < 55 THEN 'Gen X'
        ELSE 'Boomer'
    END as age_segment,
    count(*) as customer_count,
    avg(age) as avg_age
FROM LIVE.clean_customers
GROUP BY 1

{
  "name": "Orders Pipeline",
  "storage": "/pipelines/orders/",
  "target": "orders_db",
  "libraries": [
    {"notebook": {"path": "/pipelines/bronze/raw_orders"}},
    {"notebook": {"path": "/pipelines/silver/clean_orders"}},
    {"notebook": {"path": "/pipelines/gold/order_summaries"}}
  ],
  "clusters": [
    {
      "label": "default",
      "autoscale": {
        "min_workers": 1,
        "max_workers": 5
      }
    }
  ],
  "development": false,
  "photon": true,
  "channel": "CURRENT",
  "edition": "ADVANCED",
  "continuous": false
}

import dlt
from pyspark.sql import functions as F

# Source CDC stream
@dlt.view
def customers_cdc():
    return spark.readStream \
        .format("cloudFiles") \
        .option("cloudFiles.format", "json") \
        .load("/cdc/customers/")

# Apply changes to target table
dlt.create_streaming_table("customers")

dlt.apply_changes(
    target = "customers",
    source = "customers_cdc",
    keys = ["customer_id"],
    sequence_by = F.col("updated_at"),  # Sequence field to handle ordering
    apply_as_deletes = F.expr("operation = 'DELETE'"),
    apply_as_truncates = F.expr("operation = 'TRUNCATE'"),
    except_column_list = ["operation", "updated_at"],  # Columns to exclude
    stored_as_scd_type = 1  # SCD Type 1 (overwrite) or 2 (history)
)

dlt.apply_changes(
    target = "customers_history",
    source = "customers_cdc",
    keys = ["customer_id"],
    sequence_by = F.col("updated_at"),
    stored_as_scd_type = 2,  # Maintain full history
    track_history_column_list = ["email", "address", "phone"]  # Only track specific columns
)

# Using the Databricks SDK
from databricks.sdk import WorkspaceClient
from databricks.sdk.service import jobs

client = WorkspaceClient()

# Create a simple notebook job
job = client.jobs.create(
    name="My Data Pipeline",
    tasks=[
        jobs.Task(
            task_key="ingest_data",
            notebook_task=jobs.NotebookTask(
                notebook_path="/pipelines/ingestion/load_bronze",
                base_parameters={"env": "prod", "date": "{{ds}}"}
            ),
            new_cluster=jobs.ClusterSpec(
                spark_version="13.3.x-scala2.12",
                node_type_id="i3.xlarge",
                num_workers=2
            ),
            timeout_seconds=3600,
            max_retries=2,
            min_retry_interval_millis=60000  # 1 minute between retries
        )
    ],
    schedule=jobs.CronSchedule(
        quartz_cron_expression="0 0 8 * * ?",  # Daily at 8 AM
        timezone_id="America/New_York",
        pause_status=jobs.PauseStatus.UNPAUSED
    ),
    email_notifications=jobs.JobEmailNotifications(
        on_failure=["data-team@company.com"],
        on_success=["data-team@company.com"]
    )
)

{
  "name": "ETL Pipeline",
  "tasks": [
    {
      "task_key": "ingest_bronze",
      "notebook_task": {
        "notebook_path": "/pipelines/bronze/ingest"
      },
      "new_cluster": {
        "spark_version": "13.3.x-scala2.12",
        "node_type_id": "i3.xlarge",
        "num_workers": 2
      }
    },
    {
      "task_key": "transform_silver",
      "depends_on": [{"task_key": "ingest_bronze"}],
      "notebook_task": {
        "notebook_path": "/pipelines/silver/transform"
      },
      "existing_cluster_id": "{{job_cluster_id}}"
    },
    {
      "task_key": "aggregate_gold_sales",
      "depends_on": [{"task_key": "transform_silver"}],
      "notebook_task": {
        "notebook_path": "/pipelines/gold/sales_agg"
      }
    },
    {
      "task_key": "aggregate_gold_customers",
      "depends_on": [{"task_key": "transform_silver"}],
      "notebook_task": {
        "notebook_path": "/pipelines/gold/customer_agg"
      }
    },
    {
      "task_key": "update_reporting",
      "depends_on": [
        {"task_key": "aggregate_gold_sales"},
        {"task_key": "aggregate_gold_customers"}
      ],
      "notebook_task": {
        "notebook_path": "/pipelines/reporting/refresh"
      }
    }
  ]
}

# Notebook task
notebook_task = jobs.NotebookTask(
    notebook_path="/path/to/notebook",
    base_parameters={"param1": "value1"}
)

# Python Script task
python_task = jobs.SparkPythonTask(
    python_file="/path/to/script.py",
    parameters=["--env", "prod"]
)

# JAR task
jar_task = jobs.SparkJarTask(
    main_class_name="com.company.MainClass",
    jar_uri="dbfs:/jars/my_app.jar",
    parameters=["arg1", "arg2"]
)

# DLT Pipeline task
dlt_task = jobs.PipelineTask(
    pipeline_id="pipeline-id-here"
)

# SQL task
sql_task = jobs.SqlTask(
    query=jobs.SqlTaskQuery(query_id="query-id"),
    warehouse_id="warehouse-id"
)

# Python Wheel task
wheel_task = jobs.PythonWheelTask(
    package_name="my_package",
    entry_point="main",
    named_parameters={"env": "prod"}
)