Delta Lake

Overview

Delta Lake is an open-source storage layer that brings ACID transactions and time travel history to Apache Spark workloads.

What You Will Learn

Detailed Concept Explanation

Delta Lake runs on top of your existing cloud storage bucket or HDFS directory. Under the hood, Delta stores your data in standard Parquet files, but adds a Transaction Log (the _delta_log directory containing JSON transaction files).

Key Features

• ACID Transactions: Ensures data updates either succeed completely or fail with no partial writes.
• Schema Enforcement: Prevents writing data with invalid schemas, protecting tables from column corruption.
• Time Travel: The transaction log tracks changes, allowing you to query previous versions of your table (e.g. AS OF version or timestamp).

Code Examples

Input Dataset Preview

Below is the transactions dataset:

Python (PySpark) Implementation

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DeltaTest") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

data = [(101, "signup")]
df = spark.createDataFrame(data, ["userId", "action"])

# Write to Delta table format
df.write.format("delta").mode("overwrite").save("table.delta")

# Read from Delta table format
delta_df = spark.read.format("delta").load("table.delta")
delta_df.show()

Expected Output

+------+------+
|userId|action|
+------+------+
|   101|signup|
+------+------+

Execution Plan Diagram (Python & Scala)

Scala Implementation

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("DeltaScala")
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
  .getOrCreate()

import spark.implicits._
val df = Seq((101, "signup")).toDF("userId", "action")

df.write.format("delta").mode("overwrite").save("table.delta")

val deltaDF = spark.read.format("delta").load("table.delta")
deltaDF.show()

Expected Output

+------+------+
|userId|action|
+------+------+
|   101|signup|
+------+------+

SQL Implementation

-- Query historical version of a Delta table
SELECT * 
FROM delta.`table.delta` VERSION AS OF 1;

Expected Output

Executing this SQL query returns:

Common Mistakes

• Deleting Delta Log Directory: Deleting the _delta_log directory. Doing this turns the Delta table back into standard Parquet files, losing transaction history, time travel capabilities, and metadata indexes.

Best Practices

• Optimize regularly: Regularly run the OPTIMIZE command on your Delta tables to compact small partition files into larger ones, speeding up downstream reads.

Interview Perspective

Interactive Challenges

Related Topics

• Apache Iceberg
• Apache Hudi
• Parquet