How do I create a Delta table in open-source Spark?

Summary

• Install Delta Lake first. Open-source Spark needs the delta-spark package at a version that the Delta compatibility table pairs with your Spark version.
• Set spark.sql.extensions to io.delta.sql.DeltaSparkSessionExtension and spark.sql.catalog.spark_catalog to org.apache.spark.sql.delta.catalog.DeltaCatalog. The docs use these to enable SQL commands and metastore tables.
• Create the table with a write. Use df.write.format("delta").save(path) for a path table, saveAsTable for a metastore table, or SQL CREATE TABLE ... USING DELTA.
• Every write is validated and logged. Delta checks the schema and records each commit as a JSON file in _delta_log, which also powers time travel with versionAsOf.
• Delta Lake is licensed under Apache License 2.0 and is "not controlled by any single company". On Databricks it is the default table format.

How do I create a Delta table in open-source Spark?

Delta Lake is an open-source storage framework. In open-source Spark, creating a Delta table is a write with the Delta format, but you first install the Delta package and enable its Spark integration. The examples follow the Delta Lake docs and use local paths.

Which Delta Lake version matches my Spark version?

The Delta Lake releases page pairs versions like this:

Delta Lake	Apache Spark
4.1.x to 4.3.x	4.0.x, 4.1.x
4.0.x	4.0.x
3.0.x to 3.3.x	3.5.x

The quick start installs Delta Lake 4.0.0 with pip, then a compatible PySpark version:

pip install delta-spark==4.0.0
pip install pyspark==<compatible-spark-version>

How do I configure the SparkSession?

import pyspark
from delta import *

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

The two configs enable the DataSourceV2 and Catalog integration that metastore tables and SQL commands need. configure_spark_with_delta_pip makes the session download the Delta Lake JARs from Maven, which you need when you installed with pip and do not launch with --packages.

How do I create the table?

Method	Statement	Result
DataFrameWriter, path	df.write.format("delta").mode("overwrite").save("/tmp/delta/people10m")	Table defined by path
DataFrameWriter, metastore	df.write.format("delta").saveAsTable("default.people10m")	Metastore table with the DataFrame's schema
SQL DDL	CREATE TABLE IF NOT EXISTS default.people10m (id INT, firstName STRING) USING DELTA	Table with the declared schema
SQL, existing data	CREATE TABLE default.people10m USING DELTA LOCATION '/tmp/delta/people10m'	Inherits the schema, partitioning and properties of the data

A first table, straight from the quick start:

data = spark.range(0, 5)
data.write.format("delta").save("/tmp/delta-table")

df = spark.read.format("delta").load("/tmp/delta-table")
df.show()

Each write is a commit. Delta Lake records it as a JSON file in the _delta_log directory at the table root, named with the next version number zero-padded to 20 digits, such as _delta_log/00000000000000000000.json.

What do the write modes do?

Mode	Behavior
error or errorifexists (default)	Throws an exception if data already exists
append	Atomically adds new data to an existing Delta table
overwrite	Atomically replaces all the data in the table
ignore	Does not save the DataFrame if data already exists

How does schema validation work?

On every write Delta Lake checks that every DataFrame column exists in the table, that data types match, and that names do not differ only by case. Columns in the table but missing from the DataFrame are set to null. To add a column, set mergeSchema on the write:

new_rows.write.format("delta").mode("append") \
    .option("mergeSchema", "true").save("/tmp/delta/people10m")

Existing rows receive NULL for the new column. Delta Lake 4.3 and above also offers .withSchemaEvolution() and INSERT WITH SCHEMA EVOLUTION. The docs advise against enabling schema evolution session-wide.

How do I read old versions and clean up?

Pass versionAsOf before load, or use SQL VERSION AS OF:

df = spark.read.format("delta").option("versionAsOf", 0).load("/tmp/delta-table")

You can time travel only while both the log and the data files are retained. By default delta.logRetentionDuration is 30 days and delta.deletedFileRetentionDuration is 7 days. VACUUM is not triggered automatically, and after it runs you lose the ability to time travel to versions older than the retention period. VACUUM people10m DRY RUN lists the files it would delete.

How do I update or delete rows?

With the two configs set, SQL works directly on a metastore table such as people10m:

UPDATE people10m SET gender = 'Female' WHERE gender = 'F';
DELETE FROM people10m WHERE birthDate < '1955-01-01';

A delete removes data from the latest version only; the files stay in storage until old versions are vacuumed. Partitioning with partitionBy("gender") can speed up queries and DML that filter on the partition columns.

How does Databricks approach this?

On Databricks, Delta Lake is the default format for all operations, and unless specified otherwise every table is a Delta Lake table. The docs say that with Spark DataFrames or SQL you get its benefits "just by saving your data to the lakehouse with default settings".

FAQs

Is Delta Lake proprietary to Databricks?

No. Delta Lake is licensed under Apache License 2.0 and describes itself as "an independent open-source project and not controlled by any single company". It joined the Delta Lake Project in 2019, a sub-project of the Linux Foundation Projects.

Can I create a Delta table on S3?

Yes. Use an s3a:// path and add Hadoop's hadoop-aws connector plus S3 credentials. By default, concurrent writes to S3 must come from a single Spark driver; writes from several drivers can lose data.

How do I convert an existing Parquet table?

CONVERT TO DELTA parquet.`<path-to-table>`

This lists the files in the directory, creates a Delta transaction log that tracks them and infers the schema from the Parquet footers. Partitioned data needs the partition column types.

Can other engines read the table?

The Delta docs say the Apache Flink connector enables reading from and writing to Delta tables from Apache Flink.