Here, we will talk about Delta Lake. Delta lake is an open source storage framework that bring reliability to data lakes. As we may know, data lakes have many limitations, such as that inconsistency and performance issues. Delta lake technology helps overcoming these challenges.
As we said, Delta Lake is an open source technology and not a proprietary technology. It's a storage framework or a storage layer, but it is not a storage format or a storage medium. It enables building lakehouse architecture. Lakehouse is a platform that unify both data warehouse and advanced analytics. So Delta Lake itself is not a data warehouse, and of course it's not a database service.
Delta Lake is a component which is deployed on the cluster as part of the Databricks runtime. If we are creating a Delta Lake table, it gets stored on the storage in one or more data files in parquet format. But along with these files, Delta stores a transaction log as well. But what is this transaction log ?
The Delta Lake transaction log, also known as Delta Log, is ordered records of every transaction performed on the table since its creation. It serves as a single source of truth. So every time we query the table, Spark checks this transaction log to retrieve the most recent version of the data. Each committed transaction is recorded in a JSON file.
It contains the operation that has been performed, whether, for example, it's an insert or update and the predicates such as conditions and filters used during this operation. In addition to all the files that have been affected because of this operation. Let us see some concrete examples.
In this scenario, we have a writer process and a reader process. Once the writer process starts, it stores the Delta Lake table in two data files in a parquet format. As soon as the writer process finishes writing, it adds the transaction log 000.json into the_delta_log directory. A reader process always starts by reading a transaction log. In this case, it reads the 000.json transaction log that contains information of the files number 1 and 2. So, it can start reading them.
In our second scenario, the writer process wants to update a record which presents in the file number 1,but in Delta Lake,instead of updating the record in the file itself, it will make a copy of this file and make the necessary updates in the new file. File number 3. It then updates the log by writing a new JSON file. This new log file knows that file number 1 is no longer needed. Now, the reader process reads the transaction log that tells that only files 2 and 3 are part of the current table version so it can start reading them.
Let us see one more scenario. Here, both processes want to work at the same time. The writer presses start writing the file number 4. On the other hand, the reader process reads the transaction log that only has information about files 2 and 3 and not file number 4 as it is not fully written yet. So it starts reading those two files, 2 and 3 which represent the most recent data at the moment. So as we can see here, Delta Lake guarantees that we will always get the most recent version of the data. Our read operation will never have a deadlock state or conflicts with any ongoing operation on the table. Finally, the writer process finishes and it adds a new file to the log.
Here is our last scenario. The writer presses start writing the file number 5 to the lake,but this time there is an error in the job, which lead to adding incomplete file. Because of this failure, Delta Lake module does not write any information to the log. Now, the reader process reads the transaction log that has no information about that incomplete file number 5. That's why the reader process will read only files 2, 3 and 4.
So as we can see Delta Lake guarantees that we will never read dirty data. Great. So that transaction log is the magic behind the scene. It allows Delta Lake to perform ACID transactions on data lakes. And it allows also to handle scalable metadata. This log also provides the full audit trail of all the changes that have happened on the table. And as we saw, the underlying file format for Delta is nothing but parquet and JSON format.
Here, we will work with Delta Lake tables. Let us first create an empty Delta Lake table. Like in SQL,we just need a CREAT TABLE statement, a table name, in our case, employees, and a table schema.
CREATE TABLE employees
(id INT, name STRING, salary DOUBLE);
Let us run our first command. Great. The table has been created. Let's confirm this. Let's go to the Data tab. Here, in the default database, we can see that the table employees has been created.
Here we can see the schema of the table, our three columns, ID, name, salary and other metadata information.
Let's come back to our notebook. Now we will insert some records all in a single transaction. Again, like in SQL, we will use INSERT INTO statements. Let's run our second command.
INSERT INTO employees
VALUES
(1, "Adam", 3500.0),
(2, "Sarah", 4020.5),
(3, "John",2999.3),
(4, "Thomas", 4000.3),
(5, "Anna", 2500.0),
(6, "Kim", 6200.3)
Here we can see that we have successfully inserted six records. Now we can simply query the table using a standard SELECT statement. Let us now see some metadata information about our table.
DESCRIBE DETAIL employees
It is an important command that allows us to explore table metadata. As we can see, there are many important information regarding our table here. For example, we can see the location of the table.
It is the location where the table files are really stored. In addition, we have also the number of file field, which indicates the number of data files in thecurrent table version. Let us copy the table location and explore the files using the %fs Magic Command.
%fs ls 'dbfs:/user/hive/warehouse/employees'
And indeed the directory contains four data files in parquet format.
In addition to the data log directory. But the question is why do we have four files for a single insert operation? This is because Spark work in parallel. If we check our cluster. We see that our cluster has four cores, so there were four executers working at the same time to insert our six new records.
Let us now come back to our notebook and see the scenario of update operations. In this scenario, we need to update the salary of all employees having a namestarts with the letter A by adding 100 to their salary.
UPDATE employees
SET salary = salary + 100
WHERE name LIKE 'A%'
Here we can see that there are two records affected by the update operation.
Let us query the table again to see the updated data. Let us now see what happened in the table directory.
We can see that there are two files have been added to the directory. As we said, rather than updating the records in the files themself, we make a copy of them. And later, Delta Lake use the transaction log to indicate which files are valid in the current version of the table. Let us confirm this by running the DESCRIBE DETAIL command.
Here,as we can see, the number of files are four and not six, and they are the four files that representthe current version of the table. So it contains the new files updated after our update command.
So if we query our delta table again. The query engine uses the transaction logs to resolve all the files that are valid in the current version and ignore all other data files.
And since the transaction log also stores all the changes to the data table, we can easily review the table history using the DESCRIBE HISTORY command.
As we can see, there are three versions of the table starting from version zero where we created the table. The version number 1 represents our insert command. So it is a write operation. And finally, our update command.
So as we can see, thanks to the transaction log, we have the full history of all operations thathave happened on the table. The transaction log is located under the _delta_log folder in the table directory. Let us explore this folder.
Each transaction is a new JSON file being written to the Delta lake transaction log. Here we can see that there are three JSON files representing the three transactions we have made onthe table. Starting from zero, the other file in the directory are just the checksum of the JSON files. Let us look inside the last file representing the update transaction.
For example, with the "add" element, we can see the new files that have been written to our table. And with the "remove" tags, we can see the list of files that have been soft deleted from our table. It means the file that no longer should be included in the table.
Here, we will talk about the advanced features of the Delta Lake. We will cover
Let us start with the time travel feature. With delta every operation on the table is automatically versioned, which provides the full audit trail of all the changes that have happened on the table. we can look at the history of the table in SQL using Describe History Command. In addition, we can query older version of the table. This can be done in two different ways, either using a timestamp. So in a SELECT statement we use the keyword TIMESTAMP AS OF and we provide the timestamp or date string. The second way is to use a version number. Since every operation on the table has a version number, we can use this version number to travel back time as well. Here using the keyword VERSION AS OF,or simply @v, which is the short syntax.
Time travel also makes it easy to do rollbacks in case of bad writes. For example, if our pipeline job had a bug that accidentally deleted user information, we can easily fix this using the RESTORE TABLE command. Either to restore the table to specific timestamp or to specific version number.
The second important feature here is compacting small files. In fact, Delta Lake can improve the speed of read queries from a table. One way to improve this speed is by compacting small files into larger ones. we trigger compaction simply by running the OPTIMIZE command. For example, if we have many small files by running the OPTIMIZE command, they will be compacted in one or more larger files which improves the table performance.
With OPTIMIZE, we can also do z-order indexing. Z-Order indexing in Delta Lake is about co-locating and reorganizing column information in the same set of files. This can be done by adding the ZORDER BY keyword to the OPTIMIZE command, followed by one or morecolumn name. So, for example, if we have a numerical column in the data files, ID for example, by applying the ZORDER on this column, the first compacted file will contain values from 1 to 50, while the other one will contain values from 51 to 100.
Z Order indexing is used by data skipping algorithm to extremely reduce the amount of data that need to be read. In our example, if we query an ID, say 30, Delta is sure now that ID 30 is in file number one, so it can easily skip the scanning the file number two, which will save a huge amount of time.
Lastly, we might be wondering what about unused data files like uncommitted files and files that are no longer in the latest state of the transaction log for the table?
In fact, Delta Lake allows us to do garbage collection by using Vacuum Command. With Vacuum command,we just need to specify the threshold of retention period for the files, so we delete all the files older than this threshold. By default, the threshold is 7 days. This means that vacuum operation will prevent us from deleting files less than 7 days old. Just to be sure that no longer running operations are still referencing any of the files to be deleted. But remember, once we run a vacuum on a Delta table, we lose the ability to time and travel back to a version older than the specified retention period, simply because the data files are no longer exist.
Here, we will see some advanced concepts in Delta Lake, such as the time travel feature. In addition, we will see Optimize and Vacuum commands. Let us review again our table history.
DESCRIBE HISTORY employees
So, here we can see the three versions of our table.
In Delta Lake,we can easily query previous versions of our table, and this feature of time travel is possible thanks to those extra data files that had been marked as removed in our transaction log. So, let us say we want to access our data before the update operation, which is the version number 1. We can simply use SELECT query with VERSION AS OF keyword, and we specify the version number,in our case, it is version 1. We can instead use timestamp to query data at the time of interest.
SELECT *
FROM employees VERSION AS OF 1
-- OR
SELECT *
FROM employees TIMESTAMP AS OF <timestamp>
So, here we see our data before the update operation. Another alternative syntax is to use @v followed by the version number.
SELECT *
FROM employees@v1
Now, imagine this scenario,We deleted our data and we need to restore them.
Here, -1 means that all data has been removed. We can simply roll back to a previous version before deletion using the RESTORE TABLE command.
Let us explore what really happened in our table.
As we can see, the RESTORE command has been recorded as a transaction. So, as we can see, Delta time travel is really a powerful feature.
Let us now talk about the OPTIMIZE command and how to compact small files, and do Z order indexing. Since the spark work in parallel, we usually end up by writing too many small files. Having many small data files negatively affect the performance of the Delta table.
In our case, we have 4 small data files. To resolve this issue,we can use OPTIMIZE command that combine files toward an optimal size. OPTIMIZE will replace existing data files by combining records and rewriting the results. Here, we can see that our 4 data files have been soft deleted, and a new file has been added that compacts those 4 files. In addition, we may noticed that we added the ZORDER indexing with our OPTIMIZE command. ZORDER indexing speeds up data retrieval when filtering on provided fields, by grouping data with similar values within the same data files.
In our case, we do Zorder by the ID column. However, on such a small data set, it does not provide any benefit. Let us confirm the output of the OPTIMIZE command by running DESCRIBE DETAIL on our table.
Here, we can see that the number of files in the current version is only one. Let us see how the OPTIMIZE operation has been recorded in our table history.
