Delta Lake: The Lakehouse Foundation

# Create a Delta table from a DataFrame
df.write.format("delta").mode("overwrite").save("/mnt/data/silver/employees")

# Create a managed Delta table (registered in metastore)
df.write.format("delta").saveAsTable("main.hr.employees")

# Append new data
new_employees.write.format("delta").mode("append").save("/mnt/data/silver/employees")

# Partitioned Delta table
events.write.format("delta") \
    .partitionBy("year", "month") \
    .mode("overwrite") \
    .save("/mnt/data/bronze/events")

-- Create a Delta table with SQL
CREATE TABLE main.hr.employees (
    id INT,
    name STRING,
    department STRING,
    salary DOUBLE,
    hire_date DATE
) USING DELTA
PARTITIONED BY (department);

-- INSERT data
INSERT INTO main.hr.employees VALUES
    (1, 'Alice', 'Engineering', 95000, '2023-01-15'),
    (2, 'Bob', 'Data Science', 105000, '2022-06-01');

-- UPDATE specific rows
UPDATE main.hr.employees
SET salary = salary * 1.10
WHERE department = 'Engineering';

-- DELETE rows
DELETE FROM main.hr.employees
WHERE hire_date < '2020-01-01';

-- Query a specific version
SELECT * FROM main.hr.employees VERSION AS OF 2;

-- Query as of a timestamp
SELECT * FROM main.hr.employees TIMESTAMP AS OF '2024-01-10T00:00:00';

-- View table history
DESCRIBE HISTORY main.hr.employees;

-- Compare two versions (data diff)
SELECT * FROM main.hr.employees VERSION AS OF 4
EXCEPT
SELECT * FROM main.hr.employees VERSION AS OF 3;

-- Restore to a previous version (undo mistakes)
RESTORE TABLE main.hr.employees TO VERSION AS OF 3;

# Read a specific version
df_v2 = spark.read.format("delta") \
    .option("versionAsOf", 2) \
    .load("/mnt/data/silver/employees")

# Read as of a timestamp
df_past = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-10") \
    .load("/mnt/data/silver/employees")

# View table history programmatically
from delta.tables import DeltaTable
dt = DeltaTable.forPath(spark, "/mnt/data/silver/employees")
history = dt.history()
display(history.select("version", "timestamp", "operation", "operationMetrics"))

# Schema enforcement: this will FAIL if new_df has extra columns
new_df.write.format("delta").mode("append").save("/mnt/data/silver/employees")

# Enable schema evolution: automatically add new columns
new_df.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", True) \
    .save("/mnt/data/silver/employees")

# Overwrite schema entirely (replace all columns)
new_schema_df.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", True) \
    .save("/mnt/data/silver/employees")

-- Add a new column
ALTER TABLE main.hr.employees ADD COLUMN email STRING;

-- Change column type (where safe)
ALTER TABLE main.hr.employees ALTER COLUMN salary TYPE DECIMAL(12,2);

-- Rename a column
ALTER TABLE main.hr.employees RENAME COLUMN name TO full_name;

-- Add a table constraint
ALTER TABLE main.hr.employees ADD CONSTRAINT salary_positive CHECK (salary > 0);

-- Classic upsert: insert new, update existing
MERGE INTO main.hr.employees AS target
USING staging.employee_updates AS source
ON target.id = source.id
WHEN MATCHED THEN
    UPDATE SET
        target.name = source.name,
        target.department = source.department,
        target.salary = source.salary
WHEN NOT MATCHED THEN
    INSERT (id, name, department, salary)
    VALUES (source.id, source.name, source.department, source.salary);

-- Full CDC merge with deletes
MERGE INTO main.hr.employees AS target
USING cdc_changes AS source
ON target.id = source.id
WHEN MATCHED AND source.operation = 'DELETE' THEN DELETE
WHEN MATCHED AND source.operation = 'UPDATE' THEN
    UPDATE SET *
WHEN NOT MATCHED AND source.operation = 'INSERT' THEN
    INSERT *;

from delta.tables import DeltaTable

dt = DeltaTable.forPath(spark, "/mnt/data/silver/employees")

dt.alias("target").merge(
    updates_df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(
    set={
        "name": "source.name",
        "salary": "source.salary",
        "updated_at": "current_timestamp()"
    }
).whenNotMatchedInsert(
    values={
        "id": "source.id",
        "name": "source.name",
        "salary": "source.salary",
        "created_at": "current_timestamp()"
    }
).execute()

-- Compact small files into larger ones (target: 1GB per file)
OPTIMIZE main.hr.employees;

-- OPTIMIZE with Z-ORDER on frequently filtered columns
OPTIMIZE main.events.clickstream
ZORDER BY (user_id, event_date);

-- OPTIMIZE only specific partitions
OPTIMIZE main.events.clickstream
WHERE event_date >= '2024-01-01'
ZORDER BY (user_id);

-- View file statistics
DESCRIBE DETAIL main.hr.employees;

-- Check number of files and sizes
SELECT * FROM (
    DESCRIBE DETAIL main.hr.employees
);

-- Remove files older than 7 days (default retention)
VACUUM main.hr.employees;

-- Remove files older than 30 days
VACUUM main.hr.employees RETAIN 720 HOURS;

-- Dry run: see what would be deleted without deleting
VACUUM main.hr.employees RETAIN 168 HOURS DRY RUN;

from delta.tables import DeltaTable

dt = DeltaTable.forPath(spark, "/mnt/data/silver/employees")

# Vacuum with 7-day retention (168 hours)
dt.vacuum(retentionHours=168)

# WARNING: Setting retention to 0 hours loses ALL time travel
# Only do this if you are certain you do not need history
# spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
# dt.vacuum(retentionHours=0)