Delta Lake with Spark

Delta Lake is an open source project that enables building a Lakehouse architecture on top of data lakes. Delta Lake provides ACID transactions, scalable metadata handling, and unifies streaming and batch data processing on top of existing data lakes, such as S3, ADLS, GCS, and HDFS.

Delta Lake offers the following:

• ACID transactions in Spark: The serializable isolation levels ensure that readers never see inconsistent data.
• Scalable Metadata Management: Leverages Spark distributed processing power to handle all the metadata for petabyte-scale tables with billions of files at ease.
• Streaming and Batch Processing: A table in Delta Lake is a batch table as well as a streaming source and sink. Streaming data ingest, batch historic backfill, and interactive queries all just work out of the box.
• Schema Enforcement: Automatically manages schema variations to prevent the insertion of bad records during data ingestion.
• Time Travel: Data versioning enables rollbacks, full historical audit trails, and reproducible machine learning experiments.
• Upserts and deletes: Supports merge, update, and delete operations to enable complex use cases like change-data-capture, slowly-changing-dimension (SCD) operations, streaming upserts, and so on.

Spark/Delta compatibility matrix

Delta Lake Compatibility with Spark 3.5.1

For Spark 3.5.1, use Delta Lake 3.2.1. Delta Lake 3.3.x requires Spark 3.5.2 or later and isn't supported with Spark 3.5.1.

Using Delta Lake 3.3.x with Spark 3.5.1 can result in a NoClassDefFoundError for SupportsNonDeterministicExpression.

To use Delta Lake 3.3.x features, use Spark 3.5.5.

Spark Shell

Initialize the Spark shell with configurations optimized for working with Delta Lake.

ODP includes the Delta Lake JARs required for the supported Spark versions. To use the Delta Lake version included with ODP, start Spark with the Delta Lake session extension and catalog configuration.

/usr/odp/current/spark3-client/bin/spark-shell \
 --master yarn --deploy-mode client \
 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \
 --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog" \
 --conf "spark.sql.warehouse.dir=hdfs:///apps/hive/warehouse"

On Spark 4.1.1, replace the client path with /usr/odp/current/spark4-client/. The two --conf values are identical across all four Spark lines shipped with ODP 3.3.6.5.

Import Statements

Import the necessary libraries for Delta Lake to enable data manipulation and querying.

import io.delta.tables._

Create Table, Insert Data, and Read Data

Example code to create a Delta Lake table, insert data into it, and perform queries to retrieve the data.

spark.sql("CREATE TABLE if not exists delta_table (     ts BIGINT,     uuid STRING,     rider STRING,     driver STRING,     fare DOUBLE,     city STRING ) USING DELTA")
val columns = Seq("ts","uuid","rider","driver","fare","city")
val data =
 Seq((1695159649087L,"334e26e9-8355-45cc-97c6-c31daf0df330","rider-A","driver-K",19.10,"san_francisco"),
 (1695091554788L,"e96c4396-3fad-413a-a942-4cb36106d721","rider-C","driver-M",27.70 ,"san_francisco"),
 (1695046462179L,"9909a8b1-2d15-4d3d-8ec9-efc48c536a00","rider-D","driver-L",33.90 ,"san_francisco"),
 (1695516137016L,"e3cf430c-889d-4015-bc98-59bdce1e530c","rider-F","driver-P",34.15,"sao_paulo"),
 (1695115999911L,"c8abbe79-8d89-47ea-b4ce-4d224bae5bfa","rider-J","driver-T",17.85,"chennai"))
var inserts = spark.createDataFrame(data).toDF(columns:_*)
inserts.write.format("delta").saveAsTable("default.delta_tablev2") 
 
// Describe Table
val deltaTable = DeltaTable.forPath(spark, "hdfs:///path/to/delta_tablev2")
deltaTable.detail().show(false)
spark.table("default.delta_tablev2").show(false)

The default.delta_tablev2 name used in the write above is a Hive-catalog table (via DeltaCatalog). For the DeltaTable.forPath(...) API, pass the on-HDFS location of the table, which you can find with DESCRIBE DETAIL default.delta_tablev2 in Spark SQL (look at the location column).

Update Data and Read Data

Show how to update records in a Delta Lake table.

spark.sql("update default.delta_tablev2 set fare = fare * 10 where rider = 'rider-A'")
spark.table("default.delta_tablev2").show(false)

Merge Data and Read Data

Demonstrate merging data from one Delta Lake table into another and reading the resultant data.

spark.sql("CREATE TABLE fare_adjustment_delta (ts BIGINT, uuid STRING, rider STRING, driver STRING, fare DOUBLE, city STRING) USING delta")
spark.sql("INSERT INTO fare_adjustment_delta VALUES (1695091554788,'e96c4396-3fad-413a-a942-4cb36106d721','rider-C','driver-M',-2.70 ,'san_francisco'),(1695530237068,'3f3d9565-7261-40e6-9b39-b8aa784f95e2','rider-K','driver-U',64.20 ,'san_francisco'),(1695241330902,'ea4c36ff-2069-4148-9927-ef8c1a5abd24','rider-H','driver-R',66.60 ,'sao_paulo'    ), (1695115999911,'c8abbe79-8d89-47ea-b4ce-4d224bae5bfa','rider-J','driver-T',1.85,'chennai')")
spark.sql("MERGE INTO delta_tablev2 AS target USING fare_adjustment_delta AS source ON target.uuid = source.uuid WHEN MATCHED THEN UPDATE SET target.fare = target.fare + source.fare WHEN NOT MATCHED THEN INSERT * ")
spark.table("default.delta_tablev2").show(false)

Delete Data

Example code for deleting specific records from a Delta Lake table, demonstrating data management capabilities.

spark.sql("delete from delta_tablev2 where rider='rider-K'")
spark.table("default.delta_tablev2").show(false)

Time Travel Query

Use Delta Lake's time travel feature to query data from different historical versions of a table.

val history = spark.sql("DESCRIBE HISTORY default.delta_tablev2")
val latest_version = history.selectExpr("max(version)").collect()
spark.sql(s"select * from default.delta_tablev2 version as of ${latest_version(0)(0)}").show(false)

Register tables in Hive Metastore via DeltaCatalog

With ODP 3.3.6.5, the spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog configuration used in the Spark Shell section enables Delta tables to be registered in the Hive 4.1 metastore and read back through spark_catalog. No workaround config is required.

A minimal end-to-end example (works on Spark 3.3.3 / 3.5.1 / 3.5.5 / 4.1.1):

CREATE TABLE default.employees (id BIGINT, name STRING, dept STRING)
USING delta
LOCATION 'hdfs:///path/to/employees';
 
INSERT INTO default.employees VALUES (1,'alice','eng'), (2,'bob','sales'), (3,'carol','eng');
UPDATE default.employees SET dept = 'ops' WHERE id = 2;
DELETE FROM default.employees WHERE id = 3;
MERGE INTO default.employees t
 USING (SELECT 4 AS id, 'dave' AS name, 'eng' AS dept) s
 ON t.id = s.id
 WHEN MATCHED THEN UPDATE SET t.dept = s.dept
 WHEN NOT MATCHED THEN INSERT *;
DESCRIBE HISTORY default.employees;
SELECT * FROM default.employees VERSION AS OF 0;

The same script is validated on Spark 3.3.3 (Delta 2.3.0), Spark 3.5.1 (Delta 3.2.1), Spark 3.5.5 (Delta 3.3.3) and Spark 4.1.1 (Delta 4.3.1) against the ODP-3.3.6.5 Hive 4.1.0 metastore.

Users on pre-3.3.6.5 ODP releases (where the HMS shim fix is not present) should keep Delta on the path-based API (spark.read.format("delta").load(path) / .write.format("delta").save(path)) and avoid DeltaCatalog - registered tables, the HMS shim rejects Spark's get_table call on Hive 4.

For more details, see Delta Lake Documentation.

Cross-Spark-Version Compatibility

Delta tables that use the default Delta protocol can be read across the supported Spark versions.

Enabling Delta features that increase the minimum reader protocol, such as Column Mapping, Deletion Vectors, or Row Tracking, can prevent older Delta Lake versions from reading the table.

Cross-Spark-Version HMS-catalog interop

Reading via DeltaCatalog (Hive Metastore) across Spark versions. Once a Delta table is registered in Hive 4.1 HMS via DeltaCatalog from any Spark line, the other three Spark lines read it identically through SELECT * FROM default.<table_name>. The DDL was authored by one Spark version; every Spark version's DeltaCatalog resolves it to the same on-HDFS location and applies its own Delta reader.

This holds because ODP defaults do NOT enable protocol-raising Delta features. If a workload turns on Column Mapping (delta.columnMapping.mode=name), Deletion Vectors (delta.enableDeletionVectors=true), or Row Tracking (delta.enableRowTracking=true), the table's minReaderVersion rises above 2 and older Delta Lake versions (Delta 2.3.0 on Spark 3.3.3, Delta 3.2.1 on Spark 3.5.1) can no longer read it. Enable those features only when every reader in the pipeline can support them.

Delta Uniform Support

Delta UniForm generates Iceberg metadata alongside Delta metadata for the same Parquet files — no data duplication. Both Delta and Iceberg readers can query the same table.

Required setup:

• The Spark session must have enableHiveSupport() (or spark.sql.catalogImplementation=hive).
• spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension
• spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog
• Both jars on the driver + executor classpath: delta-spark_2.12:3.3.3 + delta-storage:3.3.3 + delta-iceberg_2.12:3.3.2 (all shipped in /usr/odp/current/spark3-client/jars/ on ODP 3.3.6.5-1012).

Create the table with Uniform properties AND an explicit LOCATION (managed-not-transactional tables fail Hive 4.1's strict-managed-table check — see the troubleshooting note below):

CREATE TABLE default.sales_uniform (id BIGINT, name STRING, region STRING)
USING delta
LOCATION 'hdfs:///path/to/sales_uniform'
TBLPROPERTIES (
 'delta.columnMapping.mode' = 'name',
 'delta.enableIcebergCompatV2' = 'true',
 'delta.universalFormat.enabledFormats' = 'iceberg',
 'delta.minReaderVersion' = '2',
 'delta.minWriterVersion' = '7'
);

Insert data using insertInto (avoid CREATE OR REPLACE — it triggers conversion before the catalog row is populated and errors with CatalogTable is empty in txn):

df.write.format("delta").mode("append").insertInto("default.sales_uniform")

Trigger the conversion explicitly with REORG (belt-and-braces — the converter fires on every commit, but REORG guarantees an eager Iceberg-metadata write for tests):

REORG TABLE default.sales_uniform APPLY (UPGRADE UNIFORM(ICEBERG_COMPAT_VERSION=2));

Where the Iceberg metadata lives — this is the single most important thing to know:

Iceberg metadata does NOT land under the table LOCATION when LOCATION is explicit. It goes to the Hive warehouse path keyed by table name. Any polling script that watches {table_location}/metadata/ will report "no metadata generated" even though the converter ran successfully.

Read back either way:

delta_df = spark.read.format("delta").table("default.sales_uniform")
iceberg_df = spark.read.format("iceberg").load("default.sales_uniform")
assert delta_df.count() == iceberg_df.count()

Or from Hive beeline (Hive 4.1 has the Iceberg storage handler built in — no aux jar):

SELECT COUNT(*) FROM default.sales_uniform;

HMS registers the table with both providers:

• spark.sql.sources.provider = delta (Delta view)
• storage_handler = org.apache.iceberg.mr.hive.HiveIcebergStorageHandler and table_type = ICEBERG (Iceberg view)
• metadata_location = hdfs://…/warehouse/tablespace/external/hive/{table}/metadata/{version}.metadata.json