Delta Lake & ACID

Traditional data lakes (like folders of raw CSV or Parquet files on S3/HDFS) suffer from severe limitations:

• No ACID Transactions: If a write fails halfway, corrupted or duplicate files are left on disk.
• No Updates or Deletes: Modifying a row requires reading the entire table, making updates, and rewriting the whole dataset.
• No Schema Enforcement: A bad dataset with wrong columns can be written directly to the lake, corrupting downstream dashboards.

Delta Lake is an open-source storage layer that sits on top of your existing parquet files, bringing ACID transactions and relational database capabilities to distributed object stores.

Core Features of Delta Lake

1. ACID Transactions: Guarantees that write operations are atomic (either all tasks complete or none do), preventing corrupt/partial files.
2. Schema Enforcement: Automatically rejects any write that does not match the target table's schema, preventing dirty data.
3. Unified Batch & Streaming: Data can be written to and read from Delta tables concurrently as batch dataframes or real-time streams.
4. DML Support: Enables native UPDATE, DELETE, and MERGE (upserts) operations on distributed data files.

PySpark Code Example: ACID writes & DML Operations

To use Delta Lake, initialize Spark with the Delta jar package coordinates:

from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip

# 1. Initialize Spark Session configured with Delta Lake
builder = SparkSession.builder \
    .appName("Delta Lake ACID") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .master("local[*]")

spark = configure_spark_with_delta_pip(builder).getOrCreate()

# 2. Write standard DataFrame as a Delta Table
data = [(1, "Alice", 90000), (2, "Bob", 60000)]
df = spark.createDataFrame(data, ["id", "name", "salary"])

df.write \
  .format("delta") \
  .mode("overwrite") \
  .save("delta_storage_directory")

# 3. Perform a Delta DML Update (Requires DeltaTable object)
from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "delta_storage_directory")

# Update Alice's salary to 95000
deltaTable.update(
    condition = "name = 'Alice'",
    set = { "salary": "95000" }
)

# 4. Perform a MERGE (Upsert)
# Merge acts as: If key exists, UPDATE. If key is new, INSERT.
new_records = [(2, "Bob", 65000), (3, "Charlie", 70000)] # Bob is an update; Charlie is new
new_df = spark.createDataFrame(new_records, ["id", "name", "salary"])

deltaTable.alias("target").merge(
    source = new_df.alias("source"),
    condition = "target.id = source.id"
).whenMatchedUpdate(set = {
    "salary": "source.salary"
}).whenNotMatchedInsert(values = {
    "id": "source.id",
    "name": "source.name",
    "salary": "source.salary"
}).execute()

# 5. Read and view the final conformed Delta table
conformed_df = spark.read.format("delta").load("delta_storage_directory")
conformed_df.show()