Writing your Delta table

When creating a Delta table, you are writing files to some storage (e.g. file system, cloud object stores). All of the files together stored in a directory of a particular structure (more on this later) make up your table. Therefore, when we create a Delta table, we are in fact writing files to some storage location. For example, the following Spark code snippet takes an existing Spark DataFrame and writes it in the Apache Parquet storage format in the folder location /data using the Spark DataFrame API.

              dataframe.write.format("parquet").save("/data")
            

With one simple modification of the preceding code snippet, you can now create a Delta table.

Figure 1-2. Instead of parquet, simply say delta

Let’s see how that looks in code by using Apache Spark to create a Delta table from a DataFrame. In the following code example, we will first create the Spark DataFrame data and then use the write method to save the table to storage.

%python
# Create data DataFrame
data = spark.range(0, 5)
# Write the data DataFrame to /delta location
data.write.format("delta").save("/delta")

%scala
// Create data DataFrame
val data = spark.range(0, 5)
// Write the data DataFrame to /delta location
data.write.format("delta").save("/delta")

It is important to note that in most production environments when you are working with large amounts of data, it is important to partition your data. The following example partitions your data by date (a commonly followed best practice):

%python
# Write the Spark DataFrame to Delta table partitioned by date
data.write.partitionBy("date").format("delta").save("/delta")

%scala
// Write the Spark DataFrame to Delta table partitioned by date
data.write.partitionBy("date").format("delta").save("/delta")

If you already have an existing Delta table and would like to append or overwrite data to your table, include the mode method in your statement.

%python
># Append new data to your Delta table
data.write.format("delta").mode("append").save("/delta")

# Overwrite your Delta table
data.write.format("delta").mode("overwrite").save("/delta")

%scala
// Append new data to your Delta table
data.write.format("delta").mode("append").save("/delta")

// Overwrite your Delta table
data.write.format("delta").mode("overwrite").save("/delta")

Reading your Delta table

Similar to writing your Delta table, you can use the DataFrame API to read the same files from your Delta table.

%python
# Read the data DataFrame from the /delta location
spark.read.format("delta").load("/delta").show()

%scala
// Read the data DataFrame from the /delta location
spark.read.format("delta").load("/delta").show()

You can also read the table using SQL by specifying the file location after specifying delta.

%sql
SELECT * FROM delta.`/delta`

The output of this table can be seen below.

+---+
| id|
+---+
|  4|
|  2|
|  3|
|  0|
|  1|
+---+

Reading your metastore defined Delta table

In the previous sections, we have been reading our Delta tables directly from the file system. But how would you read a metastore defined Delta table? For example, instead of reading our Delta table in SQL using the file path:

%sql
SELECT * FROM delta.`/delta`

how would you read using a metastore-defined table such as:

%sql
SELECT * FROM myTable

To do this, you would need to first define the table within the metastore using the saveAsTable method or a CREATE TABLE statement.

%python
# Write the data DataFrame to the metastore defined as myTable
data.write.format("delta").saveAsTable("myTable")

%scala
// Write the data DataFrame to the metastore defined as myTable
data.write.format("delta").saveAsTable("myTable")

Note, when using the saveAsTable method, you will save the Delta table files into a location managed by the metastore (e.g. /user/hive/warehouse/myTable). If you want to use SQL or control the location of your Delta table, then you initially save it using the save method where you specify the location (e.g. /delta) and then create the table using the following SQL statements.

%sql
-- Create a table in the metastore
CREATE TABLE myTable (
  id INTEGER)
USING DELTA
LOCATION "/delta"

As noted previously, it will be important to partition your large tables.

%sql
CREATE TABLE id (
  date DATE
  id INTEGER)
USING DELTA
PARTITION BY date
LOCATION "/delta"

Note, the LOCATION property that points to the underlying files that make up your Delta table.

Single Source of Truth

Delta Lake is built on top of Apache Spark™ in order to allow multiple readers and writers of a given table to all work on the table at the same time. To show users correct views of the data at all times, the Delta Lake transaction log serves as a single source of truth – the central repository that tracks all changes that users make to the table. You could compare this transaction log feature in Delta Lake as a single source of truth to the .git directory in a git-managed source code repository.

The concept of a single source of truth is important because, over time, processing jobs will fail in your data lake. The result is that processing jobs or tasks will leave partial files that do not get cleaned up. Subsequent processing or queries will not be able to ascertain which files should or should not be included in their queries.

Figure 1-4. Partial file example

For example, the preceding diagram represents a common data lake processing scenario.

By using a transaction log to track which files are valid, we can avoid the preceding scenario. Thus, when a user reads a Delta Lake table for the first time or runs a new query on an open table that has been modified since the last time it was read, Spark checks the transaction log to see what new transactions have posted to the table, and then updates the end user’s table with those new changes. This ensures that a user’s version of a table is always synchronized with the master record as of the most recent query and that users cannot make divergent, conflicting changes to a table.

Let’s repeat the same partial file example except for this time we run this on a Delta table with the Delta transaction log.

Figure 1-5. Partial file example with Delta transaction log

The exact same actions happened as in the previous example but this time the Delta transaction log provides atomicity and ensures data reliability.

The Implementation of Atomicity on Delta Lake

One of the four properties of ACID transactions, atomicity, guarantees that operations (like an INSERT or UPDATE) performed on your data lake either complete fully or don’t complete at all. Without this property, it’s far too easy for a hardware failure or a software bug to cause data to be only partially written to a table, resulting in messy or corrupted data.

The transaction log is the mechanism through which Delta Lake is able to offer the guarantee of atomicity - if it’s not recorded in the transaction log, it never happened. By only recording transactions that execute fully and completely, and using that record as the single source of truth, the Delta Lake transaction log allows users to reason about their data, and have peace of mind about its fundamental trustworthiness, at petabyte scale.

Breaking Down Transactions Into Atomic Commits

Whenever a user performs an operation to modify a table (e.g., INSERT, DELETE, UPDATE or MERGE), Delta Lake breaks that operation down into a series of discrete steps composed of one or more of the actions below.

Update metadata

Updates the table’s metadata including but not limited to changing the table’s name, schema or partitioning.

Add file

adds a data file to the transaction log.

Remove file

removes a data file from the transaction log.

Set transaction

Records that a structured streaming job has committed a micro-batch with the given ID.

Change protocol

enables new features by switching the Delta Lake transaction log to the newest software protocol.

Commit info

Contains information around the commit, which operation was made, from where, and at what time.

Those actions are then recorded in the transaction log as ordered, atomic units known as commits.

Delta Transaction Log Protocol

In this section, we describe how the Delta transaction log brings ACID properties to large collections of data, stored as files, in a distributed file system or object-store. The protocol was designed with the following goals in mind:

Serializable ACID Writes

multiple writers can concurrently modify a Delta table while maintaining ACID semantics. (Watch this video on how each concurrent writes are handled)

Snapshot Isolation for Reads

readers can read a consistent snapshot of a Delta table, even when multiple writers are concurrently writing.

Scalability to billions of partitions or files

queries against a Delta table can be planned on a single machine or in parallel.

Self-describing

all metadata for a Delta table is stored alongside the data. This design eliminates the need to maintain a separate metastore just to read the data and also allows static tables to be copied or moved using standard filesystem tools.

Support for incremental processing

readers can tail the Delta log to determine what data has been added in a given period of time, allowing for efficient streaming.

Logstore

Think about the existence of the delta files for a second. The logs, versions, and files that are being generated must exist somewhere, some system or store for files. LogStore is the general interface for all critical file system operations required to read and write the Delta transaction log. Because most storage systems do not provide atomicity guarantees out-of-the-box, Delta Lake transactional operations go through the LogStore API instead of accessing the storage system directly.

• Any file written through this store must be made visible, atomically. In other words, it should be visible in its entirety or not visible at all. It should not generate partial files.

• Only one writer must be able to create a file at the final destination. This is because many processes can occur simultaneously and to ensure decent speed, many writers write to their own files in parallel.

• The Logstore offers ACID consistent listing of files

The Delta Lake Transaction Log at the File Level

As noted previously, when a Delta table is created, that table’s transaction log is automatically created in the _delta_log subdirectory. As they make changes to that table, those changes are recorded as ordered, atomic commits in the transaction log. Each commit is written out as a JSON file, starting with 000000.json. Additional changes to the table generate subsequent JSON files in ascending numerical order so that the next commit is written out as 000001.json, the following as 000002.json, and so on. Each numeric JSON file increment represents a new version of the table.

Figure 1-6. Delta on disk

Implementing Atomicity

An important thing to understand is the transaction log is the single source of truth for your Delta table. So any reader that’s reading through your Delta table, will take a look at the transaction log first. Therefore changes to the table are stored as ordered atomic units called commits.

To continue the above example, we add additional records to our Delta table from the data files 1.parquet and 2.parquet. That transaction would automatically be added to the transaction log, saved to disk as commit 000000.json. Then, perhaps we change our minds and decide to remove those files (e.g., run a DELETE from table operation) and add a new file instead (3.parquet) through an INSERT operation. Those actions would be recorded as the next commit - and table version - in the transaction log, as 000001.json, as shown in the following diagram.

Figure 1-7. Implementing atomicity in Delta

Even though 1.parquet and 2.parquet are no longer part of our Delta Lake table, their addition and removal are still recorded in the transaction log because those operations were performed on our table – despite the fact that they ultimately canceled each other out. Delta Lake still retains atomic commits like these to ensure that in the event we need to audit our table or use “time travel” (to discuss in chapter 3) to see what our table looked like at a given point in time, we could do so accurately.

To provide context, let’s continue with our notebook and read the JSON file that makes up the first transaction (version 1) inserting data into our Delta table.

>%python
# Read the transaction log of version 1
j0 = spark.read.json("/.../000001.json")

%scala
// Read the transaction log of version 1
val j0 = spark.read.json("/.../000001.json")

Figure 1-8. Reviewing the transaction log structure

The transaction contains many pieces of information stored as strings or columns within the transaction log JSON. Let’s focus on the commit, add, and CRC pieces.

%python
# Commit Information
display(j0.select("commitInfo").where("commitInfo is not null"))

%scala
// Commit Information
display(j0.select("commitInfo").where("commitInfo is not null"))

clusterId: "0127-045215-pined152"
isBlindAppend: true
isolationLevel: "WriteSerializable"
notebook: {"notebookId": "8476282"}
operation: "STREAMING UPDATE"
operationMetrics: {"numAddedFiles": "1", "numOutputBytes": "492", "numOutputRows": "0", "numRemovedFiles": "0"}
operationParameters: {"epochId": "0", "outputMode": "Append", "queryId": "892c1abd-581f-4c0f-bbe7-2c386feb3dbd"}
readVersion: 0
timestamp: 1603581133495
userId: "100599"userName: "denny[dot]lee[@]databricks.com"

%python
# Commit Information
display(j0.select("add").where("add is not null"))

%scala
// Commit Information
display(j0.select("add").where("add is not null"))

dataChange: true
modificationTime: 1603581134000
path: "part-00000-95648a41-bf33-4b67-9979-1736d07bef0e-c000.snappy.parquet"
size: 492
stats: "{"numRecords":0,"minValues":{},"maxValues":{},"nullCount":{}}"

                %sh head /dbfs/ml/loan_by_state/_delta_log/00000000000000000001.crc
              

{
"tableSizeBytes":1855,
"numFiles":2,
"numMetadata":1,
"numProtocol":1,
"numTransactions":1
}

Quickly Recomputing State With Checkpoint Files

Prior to this, our focus is that of a single transaction log in the form of JSON files. But for large-scale systems or any streaming system, this would result in creating the “small-file” problem where it becomes ever more inefficient to query the transaction log folder (i.e. _delta_log subdirectory).

To alleviate this issue, Delta Lake creates a checkpoint file in Parquet format after it creates the 10th commits (i.e. transaction).

Figure 1-9. Create transaction log after every 10th commit

These checkpoint files save the entire state of the table at a point in time – in native Parquet format that is quick and easy for Spark to read. In other words, they offer the Spark reader a sort of “shortcut” to fully reproducing a table’s state that allows Spark to avoid reprocessing what could be thousands of tiny, inefficient JSON files.

When recomputing the state of the table, Spark will read and cache the available JSON files that make up the transaction log. For example, if there have been only three committed operations or commits to the table (including the table creation), Spark will read all three files and cache the results into memory (i.e. cache version 2).

Figure 1-10. Spark caching after the third commit

Instead of continually reading the transaction log for this information, all of the Spark readers requesting data from this table can simply reference the cached copy of Delta’s state. As more commits are performed against the Delta table, more JSON files will be added to the _delta_log folder.

Figure 1-11. Spark caching after the eight commit

To continue this example, let’s say this table five additional commits, then Spark will cache version 7 of the data. Note, at this point, Spark will list all of the transactions from version 0 instead of reading from version 2 to ensure that earlier transactions have completed.

Figure 1-12. Spark caching after the 13th commit including checkpoint creation

As noted earlier, Delta Lake will create a checkpoint file (0000010.checkpoint.parquet) after the 10th commit. Delta Lake will still listFrom version 0 to avoid late transactions and cache version 12.

To read the checkpoint file, run the following statement.

%python
# Review the transaction log checkpoint file
chkpt0 = spark.read.parquet("/ml/loan_by_state/_delta_log/000010.checkpoint.parquet")

%scala
// Review the transaction log checkpoint file
val chkpt0 = spark.read.parquet("/ml/loan_by_state/_delta_log/000010.checkpoint.parquet")

The metadata you will read is a union of all of the previous transactions. This becomes apparent when you read the query for the add information.

%python
# Add Information
display(chkpt0.select("add").where("add is not null"))

%scala
// Add Information
display(chkpt0.select("add").where("add is not null"))

Figure 1-13. Add information in the checkpoint file

In addition to the checkpoint file being in Parquet format (thus Spark can read it even faster) and containing all of the transactions prior to it, notice how the stats in the original JSON file were in string format.

stats: "{"numRecords":7,"minValues":{"addr_state":"IA","count":3,"stream_no":3},"maxValues":{"addr_state":"TX","count":9,"stream_no":3},"nullCount":{"addr_state":0,"count":0,"stream_no":0}}"
            

As part of the checkpoint creation process, there is a stats_parsed column that contains the statistics as nested columns instead of strings.

stats_parsed:{
"numRecords": 7, 
"minValues": {
"addr_state": "IA", 
"count": 3, 
"stream_no": 3
}, 
"maxValues": {
"addr_state": "TX", 
"count": 9, 
"stream_no": 3
}, 
"nullCount": {
"addr_state": 0, 
"count": 0, 
"stream_no": 0
}
}

By using nested columns instead of strings, Spark can read the statistics significantly faster especially when it needs to read all of the files created for petabyte-scale data lakes.

Figure 1-14. Spark caching after the 15th commit including checkpoint creation

As it is important to note that once the Parquet checkpoint file has been created, subsequent listFrom calls are from the checkpoint file instead of going back to version 0. Thus after the 15 commits, Delta will cache version 14 but need only to listFrom the version 10 checkpoint file.

What Is Optimistic Concurrency Control?

Optimistic concurrency control is a method of dealing with concurrent transactions that assumes that transactions made to a table by different users can complete without conflicting with one another. It is incredibly fast because when dealing with petabytes of data, there’s a high likelihood that users will be working on different parts of the data altogether, allowing them to complete non-conflicting transactions simultaneously.

When working with tables, as long as different clients are modifying different parts of the table (e.g. different partitions) or performing actions that do not conflict (e.g. two clients reading from the table), those operations do not conflict so we can optimistically let them all complete their task. But for situations where clients modify the same parts of the table concurrently, Delta Lake has a protocol to resolve this.

Solving conflicts optimistically

Ensuring serializability

Another key piece that Delta requires for consistent guarantees is mutual exclusion. We need to agree on the order of changes, even when there are multiple writers and this provides a guarantee in databases called serializability. The fact that even though things are happening concurrently, you could play them as if they happened in a synchronous ordered manner.

For example, we have user 1 reading 000000.json

Figure 1-15. User 1 reading

And we have user2 that reads 000001.json

.

Figure 1-16. Users 1 and 2 reading the transaction log concurrently

As you’re trying to commit 000002.json, user2 two wins and user1 takes a look and sees that, 000002.json is already there. Due to this requirement of mutual exclusion, it will have to say, “Oh this commit failed” and let me try to commit 000003.json instead.

Figure 1-17. Ensuring serializability

Applying Optimistic Concurrency Control in Delta

In order to offer ACID transactions, Delta Lake has a protocol for figuring out how commits should be ordered (known as the concept of serializability in databases) and determining what to do in the event that two or more commits are made at the same time.

Delta Lake handles these cases by implementing a rule of mutual exclusion, then attempting to solve any conflict optimistically. This protocol allows Delta Lake to deliver on the ACID principle of isolation, which ensures that the resulting state of the table after multiple, concurrent writes is the same as if those writes had occurred serially, in isolation from one another.

In general, the process proceeds like this:

1. Record the starting table version.

2. Record reads/writes.

3. Attempt a commit.

4. If someone else wins, check whether anything you read has changed.

5. Repeat.

For example, if two users read from the same table and then both attempt to insert data at the same time.

Figure 1-18. Solving conflicts optimistically

• Delta Lake records the starting table version of the table (version 0) that is read prior to making any changes.

• Users 1 and 2 both attempt to append some data to the table at the same time. Here, we’ve run into a conflict because only one commit can come next and be recorded as 000001.json.

• Delta Lake handles this conflict with the concept of mutual exclusion, i.e. only one user can successfully commit 000001.json. User 1’s commit is accepted, while User 2’s is rejected.

• Rather than throw an error for User 2, Delta Lake prefers to handle this conflict optimistically. It checks to see whether any new commits have been made to the table, and updates the table silently to reflect those changes, then simply retries User 2’s commit on the newly updated table (without any data processing), successfully committing 000002.json.

In the vast majority of cases, this reconciliation happens silently, seamlessly, and successfully. However, in the event that there’s an irreconcilable problem that Delta Lake cannot solve optimistically (for example, if User 1 deleted a file that User 2 also deleted), the only option is to throw an error.

As a final note, since all of the transactions made on Delta Lake tables are stored directly to storage, this process satisfies the ACID property of durability, meaning it will persist even in the event of system failure.

Time Travel

Every table is the result of the sum total of all of the commits recorded in the Delta Lake transaction log – no more and no less. The transaction log provides a step-by-step instruction guide, detailing exactly how to get from the table’s original state to its current state.

Therefore, we can recreate the state of a table at any point in time by starting with an original table, and processing only commits made prior to that point. This powerful ability is known as “time travel,” or data versioning, and can be a lifesaver in any number of situations. For more information, read the blog post Introducing Delta Time Travel for Large Scale Data Lakes, or refer to the Delta Lake time travel documentation.

Data Lineage and Debugging

As the definitive record of every change ever made to a table, the Delta Lake transaction log offers users a verifiable data lineage that is useful for governance, audit, and compliance purposes. It can also be used to trace the origin of an inadvertent change or a bug in a pipeline back to the exact action that caused it. Users can run DESCRIBE HISTORY to see metadata around the changes that were made.

Configure Log and Data History

Warning

We do not recommend that you set a retention interval shorter than 7 days because old snapshots and uncommitted files can still be in use by concurrent readers or writers to the table. If vacuum cleans up active files, concurrent readers can fail or, worse, tables can be corrupted when vacuum deletes files that have not yet been committed.

Delta Lake has a safety check to prevent you from running a dangerous vacuum command. If you are certain that there are no operations being performed on this table that take longer than the retention interval you plan to specify, you can turn off this safety check by setting the Apache Spark configuration property spark.databricks.delta.retentionDurationCheck.enabled to false. You must choose an interval that is longer than the longest-running concurrent transaction and the longest period that any stream can lag behind the most recent update to the table.

DESCRIBE TABLE

Describe Table returns the basic metadata information of a table. The metadata information includes column name, column type and column comment. Optionally you can specify a partition spec or column name to return the metadata pertaining to a partition or column respectively.

Display detailed information about the specified columns, including the column statistics collected by the following command. Note, this command is applicable to only metastore-defined tables.

%sql
DESCRIBE TABLE loan_by_state;
 
# result

+---------------+---------+-------+
|       col_name|data_type|comment|
+---------------+---------+-------+
|     addr_state|   string|       |
|          count|   bigint|       |
|      stream_no|      int|       |
|               |         |       |
| # Partitioning|         |       |
|Not partitioned|         |       |
+---------------+---------+-------+

Along with the basic metadata information and partitioning, DESCRIBE TABLE EXTENDED returns detailed table information such as parent database, table name, location of table, provider, table properties, etc.

%sql
DESCRIBE TABLE EXTENDED loan_by_state;

+--------------------+--------------------+-------+
|            col_name|           data_type|comment|
+--------------------+--------------------+-------+
|          addr_state|              string|       |
|               count|              bigint|       |
|           stream_no|                 int|       |
|                    |                    |       |
|      # Partitioning|                    |       |
|     Not partitioned|                    |       |
|                    |                    |       |
|# Detailed Table ...|                    |       |
|                Name|denny_db.loan_by_...|       |
|            Location|dbfs:/ml/loan_by_...|       |
|            Provider|               delta|       |
|    Table Properties|[delta.checkpoint...|       |
+--------------------+--------------------+-------+

For the current schema of the columns described by DESCRIBE TABLE, refer to Spark SQL Guide > DESCRIBE TABLE.

Clone your Delta Tables

Table cloning or creating copies of tables in a data lake or data warehouse has several practical uses. But, given the expansive volume of data in tables in a data lake and the rate of its growth, making physical copies of tables is an expensive operation. Using CLONE makes the process simpler and cost-effective with the help of table clones.

What are clones anyway?

Clones are replicas of a source table at a given point in time. They have the same metadata as the source table: same schema, constraints, column descriptions, statistics, and partitioning. However, they behave as a separate table with a separate lineage or history. Any changes made to clones only affect the clone and not the source. Any changes that happen to the source during or after the cloning process also do not get reflected in the clone due to snapshot isolation. As of this writing, there are two types of clones: shallow or deep.

Where do clones help?

There are many scenarios where you need a copy of your datasets – for exploring, sharing, or testing ML models or analytical queries. Below are some example use cases.

%sql
-- Shallow clone table
CREATE TABLE delta.`/some/test/location` SHALLOW CLONE prod.events

%python
# Shallow clone table
DeltaTable.forName("spark", "prod.events").clone("/some/test/location", isShallow=True)

%scala
// Shallow clone table
DeltaTable.forName("spark", "prod.events").clone("/some/test/location", isShallow=true)

%sql
-- Create a shallow clone
CREATE TABLE temp.staged_changes SHALLOW CLONE prod.events;

-- Test out deleting table from shallow clone table
DELETE FROM temp.staged_changes WHERE event_id is null;

-- Update shallow clone table
UPDATE temp.staged_changes SET change_date = current_date() WHERE change_date is null;
...
-- Perform your verifications

%sql
-- If no changes have been made to the source
REPLACE TABLE prod.events CLONE temp.staged_changes;

-- If the source table has changed
MERGE INTO prod.events USING temp.staged_changes
ON events.event_id <=> staged_changes.event_id
WHEN MATCHED THEN UPDATE SET *;

-- Drop the staged table
DROP TABLE temp.staged_changes;

# Run your ML workloads using Python and then
DeltaTable.forName(spark, "feature_store").cloneAtVersion(128, "feature_store_bf2020")

%sql
-- Data migration using deep clone
CREATE TABLE delta.`zz://my-new-bucket/events` CLONE prod.events;
ALTER TABLE prod.events SET LOCATION 'zz://my-new-bucket/events';

%sql
-- The following code can be scheduled to run at your convenience
-- for data sharing
CREATE OR REPLACE TABLE data_science.events CLONE prod.events;

%sql
-- The following code can be scheduled to run at your convenience
-- for data archiving
CREATE OR REPLACE TABLE archive.events CLONE prod.events;

For more on clones

For more information on clones:

• Delta CLONE Language Manual

• How to Easily Clone Your Delta Lake Data Tables with Databricks

• Attack of the Delta Clones (Against Disaster Recovery Availability Complexity)

• We will have an Operational scenarios chapter for more examples