Delta Lake

# Install: pip install delta-spark pyspark

from pyspark.sql import SparkSession
from delta.tables import DeltaTable

# 1. Create SparkSession with Delta Lake support
spark = SparkSession.builder \
    .appName("DeltaLakeDemo") \
    .config("spark.sql.extensions",
            "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 2. Create a Delta table from a DataFrame
data = [(1, "Alice", 50000), (2, "Bob", 60000), (3, "Carol", 55000)]
df = spark.createDataFrame(data, ["id", "name", "salary"])
df.write.format("delta").mode("overwrite").save("/tmp/employees")

# 3. Read the Delta table
delta_df = spark.read.format("delta").load("/tmp/employees")
delta_df.show()

# 4. Update a record (ACID transaction!)
dt = DeltaTable.forPath(spark, "/tmp/employees")
dt.update(
    condition="id = 2",
    set={"salary": "70000"}
)

# 5. TIME TRAVEL — read the original version (before update)
original = spark.read.format("delta") \
    .option("versionAsOf", 0) \
    .load("/tmp/employees")
original.show()   # Bob still shows salary = 60000!

# 6. MERGE (upsert) — update existing rows, insert new ones
new_data = [(2, "Bob", 75000), (4, "Dave", 65000)]
new_df = spark.createDataFrame(new_data, ["id", "name", "salary"])

dt.alias("existing").merge(
    new_df.alias("updates"),
    "existing.id = updates.id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

# 7. View Delta table history
dt.history().select("version", "timestamp", "operation").show()

# 8. Optimise (compact small files + Z-order for faster queries)
spark.sql("OPTIMIZE delta.`/tmp/employees` ZORDER BY (id)")