Databricks

Delta Lake: Up and Running

Modern Data Lakehouse Architectures with Delta Lake

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

Bennie Haelen and Dan Davis

Delta Lake: Up and Running

by Bennie Haelen and Dan Davis

Copyright © 2023 O’Reilly Media Inc. All rights reserved.

Printed in the United States of America.

Published by O’Reilly Media, Inc., 1005 Gravenstein Highway North, Sebastopol, CA 95472.

O’Reilly books may be purchased for educational, business, or sales promotional use. Online editions are also available for most titles (http://oreilly.com). For more information, contact our corporate/institutional sales department: 800-998-9938 or corporate@oreilly.com.

Editors: Gary O’Brien and Aaron Black

Production Editor: Beth Kelly

Copyeditor: FILL IN COPYEDITOR

Proofreader: FILL IN PROOFREADER

Indexer: FILL IN INDEXER

Interior Designer: David Futato

Cover Designer: Karen Montgomery

Illustrator: Kate Dullea

November 2023: First Edition

Revision History for the Early Release

2022-11-10: First Release

2022-11-21: Second Release

2023-02-23: Third Release

2023-04-19: Fourth Release

See http://oreilly.com/catalog/errata.csp?isbn=9781098139728 for release details.

The O’Reilly logo is a registered trademark of O’Reilly Media, Inc. Delta Lake: Up and Running, the cover image, and related trade dress are trademarks of O’Reilly Media, Inc.

The views expressed in this work are those of the authors and do not represent the publisher’s views. While the publisher and the authors have used good faith efforts to ensure that the information and instructions contained in this work are accurate, the publisher and the authors disclaim all responsibility for errors or omissions, including without limitation responsibility for damages resulting from the use of or reliance on this work. Use of the information and instructions contained in this work is at your own risk. If any code samples or other technology this work contains or describes is subject to open source licenses or the intellectual property rights of others, it is your responsibility to ensure that your use thereof complies with such licenses and/or rights.

This work is part of a collaboration between O’Reilly and Databricks. See our statement of editorial independence.

978-1-098-13966-7

[FILL IN]

Chapter 1. Introduction

A Note for Early Release Readers

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 1st chapter of the final book. The GitHub repo is https://github.com/benniehaelen/delta-lake-up-and-running.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at gobrien@oreilly.com.

As data engineers, we want to build large-scale data, machine learning, data science, and AI solutions that offer state-of-the-art performance. We build these solutions by ingesting large amounts of source data. We cleanse, normalize and combine the data, and ultimately present it to the downstream applications through an easy to consume data model.

As the amount of data that we need to ingest and process is ever-increasing, we need the ability to horizontally scale our storage. Additionally, we need the ability to dynamically scale our compute resources to address processing and consumption spikes. Since we are combining our data sources together into one data model, we not only need to append data to tables, but we often need to insert, update, or delete (i.e., MERGE or UPSERT) records, based upon complex business logic. We want to be able to perform these operations with transactional guarantees, and without having to re-write large data files.

In the past, the above set of requirements had been addressed by two distinct toolsets. The horizontal scalability and de-coupling of storage and compute was offered by cloud-based data lakes, while transactional guarantees were offered by relational data warehouses. However, traditional data warehouses coupled storage and compute into an on-premises appliance, and do not have the degree of horizontal scalability associated with data lakes.

Delta Lake brings capabilities such as transactional reliability and support for UPSERTs and MERGEs to data lakes while maintaining the dynamic horizontal scalability and separation of storage and compute of data lakes. Delta Lake is one the enablers for building Data lakehouses, an open data architecture that combines the best of data warehouses and data lakes.

The goal of this book is to provide experienced data practitioners with practical instructions on how to set up Delta Lake and start using its unique features. First, we’ll discuss why Delta Lake is an important tool for building modern enterprise data platforms and data science and AI solutions, followed by instructions on how to set up Delta Lake with Spark. Each of the subsequent chapters will walk you through the fundamental functions and operations of Delta Lake using step-by-step instructions and real-world examples.

In this introduction, we will take a brief look at relational databases and how they evolved into data warehouses. Next, we will look at the key drivers behind the emergence of data lakes. We will address the benefits and drawbacks of each architecture, and finally show how the Delta Lake storage layer combines the benefits of each architecture, enabling the creation of data lakehouse solutions.

Introduction of Relational Databases

In his historic 1970 paper¹ EF Codd introduced the concept of looking at data as logical relations, independent of the physical data storage. This logical relation between data entities became known as a database model or schema. Codd’s writings led to the birth of the relational database. The first relational database systems were introduced in the mid-1970s by IBM and UBC.

Relational databases and their underlying SQL language became the standard storage technology for enterprise applications throughout the 1980s and 1990s. One of the main reasons behind this popularity was that relational databases offered a concept called transactions. A database transaction is a sequence of operations on a database that satisfies the ACID properties, ACID is an acronym which stands for four properties: atomicity, consistency, isolation, and durability.

Atomicity ensures that all changes made to the database are executed as a single operation. This means that the transaction succeeds only when all changes have been performed successfully. For example, when I use my online banking system to transfer money from savings to checking, the consistency property will guarantee that the operation will only succeed when the money is deducted from my savings account and added to my checking account. The complete operation will either succeed or fail as a complete unit.

The Consistency property guarantees that database transitions from one consistent state at the beginning of the transaction to another consistent state at the end of the transaction. In our earlier example, the transfer of the money would only happen if my savings account had sufficient funds. If not, the transaction would fail, and the balances would stay in their original, consistent state.

Isolation ensures that concurrent operations that are happening within the database are not affecting each other. This property ensures that when multiple transactions are executed concurrently, their operations do not interfere with each other.

Durability refers to the persistence of committed transactions. It guarantees that once a transaction is completed successfully, it will result in a permanent state even in the event of a system failure. In our money transfer example, durability will ensure that updates made to both my savings and checking account are persistent and can survive a potential system failure.

Database systems continued to mature throughout the 1990s, and the advent of the Internet in the mid-1990s led to an explosive growth of data and the need to store this data.

Enterprise applications were using the RDBMS technology very effectively. Flagship products such as SAP and Salesforce would collect and maintain massive amounts of data.

However, this development was not without its drawbacks. Enterprise applications would store the data in their own, proprietary formats, leading to the rise of data silos. These data silos were owned and controlled by one department or business unit. Over time, organizations recognized the need to develop an Enterprise view across these different data silos, leading to the rise of data warehouses.

Data Warehouses

While each enterprise application has some type of reporting built-in, business opportunities were missed because of the lack of a comprehensive view across the organization. At the same time, organizations recognized the value of analyzing data over longer periods of time. Additionally, they wanted to be able to slice and dice the data over several cross-cutting subject matters such as customers, products, and other business entities.

This led to the introduction of the data warehouse, a central relational repository of integrated, historical data from multiple data sources that presents a single integrated, historical view of the business with a unified schema, covering all perspectives of the enterprise.

Data Warehouse Architecture

A simple representation of a typical data warehouse architecture is shown in Figure 1-1.

Figure 1-1. Data Warehouse Architecture

When we look at the diagram in figure x.x, we start with the data source layer on the left. Organizations need to ingest data from a set of heterogeneous data sources. While the data from the organization’s ERP system(s) forms the backbone of the organizational model, we need to augment this data with the data from the operational systems that are running the day-to-day operations such as human relationship (HR) systems and workflow management software. Additionally, organizations might want to leverage the customer interaction data covered by their CRM- and POS systems. In addition to the core data sources listed above, there is a need to ingest data from a wide array of external data sources, in a variety of formats such as spreadsheets, CSV files etc.

These different source systems each have their own data format. Therefore, the data warehouse contains a staging area where the data from the different sources can be brought together into one common format. To do this the system must ingest the data from the original data sources. The actual ingestion process varies by data source types. Some systems allow direct database access, others allow data to be ingested through an API, while many data sources still rely on file extracts.

Next, the data warehouse needs to transform the data into a standardized format, allowing the downstream processes to easily access the data. Finally, the transformed data is loaded into the staging area. In relational data warehouses, this staging area is typically a set of flat relational staging tables without any primary- of foreign keys, and simple data types.

This process of extracting data, transforming it to a standard format and loading it into the data warehouse is commonly referred to as Extract, Transform and Load (ETL). ETL tools can perform several other tasks on the ingested data before finally loading the data into the data warehouse. These tasks include elimination of duplicate records. Since a data warehouse will be the one source of truth, we do not want it to contain multiple copies of the same data. Additionally, duplicate records also prevent the generation of a unique primary key for each record.

ETL tools also allow us to combine data from multiple data sources. For example, one view of our customers might be captured in CRM systems while other attributes are found in an ERP system. The organization needs to combine these different aspects into one comprehensive view of a customer. This is where we start to introduce a schema to the data warehouse. In our example of a customer, the schema will define the different columns for the customer table, which columns are required, the data type and constraints of each column etc.

Having canonical, standardized representations of columns such as date and time is important, ETL tools can ensure that all these types of columns are formatted using the same standard throughout the data warehouse.

Finally, organizations want to perform quality checks on the data in keeping with their data governance standards. This might include dropping low quality data rows which do not meet this minimal standard.

Data warehouses are physically implemented on a monolithic physical architecture, made up out of a single large node, combining memory, compute and storage. This monolithic architecture forces organizations to scale their infrastructure vertically, resulting in expensive, often over-dimensioned infrastructure, which was provisioned for peak user load, while being near idle at other times

A data warehouse typically contains data that can be classified as follows:

Metadata

Contextual information about the data. This data is often stored in a data catalog. It enables the data analysts to describe, classify and easily locate the data stored in the data warehouse.

Summary data

Automatically created by the underlying data management system. The summary data will automatically be updated as new data is loaded into the warehouse. Summary data contains aggregations across several conformed dimensions. The main purpose of the summary data is to accelerate query performance.

Raw data

Maintained in its original format without any processing. Having access to the raw data enables the data warehouse system to re-process data in the case of load failures.

The data in the warehouse is consumed in the presentation layer. This is where the consumers can interact with the data stored in the warehouse. We can broadly identify two large groups of consumers:

Human consumers

These are the people within the organization who have a need to consume the data in the warehouse. These consumers can vary from knowledge workers, who need access to the data as an essential part of their job, to executives who typically consume highly summarized data, often in the form of dashboards and KPI’s.

Internal or external Systems

The data in a data warehouse can be consumed by a variety of internal or external systems. This can include machine learning and AI toolsets, or internal applications which need to consume data in the warehouse. Some systems might directly access the data, others might work with data extracts, while still others might directly consume the data in a pub-sub model.

Human Consumers will leverage various analytical tools and technologies to create actionable insights into the data, including:

Reporting tools

These tools enable the user to develop insights into the data through visualizations such as tabular reports and a wide array of graphical representations.

Online Analytical Processing (OLAP) tools

Consumers need to slice and dice the data in a variety of ways. Online Analytical Processing (OLAP) tools present the data in a multidimensional format, allowing it to be queried from multiple perspectives. They leverage pre-stored aggregations, often stored in memory, to serve up the data with fast performance.

Data Mining

These tools allow a data analyst to find patterns in the data through mathematical correlations and classifications. They assist the analysts in recognizing previously hidden relationships between different data sources. In a way, data mining tools can be seen as a precursor to modern data science tools.

Dimensional Modeling

Data warehouses introduced the need for a comprehensive data model that spans the different subject areas in a corporate enterprise. The technique used to create these models became known as dimensional modeling.

Driven by the writings and ideas of visionaries such as Bill Inmon and Ralph Kimball, dimensional modeling was first introduced in Kimball’s seminal book The Data Warehouse toolkit: The Complete Guide to Dimensional Modeling.² Kimball defines a methodology that focuses on a bottoms-up approach, ensuring that the team delivers real value with the data warehouse as soon as possible.

A Dimensional model is described by a star schema. A star schema is a way of organizing the data for a given business process (e.g., sales) into a structure which facilitates easy analytics. It consists of two types of tables:

A fact table, which is the primary, or central table for the schema. The fact table captures the primary measurements, metrics, or “facts” of the business process Staying with our sales business process example, a Sales Fact Table would include:

Units sold

Sales amount

Fact tables have a well-defined grain. The grain of a fact table would be the level of granularity at which we store the data in our fact table. For our Sales Fact Table, the grain would be at the sales detail level.

Multiple dimension tables which are related to the fact table. A Dimension provides the context surrounding the selected business process. In a Sales scenario example, the list of dimensions could include:

Product

Customer

Salesperson

Store

The dimension tables “surround” the fact table, which is why these types of schemas are referred to as “star schemas”.

A star schema consists of fact tables, linked to their associated dimensional tables through primary and foreign key relationships. A star schema for our sales subject area is shown in Figure 1-2.

Figure 1-2. Sales Dimensional Model

Data Warehouse Benefits & Challenges

Data warehouses have inherent strengths that have served the business community well.

They serve up high quality, cleansed and normalized data from different data sources in a common format. Since data from the different departments is presented in a common format, each department will review results in line with the other departments. Having timely, accurate data is the basis for strong business decisions.

Since they store large amounts of historical data, they enable historical insights, allowing the user to analyze different time periods and trends.

Data warehouses tend to be very reliable, based upon the underlying relational database technology, which executes ACID transactions.

Warehouses are modeled with standard star-schema modeling techniques, creating fact tables and dimensions. More and more pre-built template models became available for various subject areas such as sales and CRM, further accelerating the development of such models.

Data warehouses are ideally suited for business intelligence and reporting, basically addressing the “what happened” question of the data maturity curve. Data warehouse, combined with BI tools can generate actionable insights for marketing, finance, operations, and sales.

The fast rise of the internet, social media and the availability of multi-media devices such as smart phones disrupted the traditional data landscape, giving rise to the term big data. Big data is defined as data that arrives in ever higher volumes, with more velocity, and a greater variety of formats and has higher veracity. These are known as the four Vs of data:

Volume. The volume of data created, captured, copied, and consumed globally is increasing rapidly. As described in Statistica³, over the next five years, global data creation is projected to grow to more than 200 zettabytes (a zettabyte is a 2 to the power 70 number of bytes).

Velocity. In today’s modern business climate, timely decisions are critical. To make these decisions, organizations need their information to flow quickly, ideally as close to real-time as possible. For example, stock trading applications need to have access to near-real-time data so advanced trading algorithms can make millisecond decisions, and communicate these decisions to their stakeholders. Access to timely data can give organizations a competitive advantage.

Variety. Variety refers to the number of different “types” of data that are now available. The traditional data types were all structured, and typically offered as relational databases, or extracts thereof. With the rise of big data, data now arrives in new unstructured types. Unstructured and semi-structured data types, such as IoT device messages, text, audio, and video require additional preprocessing to derive business meaning.

Veracity. Veracity defines the trustworthiness of the data. Here, we want to make sure that the data is accurate and of high quality. Data can be ingested from several sources. It is important to understand the chain of custody of the data, ensure we have rich metadata, and understand the context under which the data was collected. Additionally, we want to ensure that our view of the data is complete, with no missing components or late-arriving facts.

Data Warehouses have a hard time addressing these four Vs.

Traditional data warehouse architectures struggle to facilitate exponentially increasing data volumes. They suffer from both storage and scalability issues. With volumes reaching petabytes, it becomes challenging to scale storage capabilities without spending large amounts of money. Traditional data warehouse architectures do not use in-memory and parallel processing techniques, preventing them from vertically scaling the data warehouse.

Data Warehouse architectures are also not a good fit to address the velocity of big data. Data warehouses do not support the types of streaming architecture required to support near real-time data. ETL Data load windows can only be shortened so much until the infrastructure starts to buckle.

While data warehouses are very good at storing structured data, they are not well suited to store and query the variety of semi-structured or unstructured data.

Data warehouses have no built-in support for tracking the trustworthiness of the data. Data warehouse metadata is mainly focused on schema, and less on lineage, data quality and other veracity variables.

Further, data warehouses are based upon a closed, proprietary format and typically only support SQL-based query tools. Because of their proprietary format, data warehouses do not offer good support for data science and machine learning tools.

Because of these limitations, data warehouses are expensive to build, as a result projects often fail before going live, and those that do go live have a hard time keeping up with the ever-changing requirements of the modern business climate and the four Vs.

The limitations of the traditional data warehouse architecture gave rise to a more modern architecture, based upon the concept of a data lake.

Introducing Data Lakes

A Data Lake is a cost-effective central repository to store structured, semi-structured or unstructured data at any scale, in the form of files and blobs. The term “data lake” came from the analogy of a real river or lake, holding the water, or in this case data, with several tributaries which are flowing the water (aka “data”) into the lake in real-time. A canonical representation of a typical data lake is shown in Figure 1-3.

Figure 1-3. Canonical Data Lake

The initial data lakes and big data solutions were built with on-premises clusters, based upon the Apache Hadoop open-source set of frameworks. Hadoop was used to efficiently store and process large datasets ranging in size from gigabytes to petabytes of data. Instead of using one large computer to store and process the data, Hadoop leverages clustering of multiple commodity compute nodes to analyze large volumes of datasets in parallel more quickly.

Hadoop would leverage the MapReduce framework to parallelize compute tasks over multiple compute nodes. The Hadoop Distributed File System (HDFS) was a file system that was designed to run on standard or low-end hardware. HDFS was very fault-tolerant and supported large datasets.

Starting in 2015, cloud data lakes such as S3, ADLS, and GCS started replacing HDFS. These cloud-based storage systems have superior Service Level Agreements (SLAs) (often greater than 10 nines), offer geo-replication and most importantly, offer extremely low cost with the option to utilize even lower-cost cold storage for archival purposes.

At the lowest level, the unit of storage in a data lake is a blob of data. Blobs are by nature unstructured enabling the storage of semi-structured and unstructured data, such as large audio and video files. At a higher level, the cloud storage systems provide file semantics and file-level security on top of the blob storage, enabling the storage of highly structured data. Because of their high bandwidth ingress and egress channels, data lakes also enable streaming use cases, such as the continuous ingestion of large volumes of IoT data or streaming media.

Compute engines enable large volumes of data to be processed in an ETL-like fashion and delivered to consumers such as traditional data warehouses and machine learning and AI toolsets. Streaming data can be stored in real-time databases, and reports can be created with traditional BI and reporting tools.

Data lakes are enabled through a variety of technologies, as listed below:

Storage

Data lakes require very large, scalable storage systems, like the ones typically offered in cloud environments. The storage needs to durable, scalable and should offer interoperability with a variety of third-party tools, libraries, and drivers. Note that data lakes separate out the concepts of storage and compute, allowing both to scale independently. Independent scaling of storage and compute allows for on-demand, elastic fine-tuning of resources, allowing our solution architectures to be more flexible. The ingress and egress channels to the storage systems should support high bandwidths, enabling the ingestion or consumption of large batch volumes, or the continuous flow of large volumes of streaming data, such as IoT and streaming media.

Compute

High amounts of compute power are required to process the large amounts of data stored in the storage layer. Several compute engines are available on the different cloud platforms. The go-to compute engine for data lakes is Apache Spark. Spark is an open-source unified analytics engine, which can be deployed through various environments, such as Databricks, Microsoft Synapse or Snowflake. Big Data compute engines will leverage compute clusters. Compute clusters pool compute nodes to tackle complete data collection and processing tasks.

Formats

The shape of the data on disk defines the formats. A wide array of storage formats is available. Most data lakes use standardized formats such as Parquet or Avro.

Metadata

Modern, cloud-based storage systems maintain metadata (data about the data). This includes various timestamps that describe when data was written or accessed, data schemas and a variety of tags, which contain information about the usage and owner of the data.

Benefits and Drawbacks of Data Lakes

A data lake architecture enables the consolidation of an organization’s data assets into one central location.

Data lakes have some very strong benefits. They use open formats, such as Parquet and Avro. These formats are well-understood by a variety of tools, drivers, and libraries which enables easy interoperability, especially in our open-source ecosystem.

Data lakes are deployed on mature cloud storage sub-systems, allowing them to benefit from the scalability, monitoring ease of deployment, and the low storage costs associated with these systems. Automated DevOps tools such as Terraform have well-established drivers, enabling automated deployments and maintenance.

Unlike data warehouses, data lakes support all data types, including semi-structured and unstructured data, enabling workloads such as media processing.

Because of their high throughput ingress channels, they are very well suited for streaming use cases, such as the ingestion of IoT sensor data, media streaming, or Web clickstreams.

However, as data lakes become more popular and widely used, organizations started to recognize some challenges with traditional data lakes. While the underlying cloud storage is relatively inexpensive, building and maintaining an effective data lake requires expert skills, resulting in high-end staffing or increased consulting services costs.

While it is easy to ingest data into the data in its raw form, transforming the data into a form that can deliver business values can be very expensive. Traditional data lakes have low latency query performance, so they cannot be used for interactive queries. As a result, the organization’s data teams must still transform and load the data into something like a data warehouse, resulting in an extended time to value. This resulted in a data lake + warehouse architecture. This architecture continues to be dominant in the industry f(I have personally implemented dozens of those types of these systems), but is now declining because of the rise of data lakehouses.

Data lakes have typically used a “schema on read” strategy, where data can be ingested in any format without schema enforcement. Only when the data is read, can some type of schema be applied. This lack of schema enforcement can result in data quality issues, allowing the pristine data lake to become a “data swamp”.

Data lakes do not offer any kind of transactional guarantees. Data files can only be appended to, leading to expensive re-writes of previously written data to make a simple update. This leads to an issue referred to as the “small file problem”, where multiple small files are created for a single entity. If this issue is not managed well, these small files slow the read performance of the overall data lake, lead to stale data and wasted storage. Data lake administrators need to run repeated operations to consolidate these smaller files into larger files which are optimized for efficient read operations.

Now that we have discussed the strengths and weaknesses of both data warehouses and data lakes, we will introduce the data lakehouse, which combines the strengths, and addresses the weaknesses of both technologies.

Data Lakehouse

The concept of the data lakehouse was first introduced by Armbrust, Ghodsi, Xin and Zaharia.⁴ The authors define a lakehouse as “a data management system based upon low-cost and directly-accessible storage that provides analytics DBMS management and performance features such as ACID transactions, data versioning, auditing, indexing, caching and query optimization”.

When we unpack the above, we can define a lakehouse as a system which merges the flexibility, low cost, and scale of a data lake with the data management and ACID transactions of data warehouses, addressing the limitations of both. Like data lakes, the lakehouse architecture leverages the low-cost cloud storage systems, with the inherent flexibility and horizontal scalability that comes with those systems. It also continues to leverage the data formats like Parquet and AVRO, which enable the integration with machine learning and AI systems. Like data warehouses, a data lakehouse will enable ACID transactions on data assets stored on these low-cost storage systems, enabling the reliability that used to be the exclusive domain of relational database systems.

Data lakehouses allows the implementation of similar data structures and data management features to those in a data warehouse, directly on the same storage resources used for traditional data lakes. data lakehouses are a good bit for BOTH business intelligence AND machine learning/AI use cases.

Data lakehouses are an especially good match for most, if not all, cloud environments with separate compute and storage resources. Different computing applications can run on-demand on completely separate computing nodes, such as a Spark cluster while directly accessing the same storage data. It is, however, conceivable that one could implement a data lakehouse over an on-premises storage system such as HDFS.

Data Lakehouse benefits

An overview of the data lakehouse architecture is shown in Figure 1-4.

Figure 1-4. Data Lakehouse Architecture Overview

With the data lakehouse architecture, we no longer need to have a copy of our data in the data lake, and another copy in some type of data warehouse storage. Indeed, we can serve up our data directly from the Data Lake with comparable performance to a classical data warehouse.

Since we can continue to leverage the low-cost cloud-based storage technologies, and we no longer need to copy data from the data lake to a data warehouse, we can realize significant cost savings, both in infrastructure and staff and consulting overhead.

Since less data movement takes place and our ETL is simplified, opportunities for data quality issues are significantly reduced, and finally, because the data lakehouse combines the ability to store large data volumes and refined dimensional models, development cycles are reduced, and the time-to-value is significantly reduced.

The evolution from data warehouses to data lakes to a data lakehouse architecture is shown in Figure 1-5.

Figure 1-5. Evolution of Data Architectures

Implementing a Data Lakehouse

As we mentioned earlier, data lakehouses will leverage low-cost object stores like Amazon S3 or Azure ADLS, storing the data in an open-source format such as Apache Parquet. However, since data lakehouse implementations run ACID transactions against this data, we need a transactional metadata layer on top of the cloud storage, which defines which objects are part of which table version.

This will allow a data lakehouse to implement features such as ACID transactions and versioning within that metadata layer, while keeping the bulk of the data in the low-cost object storage. The data lakehouse client is able to keep using data in an open-source file format that they are already familiar with.

Next, we need to address system performance. As we mentioned earlier, to be effective, data lakehouse implementations need to achieve great SQL performance. Data warehouses were very good at optimizing performance because they worked with a closed storage format. This allowed them to maintain statistics, build clustered indexes, move hot data on fast SSD devices etc.

In a data lakehouse, which is based upon open-source standard file formats, we do not have that luxury, since we are unable to change the storage format. However, data lakehouses can leverage a plethora of other optimizations, which leave the data files unchanged. This included caching, auxiliary data structures such as indexes and statistics and data layout optimizations.

The final tool that can speed up analytic workloads is the development of a standard DataFrame API. Most of the popular ML tools out there, such as TensorFlow and Spark MLlib already have support for DataFrames. DataFrames were first introduced by R and pandas and provide a simple table abstraction of the data with a multitude of transformation operations, most of which originate from relational algebra.

In Spark, the Dataframe API is declarative, and lazy evaluation is used to build an execution DAG (Directed Acyclic Graph). This graph can then be optimized before any action consumes that underlying data in the Dataframe. This gives the data lakehouse several new optimization features such as caching and auxiliary data.

Figure X.X shows how these requirements fit into an overall data lakehouse system design:

Figure 1-6. Data lakehouse implementation

A number of data lakehouse implementations are available such as Apache Iceberg and Delta Lake. Since Delta Lake is the focus of this book, we will illustrate how Delta lake facilitates the requirements for implementing a data lakehouse.

Delta Lake

As was mentioned in the previous section, a possible data lakehouse solution can be built on top of Delta Lake. Delta Lake is the metadata, caching and indexing layer on top of a data lake storage that provides an abstraction level to serve ACID transactions and other management features.

Delta Lake is a file-based open-source metadata layer that enables data lakehouse implementations. Delta Lake provides ACID transactions, scalable meta data handling, a unified process model that spans batch and streaming, full audit history, and support for SQL DML statements. It can run on existing data lakes and is fully compatible with several processing engines, including Apache Spark.

Delta Lake is an open-source framework, the specification of which can be found at https://delta.io. A detailed description of how Delta Lake provides ACID transaction can be found in the work of Armbrust et al⁵

Delta lake provides the following features:

Transactional ACID guarantees

Delta Lake will make sure that all data lake transactions using Spark, or any other processing engine are committed for durability and exposed to other readers in an atomic fashion. This is made possible through the Delta transaction log. In the next chapter we will cover the transaction log in detail.

Full DML support

Traditional data lakes do not support transactional, atomic updates to the data. Delta Lake fully supports all DML operations, including deletes and updates, but also complex data merge, or upsert scenarios. This support greatly simplifies the creation of dimensions and fact tables when building a modern data warehouse (MDW).

Audit History

The Delta Lake transaction log records every change made to the data, in the order that these changes were made. Therefore, the transaction log becomes the full audit trail of any changes made to the data. This enables admins and developers to roll back to earlier versions of data after accidental deletions and edits. This feature is referred to as Time Travel and is covered in detail in chapter X: Time Travel.

Unification of batch and streaming into one processing model

Delta Lake can work with batch and streaming sinks or sources. It can perform MERGEs on a data stream, which is a common requirement when merging IoT data with device reference data. It also enables use cases where we receive CDC data from external data sources. We will cover streaming in detail in chapter xx: Streaming.

Schema enforcement and evolution

Delta lake will enforce a schema when writing or reading data from the lake. However, when explicitly enabled for a data entity, it allows for a safe evolution of the schema, enabling use cases where the data needs to evolve. Schema enforcement and evolution is covered in chapter XX.

Rich metadata support and scaling

Having the ability to support large volumes of data is great, but if the metadata cannot scale accordingly, the solution will fall short. Delta Lake will scale out all metadata processing operations by leverage Spark or other compute engines, allowing it to efficiently process the metadata for petabytes of data.

Data Lakehouse-based Architectures

A data lakehouse architecture is made up out of three layers as shown in Figure 1-6. The data lakehouse storage layer is built on standard cloud-storage technology such as Azure ADLS Gen 2 storage, or AWS s3 storage. This provides the data lakehouse with a highly scalable, low-cost storage layer.

Figure 1-7. Data lakehouse Layered Architecture

The transactional layer of the data lakehouse is provided by Delta Lake. This brings ACID guarantees to the data lakehouse, enabling an efficient transformation of raw data into curated, actionable data. Besides the ACID support, Delta lake offers a rich set of additional features, such as DML support, scalable metadata processing, streaming support and a rich audit log. The top layer of the data lakehouse stack is made up out of high-performance query and processing engines, which leverage to underlaying cloud compute resources. Supported query engines include

Apache Spark

Apache Hive

Presto

Trino

Please consult the Delta Lake website (https://delta.io) for a complete list of supported compute engines.

The Medallion Architecture

An example of a Delta Lake based data lakehouse solution architecture is provided Figure 1-7. This architectural pattern with bronze, silver and gold layers is often referred to as the Medallion Architecture. While it is only one of many data lakehouse architecture patterns, it is a great fit for Modern Data Warehouses, Data Marts, and other analytical solutions.

Figure 1-8. Data Lakehouse Solution Architecture

At the highest level, we have three components in this solution. To the left we have the different data sources. A data source can take on many forms, some examples are provided below:

A set of CSV or TXT files on an external data lake.

An on-premises database, such as Oracle or SQL Server.

Streaming data sources such as Kafka or Azure Event Hubs.

REST APIs from a SAS service such as Salesforce or ADP.

The central component implements what is called the “medallion architecture”. A medallion architecture is a data design pattern used to logically organize data in a data lakehouse, through a bronze, silver, and gold layer. The bronze layer is where we land the data ingested from our source systems on the left. Data in the bronze zone is typically landed “as-is”, but can be augmented with additional metadata such as the loading date and time, processing identifiers etc.

In the silver layer, the data from the bronze layer is cleansed, normalized, merged and conformed. This is where the Enterprise view of the data across the different subject areas is gradually coming together.

The data in the gold layer is “consumption-ready” data. This data can be in the format of a classic star schema, containing dimensions and fact tables, or it could be in any data model that is befitting to the consuming use case.

The goal of the medallion architecture is to improve the structure and quality of the data incrementally and progressively as it flows through each layer of the architecture.

On the right we have the different consumers of the data lakehouse. This includes reporting and BI consumers, machine learning tools and AI solutions.

A summary of the Bronze, Silver and Gold layers is provided below. For each layer, we explore its business value and properties, and include implementation details (such as “how it’s done):

Bronze
Silver
Gold

Business Value

Audit on exactly what was received from the source

Ability to reprocess without going back to the sources

First layer that is useful to the business

Enables data discover, self-service, ad-hoc reporting, advanced analytics and ML

The data is in a format that is easy for the business users to navigate.

Highly performant

Properties

No business rules or transformations of any kind

Should be fast and easy to get new data to this layer

Prioritize speed to market and write performance- just enough transformations

Quality data expected

Prioritize business use cases and user experience

Precalculated, business-specific transformations

Can have separate views of the data for different consumption use cases

How it’s done

Must include a copy of what was received

Typically, data is stored in folders based upon the date received

Delta Merge

Can include light modeling (3nf, vaulting)

Data quality checks should be included

Prioritize denormalized, read-optimized data models

Full Transformed

Aggregated

In the next section, we will take a more detailed look at the different layers that make up the Medallion architecture.

The Bronze Layer (raw data)

Raw data from the data sources is ingested into the Bronze layer without any transformations or business rule enforcement. This layer is the “landing zone” for our raw data, so all table structures in this layer correspond exactly to the source system structure. The format of the data source is maintained, so when the data source is a CSV file, it is stored in Bronze as a CSV file, JSON data is written as JSON etc. Data extracted from a database table is typically landed in Bronze as a Parquet or AVRO file.

At this point, no schema is required. As data is ingested, a detailed audit record is maintained, which includes the data source, whether a full or incremental load was performed, and detailed watermarks to support the incremental loads where needed. The Bronze layer includes an archival mechanism, so that data can be retained for long periods of time. This archive, together with the detailed audit records can be used to re-process data in case of a failure somewhere downstream in the medallion architecture.

The ingested data is landed in the Bronze zone “as-is”, maintaining the structure and data types of the source system format, although it is often augmented with additional metadata, such as the date and time of the load and ETL process system identifiers. The goal of the ingestion process is to land the source data quickly and easily in the bronze layer with just enough auditing and metadata to enable data lineage and re-processing.

The bronze layer is often used as a source for a Change Data Capture (CDC) process, allowing newly arriving data to be immediately processed downstream through the silver and gold layers.

The Silver Layer

In the Silver layer we first cleanse and normalize the data. We ensure that standard formats are used for constructs such as date and time, we enforce the company’s column naming standard, de-duplicate the data, and perform a series of additional data quality checks, dropping low-quality data rows when needed.

Next, related data is combined and merged together. The Delta Lake MERGE capabilities work very well for this purpose. For example, customer data from various sources (Sales, CRM, POS systems etc..) is combined into a single entity. Conformed data, which are those data entities that are re-used across different subject areas is identified and normalized across the views. In our previous example, the combined customer entity would be an example of such conformed data.

At this point, the combined “Enterprise View” of the data starts to emerge. Note that we apply a “just-enough” philosophy here, where we provide just enough detail with the least amount of effort possible, making sure that we maintain our agile approach to building our medallion architecture.

At this point, we start enforcing schema, and allow the schema to evolve downstream. The Silver layer is also the layer where we can apply GDPR and/or PII/PHI enforcement rules.

Because this is the first layer where data quality is enforced, and the Enterprise view is created, it serves as a useful data source for the business, especially for purposes such as self-service analytics and ad-hoc reporting. The silver layer proves to be a great data source for machine learning and AI use cases. Indeed, these types of algorithms work best with the “less polished” data in the silver layer instead of the consumption formats in the gold layer.

The Gold Layer

In the Gold layer, we create business level aggregates. This can be done through a standard Kimball Star Schema, an Inmon Data Mart, or any other modeling technique that fits the consumer use case. The final layer of data transformations and data quality rules are applied here, resulting in high-quality, reliable data that can serve as the “single source of truth” in the organization.

The Gold layer continuously delivers high quality, clean data to the downstream users and applications. The data model in the gold layer often includes many different perspectives, or views of the data, depending on the consumption use cases.

The Gold layer will implement several Delta Lake optimization techniques, such as partitioning, data skipping and z-ordering to ensure that we deliver quality data in a performant way.

The Delta Ecosystem

As we mentioned earlier, Delta Lake enables us to build data lakehouse architectures, which enables both data warehousing and machine learning/AI applications to be hosted directly on a data lake. Today, Delta Lake is the most widely used data lakehouse format used by over 7,000 organizations, processing exabytes of data per day.

However, data warehouses and machine learning applications are not the only application target of Delta Lake. Beyond its core transactional ACID supports which brings reliability to data lakes, Delta Lake enables us to seamlessly ingest and consume both streaming and batch data with one solution architecture.

Another important component of Delta Lake is Delta Sharing, which enables companies to share data sets with each other in a secure manner.

Delta Lake 2.0 now also supports standalone readers and writers, enabling any client (Python, Ruby or Rust) to write data directly to Delta Lake without requiring any big data engine such as Spark or Flink.

Data Lake ships with an extended set of open-source connectors, including Presto, Flink and Trino.

The Delta Lake storage layer is now used extensively on many different platforms, include Azure Data Lake Storage Gen 2, Amazon s3 and Google’s Cloud Storage. All components of Delta Lake 2.0 have been open sourced by Databricks.

The success of Delta Lake and data lakehouses has spawned a completely new ecosystem, build around the Delta technology. This ecosystem is made up of a variety of individual components including Delta Lake Storage, Delta Sharing, and Delta Connectors.

Delta lake storage

Delta lake storage is an open-source storage layer that runs on top of cloud-based data lakes. It adds transactional capabilities to data lake files and tables, thereby bringing data warehouse-like features to a standard data lake. Delta lake storage is the core component of the ecosystem because all other components depend on this layer.

Delta Sharing

Data sharing is a very common use case in the business world. For example, for preventative maintenance and diagnostic purposes, a mining company might want to securely share IoT information from their massive mining truck engines with the manufacturer. A thermostat manufacturer might want to securely share HVAC data with a public utility to optimize the power grid load on high-usage days. However, in the past, implementing a secure, reliable data sharing solution was very challenging, and required expensive, custom development.

Delta Sharing is an open-source protocol for securely sharing large datasets of Delta Lake data. It allows a user to securely share data stored in S3, ADLS or GCS. With Delta Sharing users can directly connect to the shared data, using their favorite toolsets like Spark, Rust, Power BI, without having to deploy any additional components. Notice that the data can be shared across cloud providers, without any custom development.

Delta sharing enables use cases such as:

Data stored in Azure ADLS can be processed by a Spark Engine on AWS.

Data stored in AWS S3 can be processed by Rust on GCP.

Please refer to Chapter X of this book for a detailed discussion of Delta Sharing.

Delta Connectors

The main goal of Delta Connectors was to bring Delta Lake to other big data engines outside of Apache Spark.

Delta connectors are open-source connectors that directly connect to Delta Lake. The framework includes Delta Standalone which is a Java native library which allows direct reading and writing the Delta Lake tables without requiring an Apache Spark cluster. Consuming Applications can use Delta Standalone to directly connect to Delta files written by their big data infrastructure. This eliminates the need for data duplication into another format before it can be consumed.

Other native libraries are available for:

Hive. The Hive Connector reads Delta tables directly from Apache Hive.

Flink. The Flink/Delta Connector reads and writes Delta tables from Apache Flink application. The connector includes a sink for writing to Delta tables from Apache Flink, and a source for reading Delta tables using Flink.

Sql-delta-import. This connector allows for importing data from a JDBC data source directly into a Delta table.

Power BI. The Power BI connector is a custom Power Query function which allows Power BI to read a Delta table from any file-based data source supported by Microsoft Power BI.

Delta connectors is a fast-growing ecosystem, with more connectors becoming available on a regular basis. The integrations page on the Delta Lake home page describes the currently available integrations.⁶

¹ Codd, E. F. (1970). Relational Database: A Practical Foundation for Productivity. San Jose: San Jose Research Laboratory.

² Kimball, R. (1996). The Data Warehouse toolkit: The Complete Guide to Dimensional Modeling. The Kimball Group.

³ https://www.statista.com/statistics/871513/worldwide-data-created/

⁴ https://www.cidrdb.org/cidr2021/papers/cidr2021_paper17.pdf

⁵ https://www.databricks.com/wp-content/uploads/2020/08/p975-armbrust.pdf

⁶ You can go to the Delta Connectors Github page if you are interested in directly contributing to the fast-growing Delta Connectors ecosystem.

Chapter 2. Getting Started with Delta Lake

A Note for Early Release Readers

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 2nd chapter of the final book. The GitHub repo is https://github.com/benniehaelen/delta-lake-up-and-running.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at gobrien@oreilly.com.

In the previous chapter we introduced Delta Lake and saw how it adds transactional guarantees, DML support, auditing, a unified streaming and batch model, schema enforcement and a scalable metadata model to traditional data lakes.

In this chapter, we will go hands-on with Delta Lake. We will first set up Delta Lake on a local machine with Spark installed. We will run Delta Lake samples in two interactive shells:

First, we will run the PySpark interactive shell with the Delta Lake packages. This will allow us to type in and run a simple two-line Python program that creates a Delta dataset.

Next, we will run a similar program with the Spark Scala shell. Although we do not cover the Scala language is not extensively in this book, we want to demonstrate that both the Spark Shell and Scala are options with Delta Lake.

Next, we will create a helloDeltaLake starter program in Python inside our favorite editor and run the program interactively in the PySpark shell. The environment we set up in this chapter, and the helloDeltaLake program will be the basis for most other programs we create in this book.

Once our environment is up and running, we are ready to take a deeper look at the Delta format. Since Delta Lake uses Parquet as the underlying storage medium, we first take a brief look at the Parquet format. Since partitions and partition files play an important role when we study the transaction log later, we will study the mechanism of both automatic and manual partitioning.

Next, we move on to Delta dataset and investigate how a Delta dataset adds a transaction log in the _delta_log directory.

The remainder of this chapter is dedicated to the transaction log. We will create and run several Python programs to investigate the details of transaction log entries, what kind of actions are recorded, and what Parquet files are written when and how they relate to the transaction log entries. We will look at some more complex update examples and their impact on the transaction log. Finally, we will introduce the concept of checkpoint files and how they help Delta Lake to implement a scalable metadata system.

Getting a standard Spark Image

Setting up Spark on a local machine can be daunting. You have to adjust many different settings, update packages, and so on. Therefore, I chose to use a Docker container. If you do not have Docker installed, you can download it free from their website¹. The specific container that I used was the standard Apache Spark image². To download the image, you can use the following command:

docker pull apache/spark

Once you have pulled down the image, you can start the container with the following command:

docker run -it apache/spark /bin/sh

The spark installation is in the /opt/spark directory. You can find pyspark, spark-sql and all other tools in its /opt/spark/bin directory.

I have included several instructions on how to work with the container in the readme of the book’s GitHub ³ repository.

Using Delta Lake with PySpark

As we mentioned before, Delta Lake runs on top of your existing storage and is fully compatible with the existing Apache Spark APIs. This means that it is easy to get started with Delta Lake if you already have Spark installed or a container as specified in the previous section.

With Spark in place, you can install the delta-spark 2.2.0 package. You can find the delta-spark package in its pyspark directory.⁴

Enter the following command in a command shell:

pip install delta-spark

Once you have delta-spark installed, you can run the python shell interactively like this:

pyspark --packages io.delta:delta-core_2.12:2.1.0 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

This will give you a PySpark shell from which you can interactively run commands:

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.2.2
      /_/
Using Python version 3.9.13 (tags/v3.9.13:6de2ca5, May 17 2022 16:36:42)
Spark context Web UI available at http://host.docker.internal:4040
Spark context available as 'sc' (master = local[*], app id = local-1665944381326).
SparkSession available as 'spark'.

Inside the shell, you can now run interactive PySpark commands. I always do a quick test by creating a range() with Spark, resulting in a DataFrame that I can then save in Delta Lake format (more details in the second section of this chapter).

The full code is provided below:

data = spark.range(0, 10)
data.write.format("delta").mode("overwrite").save("/book/testShell")
Below is a full run:
>>> data = spark.range(0, 10)
>>> data.write.format("delta").mode("overwrite").save("/book/testShell")
>>>

Here we see the statement to create the range(), followed by the write statement. We see that the Spark Executors do indeed run. When we open up the output directory, we can indeed find the generated Delta dataset as shown in figure x.x (again, more details on the Delta format in the next section).

Figure 2-1. Delta Dataset generated from the Python Shell

Running Delta Lake in the Spark Scala shell

We can also run Delta Lake in the interactive Spark Scala shell. As specified in the Delta Lake Quickstart, we can start the Scala shell with the Delta Lake packages as follows:

spark-shell --packages io.delta:delta-core_2.12:2.1.0 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

This will start up the interactive Scala shell:

Spark context Web UI available at http://host.docker.internal:4040
Spark context available as 'sc' (master = local[*], app id = local-1665950762666).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.2.2
      /_/
Using Scala version 2.12.15 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_311)
Type in expressions to have them evaluated.
Type :help for more information.
scala>

Inside the shell, you can now run interactive Scala commands. Let’s do a similar test on scale as we did for the PySpark shell:

The full code is provided below:

val data = spark.range(0, 10)
data.write.format("delta").mode("overwrite").save("/book/testShell")
Below is a full run:
cala> val data = spark.range(0, 10)
data: org.apache.spark.sql.Dataset[Long] = [id: bigint]
scala> data.write.format("delta").mode("overwrite").save("/book/testShell")

And, again when we check our output, we can find the generated Delta dataset.

Running Delta Lake on Databricks

The preferred option for running Delta Lake is to leverage the Databricks Community Edition⁵. The Databricks Community Edition is the version of the Databricks cloud-hosted big data platform. Users can access a cluster with a complete notebook environment and an up-to-date runtime with Delta Lake installed on this platform.

If you do not want to run Spark and Delta Lake on your local machine, you also have the option of running Delta Lake on Databricks on a cloud platform, like Azure, AWS, or Google cloud. These environments make it easy to get started with Delta Lake, since their installed runtimes already have a version of Delta Lake installed.

The additional benefit of the cloud is that you can create real Spark clusters of arbitrary size, potentially up to thousands of cores spanning hundreds of nodes to process terabytes or petabytes of data.

When using Databricks in the cloud, you have two options. You can either use their popular notebooks or you can connect your favorite development environment to a cloud-based Databricks cluster with dbx⁶. Dbx by Databricks labs is an open source tool that allows you to connect to a Databricks cluster from an editing environment such as Visual Studio Code.

Creating and Running a Spark Program: helloDeltaLake

Once we have the delta-spark packages installed creating your first PySpark program is very straightforward.

Follow these steps to create the PySpark program:

Create a new file (I named mine helloDeltaLake.py)

Add the necessary imports. At a minimum you need to import PySpark and Delta Lake:

import pyspark
from delta import *

Next, create a SparkSession builder which loads up the Delta Lake extensions, as follows:

# Create a builder with the Delta extensions
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")

Next, we can create our SparkSession object itself. We will create the SparkSession object, and print out its version to ensure that the object is valid:

# Create a Spark instance with the builder
# As a result, we now can read and write Delta files
spark = configure_spark_with_delta_pip(builder).getOrCreate()
print(f"Hello, Spark version: {spark.version}")

To verify that our Delta Lake extensions are working correctly, we create a range and write it out in Delta Lake format:

# Create a range, and save it in Delta Lake format to ensure
# that our Delta Lake extensions are indeed working
df = spark.range(0, 10)
df.write                                 \
  .format("delta")                       \
  .mode("overwrite")                     \
  .save("/book/chapter02/helloDeltaLake")

That completes the code for our starter program. You can find the full code file in the /chapter02/helloDeltaLake.py location of the book’s code repository. This code is a good place to start if you want to write your own code.

Running helloDeltaLake

To run our program, we can simply start a command prompt on Windows, or a terminal on MacOS, and navigate to the folder with our code.

We simply start PySpark with our program as input, as shown below:

pyspark < helloDeltaLake.py

When we run the program, we get our Spark version output:

Hello, Spark version: 3.2.2

And when we look at our output, we can see that we have written a valid Delta dataset. The details of the Delta Lake format are covered in the next section.

At this point, we have PySpark and Delta Lake installed successfully, and we were able to code and run a full-fledged pySpark program with Delta Lake extensions.

I will be using the above approach to write and run most of the programs in this book. If I deviate from this standard approach, I will explicitly call it out in the book.

Now that we can run our own programs, we are ready to explore the Delta Lake format in detail in the next section.

The Delta Lake Format

In this section we will dive deeper into the Delta Lake format. When we save a file using this format, we are just writing a standard Parquet file with additional metadata.

Since Delta Lake writes out data as a Parquet file, we will take a more in-depth look at the Parquet format. We first write out a simple Parquet file and take a detailed look at the artifacts written by Spark. This will give us a good understanding of files we will leverage throughout this book.

Next, we will write out a file in Delta Lake format, noticing how it triggers the creation of the _delta_log directory, containing the transaction log. We will take a detailed look at this transaction log and how it is used to generate a single source of truth . We will see how the transaction log implements the ACID atomicity property mentioned in chapter 1.

We will see how Delta Lake breaks down a transaction into individual, atomic commit actions, and how it records these actions in the transaction log as ordered, atomic units.

Finally, we will look at several use cases and investigate what Parquet files and transaction log entries are written, and what is stored in these entries.

Since a transaction log entry is written for every transaction, we might end up with multiple small files. To ensure that this approach remains scalable, Delta Lake will create a checkpoint file every 10 transactions with the full transactional state. This way, a Delta Lake reader can simply process the checkpoint file, and the few transaction entries written afterward. This results in a fast, scalable metadata system.

Parquet Files

The Apache Parquet file format has been one of the most popular big data formats for the last 20 years. Unlike row-based formats such as CSV or Avro, Parquet is column-oriented, meaning that the values of each column/field are stored next to each other, rather than in each record. Second, Parquet is open source, so it is free to use under the Apache Hadoop license and is compatible with most Hadoop data processing frameworks. Parquet is self-describing. In addition to the actual data, it contains metadata, including the schema and structure of the file.

Parquet files perform compression on a column-by-column basis and are built to support flexible compression options and extendable encoding schemas for each data type. For example, a different encoding can be used for compressing integer versus string data.

Writing a Parquet File

In the book repository, the /chapter02/writeParquetFile Python program creates a Spark DataFrame in memory, and writes it in Parquet format to the /parquetData folder using the standard PySpark API:

data = spark.range(0, 100)
data.write.format("parquet")     \
           .mode("overwrite")    \
           .save('/book/chapter02/parquetData')

When we look at what is written to disk in this case, we see the following:

 Directory of C:\book\chapter02\parquetData
10/17/2022  09:05 AM    <DIR>          .
10/17/2022  09:05 AM    <DIR>          ..
10/17/2022 09:05 AM
10/17/2022  09:05 AM               511 part-00000-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00001-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               517 part-00002-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00003-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00004-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               517 part-00005-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00006-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00007-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               517 part-00008-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00009-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               513 part-00010-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet
10/17/2022  09:05 AM               517 part-00011-a3885270-f443-40dc-bdf9-8ee9946cbb79-c000.snappy.parquet

A developer new to the big data world might be a bit shocked at this point. We only did a single write of 100 numbers, so how did we end up with 12 parquet files? A bit of elaboration might be in order.

First, the file name you specify in the write is really the name of a directory, not really a file. As we can see the directory /parquetData is a directory containing 12 parquet files.

When we look at the .parquet files, we see that we have 12 files. Spark is a highly parallel computational environment, where the system is attempting to keep each CPU core in your Spark cluster busy. In my case, I am running on my local machine, which means there is one machine in my cluster. When I look at the information for my system, I see that I have 12 cores.

When we look at the number of .parquet files that were written, we see that we have 12 files, which is equal to the number of cores on my cluster. And indeed, that is Spark’s default behavior in this scenario. The number of files will be equal to the number of available cores. If we add the following statement to our code:

data = spark.range(0, 100)
data.write.format("parquet")     \
           .mode("overwrite")    \
           .save('/book/chapter02/parquetData')
print(f"The number of files is: {data.rdd.getNumPartitions()}")

We can see in our output that we indeed have 12 files:

The number of partitions is: 12
The number of files is: 12

While this might look like overkill for this scenario where we are only writing 100 numbers, one can imagine a scenario where we are reading or writing very large files. Having the ability to split the files that can be processed in parallel can dramatically increase performance.

The .crc files we see in the output are cyclic redundancy check files. They are used by Spark to ensure that data hasn’t been corrupted. These files are always very small, so their overhead is very minimal compared to the utility that they provide. While there is a way to turn off the generation of these files, I would not recommend doing so since their benefit far outweighs their overhead.

The final files in our output are the _SUCCESS and _SUCCESS.crc files. These files are used by Spark to provide a method to confirm that all partitions have been written correctly.

Partitioning

So far, we have looked at automatic partitioning, where Spark selected the number of partitions for us. We can also manually partition our data. When we have a large dataset, and we are always using the same query pattern, we can partition our data to align with the query pattern.

In this example, we have a dataset containing green taxi trip data from 2019. It contains all the information regarding each trip (fares, distance traveled, pickup and drop off locations etc.). Say we always want to access this data by pickup date. In this case we can partition by this column, as is shown in the following code-snippet (/chapter02/manualPartitioning.py in the book repo)

df.write.partitionBy("pickupDate")       \
        .format("parquet")               \
        .save("/book/chapter02/partitionedData")

When we look at the /partitionedData directory, we see that we have sub-folders by date:

 Directory of C:\book\chapter02\partitionedData
10/17/2022  09:13 AM    <DIR>          .
10/17/2022  09:13 AM    <DIR>          ..
10/17/2022  09:13 AM                 8 ._SUCCESS.crc
10/17/2022  09:13 AM    <DIR>          pickupDate=2008-12-31
10/17/2022  09:13 AM    <DIR>          pickupDate=2009-01-01
10/17/2022  09:13 AM    <DIR>          pickupDate=2010-09-23
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-11-30
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-01
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-02
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-03
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-04
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-05
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-06
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-07
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-08
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-09
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-10
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-11
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-12
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-13
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-14
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-15
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-16
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-17
10/17/2022  09:13 AM    <DIR>          pickupDate=2019-12-18

Each of our sub-directories now only contains a number of .parquet files:

 Directory of C:\book\chapter02\partitionedData\pickupDate=2009-01-01
10/17/2022  09:13 AM             5,836 part-00000-3b12ac09-fcd3-4010-957c-caf1eba7a84b.c000.snappy.parquet
10/17/2022  09:13 AM             5,638 part-00002-3b12ac09-fcd3-4010-957c-caf1eba7a84b.c000.snappy.parquet
10/17/2022  09:13 AM             5,662 part-00004-3b12ac09-fcd3-4010-957c-caf1eba7a84b.c000.snappy.parquet
10/17/2022  09:13 AM             5,289 part-00006-3b12ac09-fcd3-4010-957c-caf1eba7a84b.c000.snappy.parquet
               4 File(s)         22,425 bytes

Now, if we want to query the trip date for a given date, Spark can immediately navigate to the correct folder and read the trip data, effectively skipping 95% of the remaining data.

Writing a Delta Dataset

So far, we have been working with Parquet files only. Now, let’s take the first example from the previous section and save it in Delta Lake format instead of Parquet (code: /chapter02/writeDeltaFile.py) All we need to do is replace the parquet format with delta, as is shown in the code:

data = spark.range(0, 100)
data.write                \
    .format("delta")      \
     .mode("overwrite")   \
     .save('/book/chapter02/deltaData')
print(f"The number of filesis: {data.rdd.getNumPartitions()}")

We get the same number of partitions:

The number of files is: 12

And when we look at the output, we see the addition of the _delta_log file, as shown below:

 Directory of C:\book\chapter02\deltaData
10/17/2022  09:36 AM    <DIR>          .
10/17/2022  09:36 AM    <DIR>          ..
10/17/2022  09:36 AM                16 .part-00000-a707e0b9-b984-4642-976c-528a71a0b060-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00001-1f8ad5e6-836d-4ab7-a2f6-dd73d22fb37b-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00002-fd9f62af-024f-4097-9b1b-68e50858a871-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00003-23f8466e-a624-4987-b786-5c8c1dd7d756-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00004-d4927a28-fa66-447b-b9f9-067f951c9e08-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00005-fe003624-6ccf-49cc-a780-bb16a601b2fd-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00006-ee7f03c3-32a7-4c35-84d1-75d99ed658a8-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00007-32c9c8e1-e0a5-485e-baac-f9d7a323c046-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00008-0f06006a-dea6-46d9-ba1f-b5ba05c74ff7-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00009-9f1fbe69-d9ba-4adc-a846-f2b5687c504e-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00010-19b88404-209e-422b-94aa-42e10c17bef0-c000.snappy.parquet.crc
10/17/2022  09:36 AM                16 .part-00011-0a84e9db-50bd-4eff-8b47-06be16983ad4-c000.snappy.parquet.crc
10/17/2022  09:36 AM               524 part-00000-a707e0b9-b984-4642-976c-528a71a0b060-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00001-1f8ad5e6-836d-4ab7-a2f6-dd73d22fb37b-c000.snappy.parquet
10/17/2022  09:36 AM               523 part-00002-fd9f62af-024f-4097-9b1b-68e50858a871-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00003-23f8466e-a624-4987-b786-5c8c1dd7d756-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00004-d4927a28-fa66-447b-b9f9-067f951c9e08-c000.snappy.parquet
10/17/2022  09:36 AM               522 part-00005-fe003624-6ccf-49cc-a780-bb16a601b2fd-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00006-ee7f03c3-32a7-4c35-84d1-75d99ed658a8-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00007-32c9c8e1-e0a5-485e-baac-f9d7a323c046-c000.snappy.parquet
10/17/2022  09:36 AM               523 part-00008-0f06006a-dea6-46d9-ba1f-b5ba05c74ff7-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00009-9f1fbe69-d9ba-4adc-a846-f2b5687c504e-c000.snappy.parquet
10/17/2022  09:36 AM               519 part-00010-19b88404-209e-422b-94aa-42e10c17bef0-c000.snappy.parquet
10/17/2022  09:36 AM               523 part-00011-0a84e9db-50bd-4eff-8b47-06be16983ad4-c000.snappy.parquet
10/17/2022  09:36 AM    <DIR>          _delta_log
              24 File(s)          6,440 bytes

The _delta_log file contains the transaction log with every single operation performed on our data.

The Delta Lake Transaction Log

The Delta Lake transaction log (also known as DeltaLog), is a sequential record of every transaction performed on a Delta Lake table since its creation. It is central to the Delta Lake functionality because it is at the core of many of its important features, including ACID transactions, scalable metadata handling, and time travel.

The main goal of the transaction log is to enable multiple readers and writers to operate on a given version of a dataset file simultaneously and to provide additional information like data skipping indexes, to the execution engine for more performant operations. The Delta Lake transaction log always shows the user a consistent view of the data and serves as a single source of truth. It is the central repository that tracks all changes the user makes to a dataset users make to a Delta Lake Dataset.

When a Delta Lake reader reads a Delta Lake Dataset for the first time or runs a new query on an open file that has been modified since the last time it was read, Delta Lake looks at the transaction log to get the latestversion of the table This ensures that a user’s version of a file is always synchronized with the master record as of the most recent query and that users cannot make divergent, conflicting changes to a file.

How the Transaction Log Implements Atomicity

In chapter 1, we learned atomicity guarantees that all operations (albeit an INSERT, UPDATE, DELETE or MERGE) performed on your file will either succeed as a whole or not succeed at all. Without atomicity, any hardware failure or software bug can cause a data file to be written partially, resulting in corrupted or at a minimum invalid data.

The transaction log is the mechanism through which Delta Lake can offer the atomicity guarantee. The transaction log is also responsible for metadata, time travel and significantly faster metadata operations for large tabular datasets.

The transaction log is an ordered record of every transaction made against a Delta table since it was created. It acts as a single source of truth and tracks all changes made to the table. The transaction log allows users to reason about their data and trust its completeness and quality level.

The simple rule is if an operation is not recorded in the transaction log, it never happened.

In the next sections, we will illustrate these principles with several examples.

Breaking down Transactions into Atomic Commits

Whenever we perform a set of operations to modify a table or storage file (such as INSERTs, UPDATEs, DELETEs or MERGEs), Delta Lake will break down that operation into a series of atomic, discrete steps composed of one or more of the actions shown in table X-X.

Table 2-1. Table X.X List of possible actions in a transaction log entry

Action | Description
Add file | Adds a file
Remove file | Removes a file
Update Metadata | Updates the table’s metadata (e.g., changing the table or file’s name, schema, or partitioning). The first transaction log entry for a table or file will always contain an Update Metadata action with the schema, the partition columns and other information.
Set transaction | Records that a structured streaming job has committed a micro-batch with the given stream ID. For more information, see Chapter X: Streaming
Change Protocol | Enables new features by switching the Delta Lake transaction log to the newest software protocol.
Commit Info | Contains information about the commit, which operation was made, from where, and at what time. Every transaction log entry will contain a Commit Info action.

These actions are recorded in the transaction log entries (*.json) as ordered, atomic units known as commits. This is similar to how the git source control system tracks changes as atomic commits. This also implies that you can re-play each of the commits in the transaction log to get to the current state of the file.

For example, if a user creates a transaction to add a new column to a table and then adds data to it, Delta Lake would break this transaction down into its component action parts, and once the transaction completes, add them to the transaction log as the following commits:

Update metadata – change the schema to include the new column.

Add file – for each new file added

The Transaction Log at the File Level

When we write a Delta table, that file’s transaction log is automatically created in the _delta_log subdirectory. As we continue to make changes to the Delta table, these changes will be automatically recorded as ordered atomic commits in the transaction log. Each commit is written as a JSON file, starting with 0000000000000000000.json. If we make additional changes to the file, Delta Lake will generate additional JSON files in ascending numerical order, so the next commit is written as 0000000000000000001.json, the following one as 0000000000000000002.json and so on.

In the remainder of this chapter, we will use an abbreviated form for the transaction log entries for readability purposes. Instead of showing up to 19 digits, we will use an abbreviated form with up to 5 digits (so, we will use 00001.json instead of the longer notation).

Additionally, we will be shortening the name of the parquet files. These names typically look as follows:

part-00007-71c70d7f-c7a8-4a8c-8c29-57300cfd929b-c000.snappy.parquet
We will abbreviate a name like this to part-00007.parquet, leaving off the GUID and the snappy.parquet portion.

In our example visualizations, we will visualize each transaction entry with the action name and the file name affected, for example, in figure X.X, where we have a Remove (file) action, and another Add (file) action in a single transaction file.

Figure 2-2. Notation for a transaction log entry

Write Multiple Writes to the Same File

Throughout this section, we will use a set of figures that describe each code step in detail. For each step, we show the following information:

The actual code snippet is shown in the second column.

Next to the code snippet we show the parquet files written as a result of the code snippet execution.

In the last column we show the transaction log entry’s JSON files. We show the action and the affected parquet file name for each transaction log entry.

For this first example we will use chapter02/MultipleWriteOperations.py from the book’s repo to show multiple writes to the same file.

Figure 2-3. Multiple writes to the same file

A step-by-step description of the different steps in figure x.x is shown below:

First, a new Delta table is written to our path. One Parquet file was written to the output path (part-00000.parquet). The first transaction log entry (00000.json) has been created in the _delta_log directory. Since this is the first transaction log entry for the file, a metadata action is recorded, together with an add file action, indicating a single partition file was added.

Next we append data to the table. You can see a new parquet file (part-00001.parquet) has been written, and you created an additional entry (00001.json) in the transaction log. Like the first step, the entry contains an add file action, because you added a new file.

We append more data. Again, a new file is written (part-00002.parquet), and a new transaction log file (00002.json ) is added to the transaction log with an add file action.

Note that each transaction log entry will also have a commitInfo action, which contains the audit information for the transaction. We omitted the commitInfo log entries on the figures for readability purposes.

The sequence of operations for writes is very important. For each write operation, the file is always written first, and only when that operation succeeds a transaction log file is added to the _delta_log folder. The transaction is only considered complete when the transaction log entry is written successfully.

Reading the latest version of a Delta Table

When the system reads a Delta table, it will iterate through the transaction log to “compile” the current state of the table. The sequence of events when reading a file is as follows:

The transaction log files are read first.

The files are read based on the log files.

Next, we will describe that sequence for the Delta table written by the previous example (multipleWriteOperations.py). Delta will read all our log files (00000.json, 00001.json and 00002.json), then it will read the three files based upon the log information, as shown in figure x.x

Figure 2-4. Read Operations

Note that the sequence of operations also implies that there could be files that are no longer referenced in the transaction log. Indeed, this is a common occurrence in UPDATE or DELETE scenarios. Delta Lake does not delete these files since they might be required again if the user uses the Time Travel feature of Delta Lake (covered in Chapter X). You can remove old, obsolete files with the VACUUM command.

Failure scenario with a write operation

Next, let’s see what happens if a write operation fails. In the previous write scenario, let’s assume the write operation in step 3 fails halfway through. Part of the Parquet file might have been written, but the transaction log entry 00002.json was not. We would have a scenario as shown in figure X.X.

Figure 2-5. Failure during the last write operation.

As you can see in figure X.X, the last transaction file is missing. According to the read sequence specified earlier, Delta Lake will read the first and second JSON transactions file, and its corresponding part-00000 and part-00001 parquet files. The Delta Lake reader will not read inconsistent data; it will read a consistent view through the first two transaction log files.

Update Scenario

The last scenario is contained in the chapter02/UpdateOperation.py code repo. To keep things simple, we have a small Delta table with patient information. We are only tracking the patientId and the name of each patient. In this use case, we first create a Delta dataset with four patients, two in each file. Next, we add data with two more patients. Finally, we update the name of our first patient. As we will see, this update has a bigger impact than we first expect. The full update scenario is shown in figure x.x.

Figure 2-6. Updates and the transaction log

In this example, we execute the following steps:

The first code snippet creates a Spark DataFrame, with the patientId and name of four patients. We write the DataFrame to a Delta dataset forcing the data into two files with the .coalesce(2). As a result, we write two files. A transaction log entry is created (00000.json) once the part-00000.parquet and part-00001.parquet files are written. Notice that the transaction log entry contains two add file actions indicating the part-00000.parquet and the part-000001.parquet files were added.

The next code snippet appends the data for two more patients (P5 and P6). This results in the creation of the part-00002.parquet file. Again, once the file is written, the transaction log entry is written (00001.json), and the transaction is complete. Again, the transaction log file has one add file action, indicating that a file (part-2.parquet) was added.

The code performs an update. In this case, we want to update the patient’s name with patientId 1 from P1 to P11. Currently, the record for patientId 1 is present in part-0. To perform an update,, the part-0 is read and a map operator is used to update any record matching the patientId of 1 from P1 to P11. Finally, a new file is written as part-3. Finally, Delta Lake writes the transaction log entry (00002.json). Notice it writes a Remove File action, saying that the part-0 file is removed, and an Add File action, saying that the part-3 file has been added. This is because the data from part-0 was rewritten into part-3, and all modified rows (along with the unmodified rows) have been added to part-3, rendering the part-0 file obsolete.

Notice that the part-0 file is not deleted by Delta Lake, since a user might want to go back in time with Time Travel (see Chapter X), in which case the file is required. The VACUUM command can clean up unused files like this (see Chapter X: Time Travel).

Now that we have seen how the data is written during an update, let’s look at how a read would determine what to read, as is shown in figure X.X.

Figure 2-7. Reading after an update

The read would proceed as follows:

The first transaction log entry is read (00000.json). This entry tells Delta Lake to include the part-0 and part-1 files.

The next entry (00001.json) is read, which tells Delta Lake to include the part-2 file.

The last entry (00002.json) is read, which informs the reader to remove the part-0 file and include part-3.

As a result, the reader ends up reading part-1, part-2 and part-3, resulting in the correct data shown in Figure x.x.

Scaling Massive Metadata

Checkpoint files

Now that we have seen how the transaction log records each operation, it is conceivable we can have many very large files with thousands of transaction log entries for a single Parquet file. How does Delta Lake scale its metadata handling without needing to read thousands of small files, which would negatively impact Spark’s reading performance? Spark tends to be most effective when reading large files, so how do we resolve this?

Once the Delta Lake writer has made the commits to the transaction log, it will save a checkpoint file in Parquet format in the _delta_log folder. The Delta Lake writer will continue to generate a new checkpoint every 10 commits.

A checkpoint file saves the entire state of the table at a given point in time. Note that “state” refers to the different actions, not the file’s actual content. So, it will contain the add file, remove file, update metadata, commit info etc. actions, with all the context information. It will save this list in native Parquet format. This will allow Spark to read the checkpoint quickly. This gives the Spark reader a “shortcut” to fully reproduce a table’s state and avoid reprocessing thousands of small JSON files, which could be inefficient.

Checkpoint File Example

Following is an example where we do multiple commits, and a checkpoint file is generated as a result. This example uses the code file chap02/ TransactionLogCheckPointExample.py from the book’s repository.

Figure 2-8. Caption to come

This example has the following steps:

The first code snippet creates a standard Spark DataFrame with several patients. Note that we apply a coalesce(1) transaction to the Dataframe, which forces that data into one single partition. Next, we write the DataFrame in Delta Lake format to a storage file. We verify that a single part-0001.parquet file was written. We also see that a single transaction log entry (00000.json) has been created in the _delta_log directory. This directory entry contains an add file action for the part-00001.parquet file.

In the next step, we set up a loop over a range(0, 9), which will loop nine times, creating a new patient tuple, then creating a DataFrame from that tuple, and writing the DataFrame to our storage file. Since we loop nine times, we create nine additional parquet files, from part-00001.parquet through part-00009.parquet. We also see nine additional transaction log entries, from 00001.json through 00009.json.

In step 3, we create one more patient tuple, convert it to a DataFrame and write it to our Delta table. This creates one additional file (part-00010.parquet). Our transaction log has our standard log entry (00010.json) with the add file action for our part-00010.parquet file. But the interesting fact is that it also creates a 000010.checkpoint.parquet file. This is the checkpoint that we mentioned earlier. A checkpoint is generated every 10 commits. This parquet file contains the entire state of our table at the time of the commit in native Parquet format.

In the last step, our code generates two more commits, creating the part-00011.parquet and part-00012.parquet, and two new log entries with add file entries pointing to these files.

If Delta Lake needs to recreate the state of the table, it will simply read the checkpoint file (000010.checkpoint.parquet), and re-apply the two additional log entries (00011.json and 00012.json).

Displaying the checkpoint file

Now that we have generated out checkpoint.parquet file, let’s have a look at its content using the /chapter02/readCheckPointFile.py python file:

# Set our output path for our Delta files
DATALAKE_PATH = "/book/chapter02/transactionLogCheckPointExample"
CHECKPOINT_PATH = "/_delta_log/00000000000000000010.checkpoint.parquet"
# Read the checkpoint.parquet file
checkpoint_df =                                \
  spark                                        \
  .read                                      \
  .format("parquet")                         \
  .load(f"{DATALAKE_PATH}{CHECKPOINT_PATH}")
# Display the checkpoint dataframe
checkpoint_df.show()

Notice how we do a parquet format read here, because the checkpoint file is indeed stored in Parquet format, not Delta format.

The contents of the checkpoint_df dataframe is shown below:

+----+--------------------+------+--------------------+--------+
| txn|                 add|remove|            metaData|protocol|
+----+--------------------+------+--------------------+--------+
|null|{part-00000-f7d9f...|  null|                null|    null|
|null|{part-00000-a65e0...|  null|                null|    null|
|null|{part-00000-4c3ea...|  null|                null|    null|
|null|{part-00000-8eb1f...|  null|                null|    null|
|null|{part-00000-2e143...|  null|                null|    null|
|null|{part-00000-d1d13...|  null|                null|    null|
|null|{part-00000-650bf...|  null|                null|    null|
|null|{part-00000-ea06e...|  null|                null|    null|
|null|{part-00000-79258...|  null|                null|    null|
|null|{part-00000-23558...|  null|                null|    null|
|null|                null|  null|                null|  {1, 2}|
|null|                null|  null|{376ce2d6-11b1-46...|    null|
|null|{part-00000-eb29a...|  null|                null|    null|
+----+--------------------+------+--------------------+--------+

As we can see, the checkpoint file contains columns for the different actions (add, remove, metadata and protocol). We see our add file actions for the different parquet files, we also see our update metadata action from when we created our Delta dataset and the change protocol action which results from the initial write.

Note that DataFrame.show() will not show the DataFrame’s record in order.The change protocol and update metadata records are always the first records in the checkpoint file, followed by the different add-file actions.

¹ https://www.docker.com/

² https://hub.docker.com/r/apache/spark

³ https://github.com/benniehaelen/delta-lake-up-and-running

⁴ https://pypi.org/project/delta-spark/

⁵ https://community.cloud.databricks.com/

⁶ https://docs.databricks.com/dev-tools/dbx.html

Chapter 3. Basic Operations on Delta Tables

A Note for Early Release Readers

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 3rd chapter of the final book. The GitHub repo is https://github.com/benniehaelen/delta-lake-up-and-running.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at gobrien@oreilly.com.

Delta tables can be created in a variety of ways. How you will create your tables will largely depend on your familiarity with the toolset. If you are primarily a SQL developer, you can use SQL’s CREATE TABLE to create a Delta Table, while Python users may prefer the DataFrameWriter API or the fine-grained and easy to use DeltaTableBuilder API.

When creating tables you can define GENERATED columns, the values of which are automatically generated based on a user-specified function over other columns in the Delta table. While some restrictions apply, generated columns are a powerful way to enrich your Delta table schemas.

Delta tables can be read by standard ANSI SQL, or by using the popular PySpark DataFrameReader API. You can write to a Delta table by using the classic SQL INSERT statement, or you can append a DataFrame to the table. Finally, a great way to quickly append large amounts of data is to leverage the SQL COPY INTO option.

Partitioning a Delta table based upon your frequently-used query pattern can dramatically improve your query and DML performance. The individual files that make up your Delta dataset will be organized in sub-directories that align to the values of your partitioning columns.

Delta Lake allows you to associate custom metadata with the commit entries in your transaction log. This can be leveraged to tag sensitive commits for auditing purposes. You can also store custom tags in your table properties, so just like you can have tags for your cloud resources, you can now associate those tags with your Delta tables. You can also modify certain Delta capabilities. For example, you can associate the delta.appendonly property to a table to prevent deletes and updates.

Creating a Delta table

Delta Lake enables us to create tables in three different ways:

SQL Data Definition Language (DDL) commands

SQL developers are already very familiar with the classic CREATE TABLE command, and you will be able to use it to create a Delta Table by adding just a few attributes.

PySpark DataFrameWriter API

Big Data Python (and Scala) developers will very likely already be very familiar with this API, and you will be able to continue to use it with Delta tables.

DeltaTableBuilder API

This is a new API is specifically designed for Delta tables. It uses the popular Builder pattern, and it gives very fine-grained control over every Delta table- and column attribute.

In the next sections you get hands on with each one of the above table creation methods.

Creating a Delta Table with SQL DDL

The version of SQL used in Spark compute environments is called Spark SQL, which is a variant of ANSI SQL supported by Spark. Spark SQL is generally compatible with the ANSI standard SQL. Refer to the Spark documentation for additional details on the Spark SQL variant.

As I mentioned earlier, you can use the standard SQL DDL commands in Spark SQL to create a Delta table¹.

%sql
-- Create a Delta table by specifying the delta format, followed
-- by the path in quotes
CREATE TABLE IF NOT EXISTS delta.`/mnt/datalake/book/chapter03/rateCard`
(
    rateCodeId   INT,
    rateCodeDesc STRING
)
USING DELTA

The notation that you are using the for the table name is the file_format | `path_to_table` notation, where the file_format is delta, and path_to_table is the path to the delta dataset.

Note

Using the above format can get tedious, since in the real world, file paths can get rather long. This is where catalogs come in. A catalog allows you to register a table with a database.table_name notation, where database is a logical grouping of tables, and table_name is a shorthand for the table. For example, if you first created a database named taxidb as follows:

%sql
create database if not exists taxidb;

Then you could create the above table as follows:

%sql
-- Create the table using the taxidb catalog
CREATE TABLE IF NOT EXISTS taxidb.rateCard
(
    rateCodeId   INT,
    rateCodeDesc STRING
)
USING DELTA
LOCATION '/mnt/datalake/book/chapter03/rateCard'

From this point forward, you can refer to this Delta table as taxidb.rateCard table, which is easier to remember and type then delta.`/mnt/datalake/book/chapter03/rateCard`, or possibly an even longer path name. The most widely used catalog in the Spark ecosystem is the hive catalog.

When running a directory listing on the data lake location where the table was created, you can see that our directory is empty (since you have not loaded any data), except for the _delta_log directory, which contains the table’s transaction log.

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/rateCard
total 12
drwxrwxrwx 2 root root 4096 Dec  2 19:02 .
drwxrwxrwx 2 root root 4096 Dec  2 19:02 ..
drwxrwxrwx 2 root root 4096 Dec  2 16:40 _delta_log

Note

Please note that, since you are running this as a shell command in the Databricks community edition environment, you have to prefix our path for the ls command with /dbfs.

When you open the _delta_log directory, yousee our first transaction log entry:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/rateCard/_delta_log
total 15
drwxrwxrwx 2 root root 4096 Dec  2 19:02 .
drwxrwxrwx 2 root root 4096 Dec  2 19:02 ..
-rwxrwxrwx 1 root root 1886 Dec  2 19:02 00000000000000000000.crc
-rwxrwxrwx 1 root root  939 Dec  2 19:02 00000000000000000000.json

In the transaction log discussion in chapter 2 I described the different actions that can be written to our transaction log entry. One of those actions is the metaData action, which describes the schema of the table, the partitioning columns (if applicable), together with other information. This metaData action is always written to our first transaction log entry created for our new table.

To find this metadata action, you can do a search for the string metadata in the transaction entry:

%sh
grep metadata /dbfs/mnt/datalake/book/chapter03/rateCard/_delta_log/00000000000000000000.json > /tmp/metadata.json
python -m json.tool /tmp/metadata.json

This produces the following output:

{
    "metaData": {
        "id": "f79c4c11-a807-49bc-93f4-2bbe778e2a04",
        "format": {
            "provider": "parquet",
            "options": {}
        },
        "schemaString": "{\"type\":\"struct\",\"fields\":[{\"name\":\"rateCodeId\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"rateCodeDesc\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}",
        "partitionColumns": [],
        "configuration": {},
        "createdTime": 1670007736533
    }
}

Here, you see that Delta Lake has written the schema of the table to the transaction log entry, together with some auditing and partitioning information.

Note

In the above command, you first perform a grep command, which searches for the string metadata in the transaction log entry. You then write the output of that to a temp file. The next line uses python -m json.tool with the temp file as input. The json.tool python module will “pretty print” the content of a JSON file, which can be really handy.

The DESCRIBE statement

The SQL DESCRIBE command can be used to return the basic metadata for a Parquet or Delta Table.

The metadata returned for a table includes one line for each column in the table with the following information:

The column name.

The column data type.

Any comments that were applied to the column.

Below is an example of the DESCRIBE command at the table level:

%sql
DESCRIBE TABLE taxidb.rateCard;
+--------------+-----------+---------+
| col_name     | data_type | comment |
+--------------+-----------+---------+
| rateCodeId   | int       | <null>  |
| rateCodeDesc | string    | <null>  |
+--------------+-----------+---------+

When you want to find the Delta Lake-specific attributes, you can also use the DESCRIBE TABLE EXTENDED command which provides more detailed metadata information, including the following generic attributes:

The Catalog Name for the database in which the table was created (in this case the Hive metastore).

The Hive Database

The table name

The location of the underlying files.

The owner of the table

The table properties

The following Delta Lake-specific attributes are also included:

The delta.minReaderVersion is the minimum required protocol reader version for a reader that can read from this Delta table.

The delta.minWriterVersion is the minimum required protocol writer version for a writer that can write to this Delta table. Please refer to the Delta Lake documentation for a full listing of all available table properties.

Below is an example of the DESCRIBE TABLE EXTENDED command.

%sql
DESCRIBE TABLE EXTENDED taxidb.rateCard;

The generates the following output:

+------------------------------+-----------------------------------------------------+---------+
| col_name                     | data_type                                           | comment |
+------------------------------+-----------------------------------------------------+---------+
| rateCodeId                   | int                                                 | <null>  |
| rateCodeDesc                 | string                                              | <null>  |
|                              |                                                     |         |
| # Detailed Table Information |                                                     |         |
| Catalog                      | hive_metastore                                      |         |
| Database                     | taxidb                                              |         |
| Table                        | ratecard                                            |         |
| Type                         | EXTERNAL                                            |         |
| Location                     | dbfs:/mnt/datalake/book/chapter03/rateCard          |         |
| Provider                     | delta                                               |         |
| Owner                        | root                                                |         |
| Table Properties             | [delta.minReaderVersion=1,delta.minWriterVersion=2] |         |
+------------------------------+-----------------------------------------------------+---------+

So far, I have covered the creation of Delta tables with theSQL Data Definition Language (DDL). In the next section, I will switch back to Python, and look at how you can use our familiar PySpark DataFrames to create new Delta tables.

Creating Delta tables with the DataFrameWriter API

Spark DataFrames resemble relational database tables or Excel spreadsheets with headers. The data resides in rows and columns of different data types. The collection of functions that lets us read, write and manipulate DataFrames is collectively known as the Spark DataFrameWriter API.

Note

Creating an Managed table

When you read the Spark and/or Delta documentation, you will hear the term managed and unmanaged table. A Delta table that is created with a location is known as an unmanaged table. For these tables, Spark only manages the metadata, and requires the user to specify the exact location where you wish to save the underlying data for the table, or alternatively, the source directory from which data will be pulled to create the table (if you are using the DataFrameWriter API).

A Delta table that is created without a location is referred to as a managed table. For managed tables, Spark manages both the metadata and the “real” data. The data is stored under the /spark-warehouse sub-folder (in the Apache Spark scenario) or the /user/hive/warehouse folder (when running on Databricks), which is the default for managed tables.

One of the benefits of the DataFrameWriter API is that you can simultaneously create a table and insert data into it from a Spark DataFrame as is shown in the following code snippet².

INPUT_PATH = '/databricks-datasets/nyctaxi/taxizone/taxi_rate_code.csv'
DELTALAKE_PATH = 'dbfs:/mnt/datalake/book/chapter03/createDeltaTableWithDataFrameWriter'
# Read the Dataframe from the input path
df_rate_codes = spark                                              \
                .read                                              \
                .format("csv")                                     \
                .option("inferSchema", True)                       \
                .option("header", True)                            \
                .load(INPUT_PATH)
# Save our DataFrame as a managed Hive table
df_rate_codes.write.format("delta").saveAsTable(taxidb.rateCard')

Here, you first populate the dataframe from the taxi_rate_code.csv file, and then save the DataFrame as a Delta table by specifying the .format(“delta”) option. The schema of the table will be the schema of our DataFrame. Notice that this will be a managed table since you did not specify a location for our data file. You can verify this by running the SQL DESCRIBE TABLE EXTENDED command as is shown.

%sql
DESCRIBE TABLE EXTENDED taxidb.rateCard;
+------------------------------+-----------------------------------------------------+---------+
| col_name                     | data_type                                           | comment |
+------------------------------+-----------------------------------------------------+---------+
| RateCodeID                   | int                                                 | <null>  |
| RateCodeDesc                 | string                                              | <null>  |
|                              |                                                     |         |
| # Detailed Table Information |                                                     |         |
| Catalog                      | hive_metastore                                      |         |
| Database                     | taxidb                                              |         |
| Table                        | ratecard                                            |         |
| Type                         | MANAGED                                             |         |
| Location                     | dbfs:/user/hive/warehouse/taxidb.db/ratecard        |         |
| Provider                     | delta                                               |         |
| Owner                        | root                                                |         |
| Is_managed_location          | true                                                |         |
| Table Properties             | [delta.minReaderVersion=1,delta.minWriterVersion=2] |         |
+------------------------------+-----------------------------------------------------+---------+

You see that the data for the table lives in the /user/hive/warehouse location, and that the type of the table is set to MANAGED.

If you run a SELECT on the table, you can see the data was indeed loaded successfully from the .csv file:

%sql
select * from taxidb.rateCard
+------------+-----------------------+
| RateCodeID | RateCodeDesc          |
+------------+-----------------------+
| 1          | Standard Rate         |
| 2          | JFK                   |
| 3          | Newark                |
| 4          | Nassau or Westchester |
| 5          | Negotiated fare       |
| 6          | Group ride            |
+------------+-----------------------+

Creating an unmanaged table

You can create an unmanaged table by specifying both the path and the name of the Delta Table.

In the following code, you execute both steps in sequence.

First, drop the existing table.

%sql
-- Drop the existing table
drop table if exists taxidb.rateCard;
Next, write out the Delta dataset and create the table.
# Next, create our Delta Table, specifying both
# the path and the Delta Table Name
df_rate_codes                           \
        .write                          \
        .format("delta")                \
        .mode("overwrite")              \
        .option('path', DELTALAKE_PATH) \
        .saveAsTable('taxidb.rateCard')

Again by performing a simple select you can verify that the data of the DataFrame has been loaded.

%sql
select * from taxidb.rateCard
+------------+-----------------------+
| RateCodeID | RateCodeDesc          |
+------------+-----------------------+
| 1          | Standard Rate         |
| 2          | JFK                   |
| 3          | Newark                |
| 4          | Nassau or Westchester |
| 5          | Negotiated fare       |
| 6          | Group ride            |
+------------+-----------------------+

Creating a Delta table with the DeltaTableBuilder API

The last way to create a Delta table is by using the DeltaTableBuilder API. Since it is designed to work with Delta tables, it offers a higher degree of fine-grained control versus the traditional DataFrameWriter API. It is easier for a user to specify additional information such as column comments, table properties, and generated columns.

The Builder design pattern is a popular design pattern in software languages. The Builder pattern aims to “Separate the construction of a complex object from its representation so that the same construction process can create different representations”. It is used to construct a complex object step by step where the final step will return the object.

The complex object you are building in this case is Delta Table. Delta tables support so many options that it is challenging to design a standard API with many arguments for a single function. Instead the DeltaTableBuilder has a number of small methods, such as addColumn(), which all return a reference to the Builder object. That way you can keep adding other calls to addColumn(), or other methods of the Builder. The final method you call is execute(), which gathers up all the attributes received, creates the Delta table and returns a reference to the table back to the caller. In order to use the DeltaTableBuilder you will need the follogin import:

from delta.tables import *

This is an example that creates a managed table.³.

#
# In this Create Table, you do NOT specify a location, so you are
# creating a MANAGED table
#
DeltaTable.createIfNotExists(spark)                              \
    .tableName("taxidb.greenTaxis")                              \
    .addColumn("RideId", "INT", comment = "Primary Key")         \
    .addColumn("VendorId", "INT", comment = "Ride Vendor")       \
    .addColumn("EventType", "STRING")                            \
    .addColumn("PickupTime", "TIMESTAMP")                        \
    .addColumn("PickupLocationId", "INT")                        \
    .addColumn("CabLicense", "STRING")                           \
    .addColumn("DriversLicense", "STRING")                       \
    .addColumn("PassengerCount", "INT")                          \
    .addColumn("DropTime", "TIMESTAMP")                          \
    .addColumn("DropLocationId", "INT")                          \
    .addColumn("RateCodeId", "INT", comment = "Ref to RateCard") \
    .addColumn("PaymentType", "INT")                             \
    .addColumn("TripDistance", "DOUBLE")                         \
    .addColumn("TotalAmount", "DOUBLE")                          \
    .execute()

Since each method returns a reference to the Builder object, you can keep calling .addColumn() to add each column. Finally, you call .execute() to create the Delta Table.

Generated Columns

Delta Lake supports generated columns which are a special type of column, the values of which are automatically generated based on a user-specified function over other columns in the Delta table. When you write to a Delta table with generated columns and don’t explicitly provide values for them, Delta Lake automatically computes the values.

Let’s create an example next. To stay with our taxi theme, you will create a simple version of a yellow taxi table.

%sql
CREATE TABLE taxidb.YellowTaxis
(
    RideId               INT        COMMENT 'This is our primary Key column',
    VendorId             INT,
    PickupTime           TIMESTAMP,
    PickupYear           INT        GENERATED ALWAYS AS(YEAR  (PickupTime)),
    PickupMonth          INT        GENERATED ALWAYS AS(MONTH (PickupTime)),
    PickupDay            INT        GENERATED ALWAYS AS(DAY   (PickupTime)),
    DropTime             TIMESTAMP,
    CabNumber            STRING     COMMENT 'Official Yellow Cab Number'
) USING DELTA
LOCATION "/mnt/datalake/book/chapter03/YellowTaxis.delta"
COMMENT 'Table to store Yellow Taxi data'

You see the columns with the GENERATED ALWAYS AS, which extract the YEAR, MONTH and DAY from the PickupTime column. The values for these columns will automatically be populated in when you insert a record.

%sql
INSERT INTO taxidb.YellowTaxis
    (RideId, VendorId, PickupTime, DropTime, CabNumber)
VALUES
    (5, 101, '2021-7-1T8:43:28UTC+3', '2021-7-1T8:43:28UTC+3', '51-986')

When you select the record, you can see that the generated columns are automatically populated.

%sql
SELECT PickupTime, PickupYear, PickupMonth, PickupDay FROM taxidb.YellowTaxis
+---------------------------+------------+-------------+-----------+
| pickupTime                | pickupYear | pickupMonth | pickupDay |
+---------------------------+------------+-------------+-----------+
| 2021-07-01 05:43:28+00:00 | 2021       | 7           | 1         |
+---------------------------+------------+-------------+-----------+

When you do explicitly provide a value for a generated column, the value must satisfy the constraint (<value> ⇔ generation expression) IS TRUE or the insert will fail with an error.

The expression you use in GENERATED ALWAYS AS can be any Spark SQL function that always returns the same result when given the same argument values, with a few exceptions I will touch on soon. You might think you could use a GENERATED column to generate a column with a unique ID like this:

%sql
CREATE OR REPLACE TABLE default.dummy
(
    ID   STRING GENERATED ALWAYS AS (UUID()),
    Name STRING
) USING DELTA

However, when you try to run this, you get the following error message:

Found uuid(). A generated column cannot use a non deterministic expression.

Indeed the UUID() function will return a different value for each invocation, which violates the above rule. There are a few exceptions to this rule for the following types of functions:

User-defined functions.

Aggregate Functions

Window Functions

Functions returning multiple rows.

GENERATED ALWAYS AS columns using the functions listed are valid, and can indeed be very useful in a number of scenarios, like calculating a standard deviation of a given sample of records.

Reading a Delta table

You have a few options when reading a table: SQL and PySpark using the DataFrameReader.

When I use a notebook in the Databricks Community Edition, I tend to use both SQL and PySpark cells within the notebook. Some things, like a quick SELECT, are just easier and faster to do in SQL, while complex operations are sometimes easier expressed in PySpark and the DataFrameReader. This is of course also dependent on the experience and preferences of the engineer. I recommend a pragmatic approach using a healthy mix of both, depending on the problem you are currently solving.

Reading a Delta Table with SQL

To read a Delta table, you can simply open a SQL cell and write your SQL query. If you set up your environment as specified in the GitHub READ.ME file, you will have a Delta dataset in the /mnt/datalake/book/chapter03/YellowTaxisDelta folder:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDelta
total 236955
drwxrwxrwx 2 root root      4096 Dec  4 18:04 .
drwxrwxrwx 2 root root      4096 Dec  2 19:02 ..
drwxrwxrwx 2 root root      4096 Dec  4 16:41 _delta_log
-rwxrwxrwx 1 root root 134759123 Dec  4 18:04 part-00000-5969fc66-b89b-4a05-b18c-76a39f6b11ed-c000.snappy.parquet
-rwxrwxrwx 1 root root 107869302 Dec  4 18:04 part-00001-42457d47-e2e3-4499-9fce-dfa2629b2a16-c000.snappy.parquet

You can quickly create a Delta table on top of this Delta dataset as follows:⁴

%sql
CREATE TABLE taxidb.YellowTaxis
USING DELTA
LOCATION "/mnt/datalake/book/chapter03/YellowTaxisDelta/"
Once you have created the table you  can do a quick count on the number of records.
%sql
SELECT
    COUNT(*)
FROM
    taxidb.yellowtaxis
This gives us teh following count:
+----------+
| count(1) |
+----------+
| 9999995  |
+----------+

You can see there are almost 10 millions rows to work with. You can use another DESCRIBE command variant to get the details of the table:

%sql
DESCRIBE TABLE FORMATTED taxidb.YellowTaxis;

The DESCRIBE TABLE FORMATTED formats the output, making it a bit more readable.

+------------------------------+-----------------------------------------------------+---------+
| col_name                     | data_type                                           | comment |
+------------------------------+-----------------------------------------------------+---------+
| RideId                       | int                                                 | <null>  |
| VendorId                     | int                                                 | <null>  |
| PickupTime                   | timestamp                                           | <null>  |
| DropTime                     | timestamp                                           | <null>  |
| PickupLocationId             | int                                                 | <null>  |
| DropLocationId               | int                                                 | <null>  |
| CabNumber                    | string                                              | <null>  |
| DriverLicenseNumber          | string                                              | <null>  |
| PassengerCount               | int                                                 | <null>  |
| TripDistance                 | double                                              | <null>  |
| RatecodeId                   | int                                                 | <null>  |
| PaymentType                  | int                                                 | <null>  |
| TotalAmount                  | double                                              | <null>  |
| FareAmount                   | double                                              | <null>  |
| Extra                        | double                                              | <null>  |
| MtaTax                       | double                                              | <null>  |
| TipAmount                    | double                                              | <null>  |
| TollsAmount                  | double                                              | <null>  |
| ImprovementSurcharge         | double                                              | <null>  |
|                              |                                                     |         |
| # Detailed Table Information |                                                     |         |
| Catalog                      | hive_metastore                                      |         |
| Database                     | taxidb                                              |         |
| Table                        | YellowTaxis                                         |         |
| Type                         | EXTERNAL                                            |         |
| Location                     | dbfs:/mnt/datalake/book/chapter03/YellowTaxisDelta  |         |
| Provider                     | delta                                               |         |
| Owner                        | root                                                |         |
| Table Properties             | [delta.minReaderVersion=1,delta.minWriterVersion=2] |         |
+------------------------------+-----------------------------------------------------+---------+

Because Spark SQL supports most of the ANSI SQL subset you can use any type of complex query. Below is an example that returns the CabNumbers with the most expensive FareAmounts over $50.

SELECT
    CabNumber,
    AVG(FareAmount) AS AverageFare
FROM
    taxidb.yellowtaxis
GROUP BY
    CabNumber
HAVING
     AVG(FareAmount) > 50
ORDER BY
    2 DESC
LIMIT 5
Which gives us:
+-----------+-------------+
| cabnumber | AverageFare |
+-----------+-------------+
| SIR104    | 111.5       |
| T628190C  | 109.0       |
| PEACE16   | 89.7        |
| T439972C  | 89.5        |
| T802013C  | 85.0        |
+-----------+-------------+

You can also use SQL directly in Python with spark.sql, using standard SQL as the argument. Below is a simple Python snippet that performs the same query as previous SQL query:

number_of_results = 5
sql_statement = f"""
SELECT
    CabNumber,
    AVG(FareAmount) AS AverageFare
FROM
    taxidb.yellowtaxis
GROUP BY
    CabNumber
HAVING
     AVG(FareAmount) > 50
ORDER BY
    2 DESC
LIMIT {number_of_results}"""
df = spark.sql(sql_statement)
display(df)

This produces the same results as the SQL.

+-----------+-------------+
| cabnumber | AverageFare |
+-----------+-------------+
| SIR104    | 111.5       |
| T628190C  | 109.0       |
| PEACE16   | 89.7        |
| T439972C  | 89.5        |
| T802013C  | 85.0        |
+-----------+-------------+

I recommend using the triple-quotes syntax, which makes it easy to span strings over multiple lines without having to use continuation lines. Also, notice how you have the variable number_of_results, and then convert the triple-quote string into an f-string and use the {} syntax to insert the variable for the limit.

Reading a table with PySpark

To read the same table in PySpark, you can use the DataFrameReader. For example, to implement the count of records, you use:

⁵df = spark.read.format("delta").table("taxidb.YellowTaxis")
print(f"Number of records: {df.count():,}")
With output:
Number of records: 9,999,995

Note that you specify the Delta format, since our table is a Delta table and you can use the .table() method to specify that you want to read the entire table. Finally, you use an f-string, this time with the “:,” formatter, which uses a comma separator for every three digits.

Next, let’s recreate the code for the top five average fares by cab number, which you did in SQL earlier. Below is the Python code:

# Make sure to import the functions you want to use
from pyspark.sql.functions import col, avg, desc
# Read YellowTaxis into our dataframe
df = spark.read.format("delta").table("taxidb.YellowTaxis")
# Perform the GROUP BY, average (AVG), HAVING and order by equivalents
# in pySpark
results = df.groupBy("CabNumber")                          \
            .agg(avg("FareAmount").alias("AverageFare"))   \
            .filter(col("AverageFare") > 50)               \
            .sort(col("AverageFare").desc())               \
            .take(5)
# Print out the result, since this is a list and not a DataFrame
# you an use list comprehension to output the results in a single
# line
[print(result) for result in results]

You’ll get the following the output:

Row(CabNumber='SIR104', AverageFare=111.5)
Row(CabNumber='T628190C', AverageFare=109.0)
Row(CabNumber='PEACE16', AverageFare=89.7)
Row(CabNumber='T439972C', AverageFare=89.5)
Row(CabNumber='T802013C', AverageFare=85.0)

You can simply use the groupBy() function to group by a column

Note

Note that the result of this is no longer a DataFrame, but a pyspark.sql.GroupedData instance, as illustrated in this code snippet:

# Perform a groupBy, and print out the type
print(type(df.groupBy("CabNumber")))

Which prints out:

<class 'pyspark.sql.group.GroupedData'>

Often, a developer new to pySpark might assume that a groupBy() returns a DataFrame, but it returns a GroupedData instance, so you have to use GroupedData methods such as agg() and filter() instance of DataFrame functions such as avg() and where()

To calculate an average, you first have to use the .agg() method, and within the method you can specify which aggregate you want to calculate, which in this case is .avg() (average). In Python, the equivalent of the HAVING condition is the .filter() method, within which you can specify the filter using a filter expression. Finally, you use the .sort() method to sort the data, and then use .take() to extract the first 5 results. Note that the .take() function will return a Python list. Since you have a list here, you can use list comprehension to output each result in the list.

Writing to a Delta table

There are various ways to write to a Delta Table. You might want to rewrite an entire table, or you might want to append to a table. The more advanced topics such as updates and merges will be discussed in the next chapter.

You first will clean out of our YellowTaxis table, so that you have a clean slate, and then you will use a traditional SQL INSERT statement to insert data. Next, you will append the data from a smaller CSV file.You will also take a quick look at the overwrite mode when writing a Delta dataset and finally you will use the SQL COPY INTO feature to merge in a large .CSV file.

Cleaning out the YellowTaxis Table

You can recreate our Delta table with a CREATE TABLE statement.⁶

%sql
CREATE TABLE taxidb.YellowTaxis
(
    RideId                  INT,
    VendorId                INT,
    PickupTime              TIMESTAMP,
    DropTime                TIMESTAMP,
    PickupLocationId        INT,
    DropLocationId          INT,
    CabNumber               STRING,
    DriverLicenseNumber     STRING,
    PassengerCount          INT,
    TripDistance            DOUBLE,
    RatecodeId              INT,
    PaymentType             INT,
    TotalAmount             DOUBLE,
    FareAmount              DOUBLE,
    Extra                   DOUBLE,
    MtaTax                  DOUBLE,
    TipAmount               DOUBLE,
    TollsAmount             DOUBLE,
    ImprovementSurcharge    DOUBLE

) USING DELTA
LOCATION "/mnt/datalake/book/chapter03/YellowTaxisDelta"

With our table setup, you are ready to insert data.

Insert Data With SQL INSERT

To insert a record into the YellowTaxis Delta Table you can use the SQL INSERT command:

%sql
INSERT INTO taxidb.yellowtaxis
(RideId, VendorId, PickupTime, DropTime,
 PickupLocationId, DropLocationId, CabNumber,
 DriverLicenseNumber, PassengerCount, TripDistance,
 RatecodeId, PaymentType, TotalAmount,
 FareAmount, Extra, MtaTax, TipAmount,
 TollsAmount, ImprovementSurcharge)
 VALUES(9999995, 1, '2019-11-01T00:00:00.000Z',
        '2019-11-01T00:02:23.573Z', 65, 71, 'TAC304',
        '453987', 2, 4.5, 1, 1, 20.34, 15.0, 0.5,
        0.4, 2.0, 2.0, 1.1)

This will insert one row.

num_affected_rows

num_inserted_rows

1

1

Verify the data has loaded correctly with a SQL SELECT statement:

%sql
select * from taxidb.YellowTaxis
+---------+----------+---------------------------+----------------------------------+------------------+----------------+-----------+---------------------+----------------+--------------+------------+-------------+-------------+------------+-------+--------+-----------+-------------+----------------------+
| RideId  | VendorId | PickupTime                | DropTime                         | PickupLocationId | DropLocationId | CabNumber | DriverLicenseNumber | PassengerCount | TripDistance | RatecodeId | PaymentType | TotalAmount | FareAmount | Extra | MtaTax | TipAmount | TollsAmount | ImprovementSurcharge |
+---------+----------+---------------------------+----------------------------------+------------------+----------------+-----------+---------------------+----------------+--------------+------------+-------------+-------------+------------+-------+--------+-----------+-------------+----------------------+
| 9999991 | 1        | 2019-11-01 00:00:00+00:00 | 2019-11-01 00:02:23.573000+00:00 | 65               | 71             | TAC304    | 453987              | 2              | 4.5          | 1          | 1           | 20.34       | 15.0       | 0.5   | 0.4    | 2.0       | 2.0         | 1.1                  |
+---------+----------+---------------------------+----------------------------------+------------------+----------------+-----------+---------------------+----------------+--------------+------------+-------------+-------------+------------+-------+--------+-----------+-------------+----------------------+

The output shows that all data has been loaded correctly.

Append a DataFrame to a Table

Now let’s append a DataFrame to our table. In this case you will load your DataFrame from a .CSV file. In order to correctly load the data, you don’t want to infer the schema. Instead you will use the schema of the YellowTaxis table which you know is correct.

You can easily extract the schema by loading up a DataFrame from the table:

df = spark.read.format("delta").table("taxidb.YellowTaxis")
yellowTaxiSchema = df.schema
print(yellowTaxiSchema)

This shows the table schema is as follows:

root
 |-- RideId: integer (nullable = true)
 |-- VendorId: integer (nullable = true)
 |-- PickupTime: timestamp (nullable = true)
 |-- DropTime: timestamp (nullable = true)
 |-- PickupLocationId: integer (nullable = true)
 |-- DropLocationId: integer (nullable = true)
 |-- CabNumber: string (nullable = true)
 |-- DriverLicenseNumber: string (nullable = true)
 |-- PassengerCount: integer (nullable = true)
 |-- TripDistance: double (nullable = true)
 |-- RatecodeId: integer (nullable = true)
 |-- PaymentType: integer (nullable = true)
 |-- TotalAmount: double (nullable = true)
 |-- FareAmount: double (nullable = true)
 |-- Extra: double (nullable = true)
 |-- MtaTax: double (nullable = true)
 |-- TipAmount: double (nullable = true)
 |-- TollsAmount: double (nullable = true)
 |-- ImprovementSurcharge: double (nullable = true)

Now that you have your schema, you can load a new DataFrame (df_for_append) from our append CSV file:

df_for_append = spark.read                            \
     .option("header", "true")        \
     .schema(yellowTaxiSchema)        \
     .csv("/mnt/datalake/book/data files/YellowTaxis_append.csv")
display(df_for_append)

You see the following output (partial output is displayed):

RideId	VendorId	 PickupTime	                 DropTime
9999996	1	2019-01-01T00:00:00.000+0000	  2022-03-01T00:13:13.000+0000
9999997	1	2019-01-01T00:00:00.000+0000	  2022-03-01T00:09:21.000+0000
9999998	1	2019-01-01T00:00:00.000+0000	  2022-03-01T00:09:15.000+0000
9999999	1	2019-01-01T00:00:00.000+0000	  2022-03-01T00:10:01.000+0000

You now have four additional rows, all with a VendorId of 1.

You can now append this CSV file to our Delta table.

df_for_append.write              \
     .mode("append")             \
     .format("delta")            \
     .save("/mnt/datalake/book/chapter03/YellowTaxisDelta")

This appends the data directly to our Delta table. Since you had one row in the table before from your INSERT statement, and you inserted 4 additional rows you know that you should now have five rows in the YellowTaxis table:

%sql
SELECT
    COUNT(*)
FROM
    taxidb.YellowTaxis
+----------+
| count(1) |
+----------+
| 5        |
+----------+

You now indeed have five rows.

Using the OverWrite mode when writing to a Delta Table

In the previous example you used the .mode(“append”) when using the DataFrameWriter API to write to a Delta Table. Delta Lake also supports the overwrite mode when writing to a Delta table. When you use this mode you will atomically replace all of the data in the table.

If you would have used the.mode(“overwrite”) in the previous code block you would have overwritten the entire YellowTaxis Delta dataset with just the df_for_append DataFrame.

Another important comment to make in this context is the fact that even if you use .mode(“overwrite”) in your code, the old files are not immediately physically deleted. Indeed, in order to support features such as time travel these files cannot be deleted immediately. You can use commands such as VACUUM to physically delete these files at a later time when you are sure they are no longer needed. Time travel and the VACUUM command are covered in detail in chapter x.

Inserting data with SQL COPY..INTO command

Finally, you can use the SQL COPY INTO command to append data to our table. This command is especially useful when you need to append very large amounts of data quickly.

You can use the following command to append the data from a CSV file:

%sql
COPY INTO taxidb.yellowtaxis
FROM (
    SELECT     RideId::Int
             , VendorId::Int
             , PickupTime::Timestamp
             , DropTime::Timestamp
             , PickupLocationId::Int
             , DropLocationId::Int
             , CabNumber::String
             , DriverLicenseNumber::String
             , PassengerCount::Int
             , TripDistance::Double
             , RateCodeId::Int
             , PaymentType::Int
             , TotalAmount::Double
             , FareAmount::Double
             , Extra::Double
             , MtaTax::Double
             , TipAmount::Double
             , TollsAmount::Double
             , ImprovementSurcharge::Double

        FROM '/mnt/datalake/book/DataFiles/YellowTaxisLargeAppend.csv'
)
FILEFORMAT = CSV
FORMAT_OPTIONS ("header" = "true")

All fields in a CSV file would be string, so you need to provide some type of schema with a SQL SELECT statement when you load the data. This provides the type of each column, ensuring that you are loading the right schema. Note the FILEFORMAT, in this case CSV, is specified. Finally, because our file has a header, you need to specify the header with FORMAT_OPTIONS.

The output of this statement is:

num_affected_rows

num_inserted_rows

9999995

9999995

You can see that you inserted almost 10 million rows in just a few seconds..

The COPY INTO command also keeps track of and will not reload any previously loaded files. You can test this by running the COPY INTO command again:

num_affected_rows

num_inserted_rows

0

0

As you can see no additional rows were loaded.

Finally, when you check your final row count,you will see that you now have one millions rows:

%sql
SELECT
    COUNT(*)
FROM
    taxidb.YellowTaxis
+----------+
| count(1) |
+----------+
| 10000000 |
+--------------+

Partitions

Delta tables are often accessed with a standard query pattern. For example data from IoT systems tends to be accessed by day, hour or even minute. The analysts querying the Yellow cab data might want to access the data by VendorId and so on.

These use cases lend themselves well to partitioning. Partitioning your data to align with your query patterns can dramatically speed up your query performance, especially when combined with other performance optimizations such as z-ordering⁷. A Delta table partition is composed of a folder with a subset of data rows that share the same value for one or more column(s).

Note

Note that this type of on-disk partitioning should not be confused with the partitioning that Spark applies when processing a DataFrame. Spark applies in-memory partitioning to enable Tasks to run in parallel and independently on a large number of nodes in a Spark cluster.

For example, the Yellow taxi data, the partitioning column could be VendorId. After your table is partitioned, individual folders will be created for each VendorId. The folder name will have VendorId=XX as the last part of their folder name:

drwxrwxrwx 2 root root 4096 Dec 13 15:16 VendorId=1
drwxrwxrwx 2 root root 4096 Dec 13 15:16 VendorId=2
drwxrwxrwx 2 root root 4096 Dec 13 15:16 VendorId=4

Once your table is partitioned, all queries with predicates that include the partition columns will run much faster, since Spark can immediately select the folder with the correct partition.

You can partition data when you create a Delta table by specifying a PARTITIONED BY clause.

Partitioning by a Single Column

Let’s take our YellowTaxis table, and create a new version that is partitioned by VendorId.

First, create the partitioned table:⁸

%sql
CREATE TABLE taxidb.YellowTaxisPartitioned
(
    RideId                  INT,
    VendorId                INT,
    PickupTime              TIMESTAMP,
    DropTime                TIMESTAMP,
    PickupLocationId        INT,
    DropLocationId          INT,
    CabNumber               STRING,
    DriverLicenseNumber     STRING,
    PassengerCount          INT,
    TripDistance            DOUBLE,
    RatecodeId              INT,
    PaymentType             INT,
    TotalAmount             DOUBLE,
    FareAmount              DOUBLE,
    Extra                   DOUBLE,
    MtaTax                  DOUBLE,
    TipAmount               DOUBLE,
    TollsAmount             DOUBLE,
    ImprovementSurcharge    DOUBLE

) USING DELTA
PARTITIONED BY(VendorId)
LOCATION "/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned"

Notice the PARTITIONED BY(VendorId) clause.

Now that you have your table, you will load the data from our old YellowTaxis table, and write that data to the new table.

First, read the data with the DataFrameReader:

input_df = spark.read.format("delta").table("taxidb.YellowTaxis")

Next, use the DataFrameWriter to write the data to the partitioned Delta dataset:

input_df                                                               \
    .write                                                             \
    .format("delta")                                                   \
    .mode("overwrite")                                                 \
    .save("/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned")

Now when you look at the table’s directory, you’ll see a subdirectory for every VendorID:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned
drwxrwxrwx 2 root root 4096 Dec  5 17:39 .
drwxrwxrwx 2 root root 4096 Dec  2 19:02 ..
drwxrwxrwx 2 root root 4096 Dec  5 16:44 VendorId=1
drwxrwxrwx 2 root root 4096 Dec  5 16:44 VendorId=2
drwxrwxrwx 2 root root 4096 Dec  5 16:44 VendorId=4
drwxrwxrwx 2 root root 4096 Dec  5 16:44 _delta_log
When you look at the distinct VendorIds, you see that you indeed only have those 3 ID’s:
%sql
SELECT
    DISTINCT(VendorId)
FROM
    taxidb.YellowTaxisPartitioned;

You will see the same IDs:

+----------+
| VendorId |
+----------+
| 2        |
| 1        |
| 4        |
+----------+

The VendorId sub-directories contain the individual Parquet files, as shown below for VendorId = 4:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned/VendorId=4
total 3378
drwxrwxrwx 2 root root   4096 Dec  5 17:41 .
drwxrwxrwx 2 root root   4096 Dec  5 17:39 ..
-rwxrwxrwx 1 root root 627551 Dec  5 17:41 part-00000-...parquet
-rwxrwxrwx 1 root root 618844 Dec  5 17:41 part-00001-...parquet
-rwxrwxrwx 1 root root 616377 Dec  5 17:41 part-00002-...parquet
-rwxrwxrwx 1 root root 614035 Dec  5 17:41 part-00003-...parquet
-rwxrwxrwx 1 root root 612410 Dec  5 17:41 part-00004-...parquet
-rwxrwxrwx 1 root root 360432 Dec  5 17:41 part-00005-...parquet

Partitioning by Multiple Columns

You don’t have to partition by just one column. Indeed, you can use multiple hierarchical columns as partitioning columns. For example, for IoT data, you might want to partition by Day, Hour and Minute, because that is the most commonly used query pattern.

For example, let’s assume that you would not only want your YellowTaxis table partitioned by VendorId, but also by RateCodeId. First, you would have to drop your existing YellowTaxisPartitioned table and its underlying files. Next, you can recreate your table:

%sql
-- Create the table
CREATE TABLE taxidb.YellowTaxisPartitioned
(
    RideId                  INT,
    …
) USING DELTA
PARTITIONED BY(VendorId, RatecodeId) -- Partition by VendorId AND rateCodeId
LOCATION "/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned"

Notice the updated partition clause: PARTITIONED BY(VendorId, RatecodeId).

After this, you can reload the table the same way you did before. Once the table is loaded, you can take another look at the directory structure. The first level still looks the same:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned
drwxrwxrwx 2 root root 4096 Dec 13 15:33 .
drwxrwxrwx 2 root root 4096 Dec  2 19:02 ..
drwxrwxrwx 2 root root 4096 Dec 13 15:16 VendorId=1
drwxrwxrwx 2 root root 4096 Dec 13 15:16 VendorId=2
drwxrwxrwx 2 root root 4096 Dec 13 15:16 VendorId=4
drwxrwxrwx 2 root root 4096 Dec 13 15:16 _delta_log

When you take a look at the VendorId=1 directory, you see the partitioning by RatecodeId:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned/VendorId=1
drwxrwxrwx 2 root root 4096 Dec 13 15:35 .
drwxrwxrwx 2 root root 4096 Dec 13 15:33 ..
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=1
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=2
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=3
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=4
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=5
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=6
drwxrwxrwx 2 root root 4096 Dec 13 15:16 RatecodeId=99

And finally, when you query at the RatecodeId level:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned/VendorId=1/RatecodeId=1

You can see the Parquet files for that partition:

drwxrwxrwx 2 root root     4096 Dec 13 15:35 .
drwxrwxrwx 2 root root     4096 Dec 13 15:35 ..
-rwxrwxrwx 1 root root 10621353 Dec 13 15:35 part-00000-...parquet
-rwxrwxrwx 1 root root 10547673 Dec 13 15:35 part-00001-...parquet
-rwxrwxrwx 1 root root 10566377 Dec 13 15:35 part-00002-...parquet
-rwxrwxrwx 1 root root 10597523 Dec 13 15:35 part-00003-...parquet
-rwxrwxrwx 1 root root 10570937 Dec 13 15:35 part-00004-...parquet
-rwxrwxrwx 1 root root  6119491 Dec 13 15:35 part-00005-...parquet
-rwxrwxrwx 1 root root 13820133 Dec 13 15:35 part-00007-...parquet
-rwxrwxrwx 1 root root 24076060 Dec 13 15:35 part-00008-...parquet
-rwxrwxrwx 1 root root  6772609 Dec 13 15:35 part-00009-...parquet

Note

While this type of partitioning by multiple columns is supported, I want to point out some pitfalls. The number of files created will be the products of the cardinality of both columns, so in this case the number of Vendors times the number of Rate Cards. This can lead to the “Small Files Problem” where a large number of small parquet files are created.

Sometimes other solutions, such as Z-ordering can be more effective than partitioning. The Performance Tuning chapter of this book covers this

End note box

Checking if a Partition Exist

To determine whether a table contains a specific partition you can use the statement:

SELECT COUNT(*) > 0 FROM <table-name> WHERE <partition-column> = <value>

If the partition exists, true is returned.

The following SQL statement checks if the partition for VendorId=1 and RatecodeId=99 exists:

%sql
SELECT
    COUNT(*) > 0 AS `Partition exists`
FROM
    taxidb.YellowTaxisPartitioned
WHERE
    VendorId = 2 AND RateCodeId = 99

This will return true since this partition exists as was shown earlier.

Selectively updating Delta Partitions with ReplaceWhere

In the previous section, you saw how you can significantly speed up query operations by partitioning data. You can also selectively update one or more partitions with the replaceWhere option. Selectively applying updates to certain partitions is not always possible, some updates need to apply to the entire data lake. But, when applicable these selective updates can result in significant speed gains. Delta Lake can update partitions with excellent performance, while at the same time guaranteeing data integrity.

To see replaceWhere in action, let’s take a look at a particular partition:

%sql
SELECT
    RideId, VendorId, PaymentType
FROM
    taxidb.yellowtaxispartitioned
WHERE
    VendorID = 1 AND RatecodeId = 99 LIMIT 5

You will see a mixture of payment types in the results:

+---------+----------+-------------+
| RideId  | VendorId | PaymentType |
+---------+----------+-------------+
| 1137733 | 1        | 1           |
| 1144423 | 1        | 2           |
| 1214030 | 1        | 1           |
| 1223028 | 1        | 1           |
| 1300054 | 1        | 2           |
+---------+----------+-------------+

Let’s assume that you have a business reason that states that all PaymentTypes for VendorId=1 and RatecodeId=99 should be 3. You can use the following pySpark expression with replaceWhere to achieve that result:

 from pyspark.sql.functions import *
from pyspark.sql.functions import *
spark.read                                                              \
    .format("delta")                                                    \
    .load("/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned")   \
    .where((col("VendorId") == 1) & (col("RatecodeId") == 99))          \
    .withColumn("PaymentType", lit(3))                                  \
    .write                                                              \
    .format("delta")                                                    \
    .option("replaceWhere", "VendorId = 1 AND RateCodeId = 99")         \
    .mode("overwrite")                                                  \
    .save("/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned")

When you now look for the distinct PaymentTypes for this partition:

%sql
SELECT
    DISTINCT(PaymentType)
FROM
    taxidb.yellowtaxispartitioned
WHERE
    VendorID = 1 AND RatecodeId = 99

You see indeed that you only have the PaymentType = 3.

+-------------+
| PaymentType |
+-------------+
| 3           |
+-------------+

You can verify that the other partitions are not affected:

%sql
SELECT
    DISTINCT(PaymentType)
FROM
    taxidb.yellowtaxispartitioned
ORDER BY
    PaymentType
which shows all PaymentTypes:
+-------------+
| PaymentType |
+-------------+
| 1           |
| 2           |
| 3           |
| 4           |
+-------------+

ReplaceWhere can be particularly useful when you have to run an operation that can be computationally expensive, but you only need to run it on certain partitions.

In the YellowTaxi scenario, let’s assume that your data science team has requested that you run one of their algorithms on the YellowTaxis table. Initially, you can run it on your smallest partition, quickly retrieve the results and when approved, run the algorithm on all remaining partitions overnight.

User-Defined Metadata

For auditing or regulatory purposes, you might want to add a tag to certain SQL operations. For example, your project might require that you tag INSERTs to certain tables with a “General Data Protection Regulation” (GDPR) tag. Once you tag our INSERT with this tag, the auditing tool will be able to generate a complete list of statements which contain this particular tag.

You can specify these tags as user-defined strings in metadata commits made by a SQL operation. You can do this either by using the DataFrameWriter’s option userMetadata, or by using the SparkSession configuration spark.databricks.delta.commitInfo.userMetadata. If both options are specified, the DataFrameWriter’s option takes precedence.

Using the SparkSession to set Custom Metadata

Let’s look at the SparkSession configuration first. Assume that you have an INSERT operation, to which you want to assign a (GDPR) tag for auditing purposes. Below is a SQL example:

%sql
SET spark.databricks.delta.commitInfo.userMetadata=my-custom-metadata= { "GDPR": "INSERT Request 1x965383" };

This tag will apply to the next operation, which is a standard INSERT:

INSERT INTO taxidb.yellowtaxisPartitioned
(RideId, VendorId, PickupTime, DropTime,
 PickupLocationId, DropLocationId, CabNumber,
 DriverLicenseNumber, PassengerCount, TripDistance,
 RatecodeId, PaymentType, TotalAmount,
 FareAmount, Extra, MtaTax, TipAmount,
 TollsAmount, ImprovementSurcharge)
 VALUES(10000000, 3, '2019-11-01T00:00:00.000Z',
        '2019-11-01T00:02:23.573Z', 65, 71, 'TAC304',
        '453987', 2, 4.5, 1, 1, 20.34, 15.0, 0.5,
        0.4, 2.0, 2.0, 1.1)

Note that there is nothing special in the INSERT;it is a standard operation. The GDPR tag will automatically be applied to the CommitInfo in the transaction log. If you search the transaction log for the latest .json file you’ll see that the 00004.json is the last log entry:

%sh
ls -al /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned/_delta_log/*.json
Then w:
-rwxrwxrwx 1 .../_delta_log/00000000000000000000.json
-rwxrwxrwx 1 .../_delta_log/00000000000000000001.json
-rwxrwxrwx 1 .../_delta_log/00000000000000000002.json
-rwxrwxrwx 1 .../_delta_log/00000000000000000003.json
-rwxrwxrwx 1 .../_delta_log/00000000000000000004.json

nWhen you look at the 00004.json commit file, you can see the GDPR entry.

%sh
grep commit /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned/_delta_log/00000000000000000004.json > /tmp/commit.json
python -m json.tool /tmp/commit.json

This is the GDPR entry you’ll see:

{
    "commitInfo": {
        ….,
        "notebook": {
            "notebookId": "1106853862465676"
        },
        "clusterId": "0605-014705-r8puunyx",
        "readVersion": 3,
        "isolationLevel": "WriteSerializable",
        "isBlindAppend": true,
        "operationMetrics": {
           …
        },
        "userMetadata": "my-custom-metadata= { \"GDPR\": \"INSERT Request 1x965383\" }",
        "engineInfo": "Databricks-Runtime/10.4.x-scala2.12",
        "txnId": "99f2f31c-8c01-4ea0-9e23-c0cbae9eb82a"
    }
}

Note

The SET statement will stay in effect for subsequent operations within your current Spark session, so if you want to continue inserting data without adding the GDPR metadata, you need to update the SET to an empty string, or use the RESET operation. Be aware that RESET will reset all spark properties, not just the metadata one!

Using the DataFrameWriter to set Custom Metadata

You can also use the DataFrameWrite with the “userMetadata” option to insert custom tags, as is shown in the example below:

df_for_append.write                                                \
            .mode("append")                                        \
            .format("delta")                                       \
            .option("userMetadata", '{"PII": "Confidential XYZ"}') \
            .save("/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned")

When you look at the corresponding JSON entry, will see see the tags in the commitInfo:

%sh
grep commit /dbfs/mnt/datalake/book/chapter03/YellowTaxisDeltaPartitioned/_delta_log/00000000000000000005.json > /tmp/commit.json
python -m json.tool /tmp/commit.json
{
    "commitInfo": {
        ….
        …
        "userMetadata": "{\"PII\": \"Confidential XYZ\"}",
	…
    }
}

¹ Git repo location: /Chapter03/02 - CreateDeltaTableWithSql

² Git repo location: /chapter03/04 - The DataFrameWriter API

³ Git repo location: /chapter03/05 - The DeltaTableBuilder API

⁴ Git Repo location: /chapter03/07 - Read Table with SQL

⁵ Git Repo Location: /chapter03/Read Table with PySpark

⁶ Git Repo Location: /chapter03/10 - Writing to a Delta Table

⁷ Z-ordering is covered in the next chapter: Performance Tuning

⁸ Git repo Location: /chapter03/11 - Partitions

Chapter 4. Performance Tuning

A Note for Early Release Readers

With Early Release ebooks, you get books in their earliest form—the author’s raw and unedited content as they write—so you can take advantage of these technologies long before the official release of these titles.

This will be the 5th chapter of the final book. The GitHub repo is https://github.com/benniehaelen/delta-lake-up-and-running.

If you have comments about how we might improve the content and/or examples in this book, or if you notice missing material within this chapter, please reach out to the editor at gobrien@oreilly.com.

Any time that you are storing and retrieving data, whether it be with a traditional RDBMS or with Delta tables, how you choose to organize the data in the underlying storage format can have significant effects on the time it takes to perform table operations and queries.

In general, performance tuning refers to the process of optimizing the performance of a system, and in the context of Delta tables this involves optimizing how the data is stored and retrieved.Historically, how data is retrieved is accomplished by either processing data at a faster rate (increasing RAM or CPU) or by reducing the amount of data that needs to be read by skipping non-relevant data. Delta Lake provides a number of different techniques that can be combined to accelerate data retrieval by efficiently reducing the amount of files and data that needs to be read during operations.

By leveraging good performance tuning strategies on your Delta tables, you can significantly improve performance of execution times, especially when dealing with large tables or resource-intensive data lake operations and queries.

Compaction

When performing DML operations on a Delta table, oftentimes new data is written in very small, new files and directories. The resulting files can result in slower queries and operations because of the volume of file metadata and total number of files that need to be read.

Due to the problems that many small files can cause as described above, ideally you should rewrite a large number of small files into a smaller number of larger files. By repartitioning a table to a smaller number of files, you can compact it for better performance. This consolidation of files is called compaction, or bin-packing. When performing compaction you can also specify the option dataChange = False, indicating that the operation does not change the data and simply rearranges the data layout. This would ensure that any concurrent operations on the table are minimally affected during compaction.

The example below can be found in the script “Compaction”¹ that should be run after executing the “Chapter Initialization” script for chapter 5², which creates a Delta table with data spread across 1000 files in the Delta dataset. This example will demonstrate using the repartition method to specify how many files you want the data to be partitioned into, while also specifying the option dataChange = False to only rearrange the data:

# define the path and number of files to repartition
path = "/mnt/datalake/book/chapter05/YellowTaxisDelta"
numberOfFiles = 5

# read the delta table and repartition it
spark.read\
 .format("delta")\
 .load(path)\
 .repartition(numFiles)\
 .write\
 .option("dataChange", "false")\
 .format("delta")\
 .mode("overwrite")\
 .save(path)

***Begin warning box***

When data is compacted, by default Delta Lake sets dataChange = true. This can break concurrent operations on the table, such as down-stream streaming consumers when the table is used as a streaming source.

Conversely, when dataChange = false on an operation that changes data can corrupt the data in the table.

***End warning box***

Once you repartition the table, you will see that the table is comprised of 5 files:

%sql
DESCRIBE DETAIL taxidb.YellowTaxis

Output (only relevant portions shown):

+--------+-------------------------------+----------+
| format | name                          | numFiles |
+--------+-------------------------------+----------+
| delta  | spark_catalog.taxidb.tripdata | 5        |
+--------+-------------------------------+----------+

By using compaction, you can create fewer part-files that aim for a similar size and that optimizes the table reads.

Optimize

Compaction allows you to manually specify how many large files you want smaller files consolidated into. In Delta Lake, you can trigger this compaction and let Delta Lake determine the optimal number of large files you want with the OPTIMIZE command.

The OPTIMIZE command aims to remove unnecessary files from the transaction log while also producing evenly-balanced data files with respect to their size on disk. The smaller files are compacted into new, larger files up to 1GB.

Figure 5-1 shows how OPTIMIZE consolidates smaller files into larger files. Keep in mind, OPTIMIZE does not take into account how the data is organized within the files.

Figure 4-1. 1 - Delta datasets before and after OPTIMIZE

As you can see in Figure 5-1:

Your Delta table is comprised of small files containing data with no particular order. In this case there are four files with two rows each.

You run OPTIMIZE to reduce the number of files that need to be read during operations.

The small files in your Delta table are compacted into new, larger files up to 1 GB. In this case, two files with four rows each.

***Begin note box***

Unlike compaction achieved through the repartition method, there is no need to specify the dataChange option. OPTIMIZE uses snapshot isolation when performing the command so concurrent operations and downstream streaming consumers remain uninterrupted.

***End note box***

Let’s walk through an example of OPTIMIZE. You will start with a taxidb.YellowTaxis table that has been repartioned into 1000 files to simulate a scenario where you are consistently inserting data into a table³. Using the “Optimize” script for chapter 5, run the OPTIMIZE command. The output will provide metrics of the operation.

%sql

OPTIMIZE taxidb.YellowTaxis

Output (only relevant portions shown):

+-----------------------------------------------------------------------------------------------------+
| metrics                                                                           |
+-----------------------------------------------------------------------------------------------------+
| {"numFilesAdded": 9, "numFilesRemoved": 1000 "filesAdded":{...”totalFiles”: 9,                     |
|  “totalSize”: 2096274374..."filesRemoved":{...”totalFiles”: 1000, “totalSize”: 2317072851           |
+-----------------------------------------------------------------------------------------------------+

You can see below that after running the OPTIMIZE command on the table that 1,000 files were removed and 9 files were added.

***Begin note box***

When comparing the total size of files removed to files added, you will notice that the size of the file stayed relatively the same, even increasing slightly. The NYC Taxi dataset is mainly integers, so you see little compression from organizing the data. If your data contains lots of string values, you will see much better compression after running OPTIMIZE.

***End note box***

It is important to note that the 1,000 files that were removed were not physically removed from the underlying storage, rather they were only logically removed from the transaction log. These files will be physically removed from the underlying storage next time you run VACUUM, which will be covered in detail in chapter 7.

Optimization using OPTIMIZE is also idempotent, meaning that if it is run twice on the same table or subset of data, the second run has no effect. You can see below that if you run the same command again on the taxidb.YellowTaxis table the file statistics indicate 0 files were added and 0 files were removed.

%sql
OPTIMIZE taxidb.YellowTaxis

Output (only relevant portions shown):

+-----------------------------------------------------------------------------------------------------+
| metrics                                                                           |
+-----------------------------------------------------------------------------------------------------+
| {"numFilesAdded": 0, "numFilesRemoved": 0 "filesAdded":{...”totalFiles”: 0, “totalSize”: 0...       |
+-----------------------------------------------------------------------------------------------------+

You can also optimize on specific subsets of data rather than optimizing the entire table. This can be useful when you are only performing DML operations on a specific partition (you will learn more about partitions later in this chapter) and need to optimize just that partition(s). You can specify an optional partition predicate using WHERE. Suppose you are only adding and updating data on a regular basis in the partition for the current year, in this case that is 2022. After adding 2022 to the partition predicate, you will notice that after running the command below that only 17 files were removed and 4 files added in the specified partition.

%sql
OPTIMIZE taxidb.YellowTaxis WHERE PickupYear = 2022

Output (only relevant portions shown):

+-----------------------------------------------------------------------------------------------------+
| metrics                                                                           |
+-----------------------------------------------------------------------------------------------------+
| {"numFilesAdded": 4, "numFilesRemoved": 17 "filesAdded":{...”totalFiles”: 4, “totalSize”:1020557526 |
+-----------------------------------------------------------------------------------------------------+

OPTIMIZE Considerations

While OPTIMIZE can help improve the speed of queries, there are a few things to consider before running this command on all tables to help ensure its effectiveness:

The OPTIMIZE command is effective for tables, or table partitions, that you write data continuously to and thus contain large amounts of small files.

The compacted file size is decided by the Spark config of maxFileSize (1GB by default).

The OPTIMIZE command is not effective for tables with static data or tables where data is rarely updated because there are few small files to coalesce into larger files.

The OPTIMIZE command can be a resource-intensive operation that takes time to execute. You can incur costs from your cloud provider while running your compute engine to perform the operation. It is important to balance these resource-intensive operations with the ideal query performance for your tables.

ZOrder

While OPTIMIZE aims to consolidate files, Z-Ordering allows you to read the data in those files more efficiently by optimizing your data layout. ZORDER BY is a command that refers to the way that data is arranged in files based on their values. Specifically, this technique clusters and colocates related information in the same set of files to allow for faster data retrieval. This colocality is automatically used by Delta Lake in data-skipping algorithms, which you will learn more about in the next section of this chapter.

Z-ordering can be used to improve the performance of queries that filter on the specified Z-Order columns. Performance is improved because it allows queries to more efficiently locate the relevant rows, and it also allows joins to more efficiently locate rows with matching values. This efficiency can ultimately be attributed to the reduction in the amount of data that needs to be read during queries.

To demonstrate OPTIMIZE combined with Z-ordering, you will reset and clear the optimization we did on the taxidb.YellowTaxis earlier in the chapter by repartitioning the existing table into 1000 smaller files:

# define the path and number of files to repartition

path = “/mnt/datalake/book/chapter05/YellowTaxisDelta”

numberOfFiles = 1000

# read the delta table and repartition it

spark.read.format(“delta”).load(path).repartition(numberOfFiles)\

.write\

.option(“dataChange”, “false”)\

.format(“delta”)\

.mode(“overwrite”)\

.save(path)

To get a baseline for your initial query, use the “ZOrder” script and run the query below:

%sql

—-baseline query
SELECT

COUNT(*) as count,

SUM(total_amount) as total_amount,

PickupDate

FROM

taxidb.tripData

WHERE

PickupDate IN (’2021-01-01', ’2022-01-01')

GROUP BY

PickupDate

This query will give you a general baseline for how long the execution will take when the underlying Delta dataset has many, small files and the data is not organized in any particular order. You can apply ZORDER BY with the OPTIMIZE command to consolidate files and effectively order the data in those files. This will, in turn, significantly decrease the time it takes to fetch the query results since the data is easier to locate. This is generally most effective when used on a high-cardinality column and a column used frequently in query predicates, which means that the column that you apply Z-Ordering to impacts how well the data is retrieved.

%sql
OPTIMIZE taxidb.tripData ZORDER BY PickupDate

Output (only relevant portions shown):

+-----------------------------------------------------------------------------------------------------+
| metrics                                                                           |
+-----------------------------------------------------------------------------------------------------+
| {"numFilesAdded": 2, "numFilesRemoved": 17 "filesAdded":{...”totalFiles”: 3, “totalSize”:178915007..|
|..."zOrderStats":{"strategyName": "minCubeSize(107374182400)", "inputCubeFiles": {"num": 0, "size":  |
| 0}, "inputOtherFiles": {"num": 7, "size": 1943357771}, "inputNumCubes": 0, "mergedFiles": {"num": 7,| |"size": 1943357771}, "numOutputCubes": 1, "mergedNumCubes": null}                    |
+-----------------------------------------------------------------------------------------------------+

Now that you added Z-Ordering, you can see the detailed zOrderStats highlighted in the output, which includes the strategy name, input cube files, and other statistics about the ZORDER BY operation.

When you run the same baseline query that was executed before the OPTIMIZE and ZORDER BY command, you should notice a significant increase in the time it takes to retrieve your query results. The time it takes to retrieve your results will vary depending on your cluster configuration, but I consistently noticed around a 70% decrease in time it took to return my query results due to the optimizations.

In this case, adding Z-Order increased the query engine’s efficiency in reading your data, and OPTIMIZE coalesced the small files into larger ones. This can be difficult to show using large datasets, but Figure 5-2 illustrates how consolidation and ordering occur using the taxidb.YellowTaxis table:

Figure 4-2. 2 - taxidb.tripData Delta files before and after OPTIMIZE and Z-Ordering.

The numbered steps in the figure show:

Query the Delta table called taxiDb.YellowTaxis and count the number of records in the first quarter of the year 2022, or between the dates 2022-01-01 and 2022-12-31.

Your Delta table is comprised of small files containing data with no particular order. When you run the baseline query on these small files, it will scan each file until it can return the query results.

To get better performance during query executions, you run OPTIMIZE and ZORDER BY.

Small files are coalesced into larger ones and the data is sorted by the Z-Order columns in ascending order, which is PickupDate.

Data is read from the first file and returns your query results. Since your data is colocacted and sorted, the query engine does not need to read the second file that contains dates after 2022-03-31, so that file is ignored, or skipped, from read operations.

You can see that before we ran any optimization on the table that the data was organized into smaller files with no order to the data. When running the baseline query, the query engine had to scan all the Delta Lake files to find our query predicate WHERE PickupDate BETWEEN ‘2022-01-01’ AND ‘2022-03-31’. Once you applied OPTIMIZE with ZORDER BY, the data was coalesced into larger files, and the data was sorted by the column PickupDate in ascending order. This allowed the query engine to read the first file based on the query predicate, and ignore, or skip, the second file to gather your results.

ZOrder Considerations

You can specify multiple columns for ZORDER BY as a comma-separated list in the command. However, the effectiveness of the locality drops with each additional column.

%sql
OPTIMIZE taxidb.YellowTaxis ZORDER BY PickupDate, VendorId

Similar to OPTIMIZE, you can apply Z-Ordering to specific subsets of data, such as partitions, rather than applying it to the entire table:

%sql
OPTIMIZE taxidb.YellowTaxis ZORDER BY PickupDate, VendorId
WHERE PickupYear >= 2022

If you expect a column to be commonly used in query predicates and if that column has high cardinality (that is, a large number of distinct values), then use ZORDER BY.

Unlike OPTIMIZE, Z-ordering is not idempotent but aims to be an incremental operation. The time it takes for Z-ordering is not guaranteed to reduce over multiple runs. However, if no new data was added to a partition that was just Z-ordered, another Z-ordering of that partition will not have any effect⁴.

Data Skipping

At the beginning of this chapter, we noted that one of the key characteristics around performance tuning includes skipping non-relevant data to reduce the amount of data that needs to be read. Data skipping is also a key feature of Delta Lake.. If you are running a query, for example, that only needs to process data from the past month, Delta Lake can skip over data that is older than that.

Delta Lake automatically maintains the minimum and maximum value for up to 32 fields for files and stores those values as part of the metadata. Delta Lake uses these minimum and maximum ranges to skip the files that are out of the range of the querying field values. This is called Data Skipping.

You do not need to configure or specify data skipping as this feature is activated whenever applicable in Delta Lake, but the effectiveness greatly depends on the layout of your data. In order to maximize the effectiveness of Data Skipping, data can be consolidated, clustered and colocated using OPTIMIZE and ZORDER BY, as mentioned earlier, so that minimum and maximum ranges are narrow and, ideally, non-overlapping.

Figure 4-3. - taxidb.tripData Delta files before and after OPTIMIZE and ZORDER BY along with how data is retrieved using data skipping.

The numbered steps in Figure 4-3 show:

Query the Delta table called taxiDb.YellowTaxis and count the number of records WHERE PickupDate = ‘2022-06-30’.

Your Delta table is comprised of small files containing data with no particular order

You run OPTIMIZE with ZORDER BY for better performance during query executions.

Small files are coalesced into larger ones and the data is sorted by the Z-Order columns, which is PickupDate.

Data skipping is leveraged since you are looking for the query predicate PickupDate = ‘2022-06-30’. The first file is skipped because Delta Lake knows that the query predicate is not contained in this file since it falls outside the range of the min and max values in the file statistics.

Data is quickly read from the second file because Delta Lake knows to scan this file because the search predicate falls inside the range of the min and max values.

File statistics are one of the key features that enable data skipping. Delta Lake collects the following statistics for each file:

Number of records

Minimum values for each of the first 32 columns

Maximum values for each of the first 32 columns

Number of null values for each of the first 32 columns

Delta Lake collects these statistics on the first 32 columns defined in your table schema. Please note, each field within nested columns (e.g. array) counts as a column. If the columns you are attempting to apply Z-Ordering to are not in the first 32 columns, no statistics are collected for those columns. You can configure statistics collection on certain columns by reordering columns in the schema, or you can increase the number of columns to collect statistics on by using delta.dataSkippingNumIndexedCols.

Figure 4-4. - file statistics collected on the first 32 columns

Figure 4-5 shows that by default only statistics for the first 32 columns are collected on a table. And for the purpose of collecting statistics, each field within a nested column is considered an individual column.

To find examples of file statistics on the taxidb.tripData table, you can look at some of the metadata collected in the last transaction entry:

%sh
# find the last transaction entry and search for "add" to find an added file
# the output will show you the file stats stored in the json transaction entry for the last file added
grep "\"add"\" "$(ls -1rt /dbfs/…/_delta_log/*.json | tail -n1)" | sed -n 1p > /tmp/commit.json | sed -n 1p > /tmp/commit.json
python -m json.tool < /tmp/commit.json

This produces the following output (only relevant portions shown):

stats:"{\"numRecords\":12177114,\"minValues\":{\"VendorID\":1,\"tpep_pickup_datetime\":\"2022-01-01T00:00:08.000Z\"...."maxValues\":{\"VendorID\":6,\"tpep_pickup_datetime\":\"2022-11-01T01:27:35.000Z\"....”nullCount\":{\"VendorID\":0,\"tpep_pickup_datetime\":0

From this output, you can see that the minimum values and maximum values were captured in the last file added in the OPTIMIZE and ZORDER BY command executed against the taxidb.tripData table, along with the null count. Statistics were collected on all columns in your table because it contains fewer than 32. This metadata is collected for every file added during the operation.

If the table contains more than 32 columns, you can also change the number of columns that statistics are collected for using the table property delta.dataSkippingNumIndexedCols:

%sql

ALTER TABLE
  table_name
SET
  TBLPROPERTIES ('delta.dataSkippingNumIndexedCols' = '<value>');

It may not be effective to collect minimum and maximum values on some columns because collecting statistics on long values like strings or binary can be an expensive operation You can either configure the table property delta.dataSkippingNumIndexedCols to avoid columns containing long values or move columns containing long values to a column greater than delta.dataSkippingNumIndexedCols using ALTER TABLE ALTER COLUMN⁵. Chapter 7 discusses updating a table’s schema and changing ordering in more detail.

Partitioning

In an effort to further increase performance on large tables,, Delta Lake partitioning allows you to organize a Delta Lake table by dividing the data into smaller chunks called partitions.

You can create partitions based on the values in one or more columns of the table, with the most common being date, which can speed up queries against the table, as well as data manipulation commands such INSERT, UPDATE, MERGE, and DELETE.

When you partition a table, the underlying dataset is organized into different directories, and sub-directories, for each partition:

Figure 4-5. - The underlying dataset of a Delta table is organized into different directories and subdirectories depending how many columns the table is partitioned on.

The numbered steps in Figure 4-5 show:

A Delta table with no partitions is organized into a single directory.

A Delta table partitioned on a single column has a directory created for each of the partition values.

A Delta table partitioned on multiple columns has a directory created for each partition value, and then subdirectories created for each of the additional columns defined for the partition.

When you can selectively query a partition, rather than scanning all the files in the dataset, Delta Lake quickly scans the appropriate directory(ies), or partitions, to perform your operation which results in faster operations. Delta Lake automatically tracks the sets of partitions present in a table and updates the list as data is added or removed, so there is no need to run ALTER TABLE to account for new partitions.

To create a partitioned table, you can use the PARTITIONED BY clause in the table definition using SQL:

—-create partitioned table using SQL
CREATE TABLE tripData(PickupMonth INTEGER,
                       VendorID   INTEGER,
                       TotalAmount DOUBLE)
         PARTITIONED BY(PickupMonth)

—-use the PARTITION specification to INSERT into a table
INSERT INTO tripData
         PARTITION(PickupMonth= '12') (VendorId, TotalAmount)
         SELECT VendorId, TotalAmount FROM decemberTripData;

-- drop partitions
ALTER TABLE student DROP PARTITION(PickupMonth = '12');

The script⁶ below demonstrates how to write a partitioned Delta table from a parquet file, while also adding a column to partition on:

# import month from sql functions
from pyspark.sql.functions import (month)

# define the source path and destination path
source_path = "/mnt/datalake/book/chapter05/YellowTaxisParquet/2021"
destination_path = "/mnt/datalake/book/chapter05/YellowTaxisPartitionDelta/"

# read the delta table, add a month column to partitions on, and write it using a partition.
# make sure to overwrite the existing schema if the table already exists since we are adding partitions
spark.read.format("parquet")\
.load(source_path)\
.withColumn('PickupMonth', month('tpep_pickup_datetime'))\
.write\
.partitionBy('PickupMonth')\  # define your partition
.format("delta")\
.option("overwriteSchema", "true")\
.mode("overwrite")\
.save(destination_path)

Note

To overwrite the schema or change partitioning on an existing Delta table, set .option("overwriteSchema", "true").

To view the partitions, you can view the directories created in the Delta Dataset:

# import OS module
import os

# list files and directories in directory
print(os.listdir('/dbfs/'+ destination_path))

This produces the following output:

['PickupMonth=1','PickupMonth=10','PickupMonth=11','PickupMonth=12','PickupMonth=2', 'PickupMonth=3','PickupMonth=4','PickupMonth=5','PickupMonth=6','PickupMonth=7', 'PickupMonth=8','PickupMonth=9','_delta_log']

In this output you can see that the Delta dataset not only contains the transaction log directory, but also a directory for each of the values in the partition, in this case months 1 - 12.

Delta Lake also makes it easy to update only specified partitions using replaceWhere, which you learned about in Chapter 3. Let’s assume that you have a business requirement that says when the payment type is 4 in the month of December, it needs to be updated to 5. You can use the following pySpark expression with replaceWhere to achieve that result:

from pyspark.sql.functions import (lit)
spark.read                                                              \
    .format("delta")                                                    \
    .load(destination_path)                                             \
    .where((col("PickupMonth") == '12') & (col("payment_type") == '3')) \
    .withColumn("PaymentType", lit(4))                                  \
    .write                                                              \
    .format("delta")                                                    \
    .option("replaceWhere", "PickupMonth = 12 AND payment_type = 3")    \
    .mode("overwrite")                                                  \
    .save(destination_path)

Note

The Delta table schema cannot be overwritten when using replaceWhere.

In the command above, you will notice that you loaded just a single partition using a WHERE clause. Reading partitions directly is not necessary, but using the WHERE clause enables data skipping, such as spark.read.table("<delta_table_path>").where("PickupMonth = '12').

While using the WHERE clause for reading data can be very effective, you can also use the WHERE clause in combination with performance tuning commands, compaction, OPTIMIZE, and ZORDER BY to perform those operations only on a specified partition(s). This can be especially helpful when you are writing new data to specific partitions (e.g., inserting data for the current month). If the WHERE clause is not used, then the entire table is scanned by default .

For example, you can perform compaction on a single partition:

# read a partition from the delta table and repartition it
spark.read.format("delta")          \
.load(path)                         \
.where((col("PickupMonth") > '10')  \
.repartition(5)                     \
.write                              \
.option("dataChange", "false")      \
.format("delta")                    \
.mode("overwrite")                  \
.save(destination_path)

You can also perform OPTIMIZE and ZORDER BY on specified partitions:

%sql
OPTIMIZE 'destination_path' ZORDER BY PickupDate WHERE PickupMonth >= 10

Partitioning Considerations

Partitions can be very beneficial, especially for very large tables, but there are a few things to consider when partitioning tables:

Select your partition column(s) carefully. If the cardinality of a column will be very high, do not use that column for partitioning. For example, partitioning by a column timestamp that may have 1 million distinct timestamps is a bad partitioning strategy. This is why you added date columns in the examples above–they help serve as appropriate partitioning columns.

The most commonly used partition column is date.

You can partition by a column if you expect data in that partition to be at least 1 GB. Tables with fewer, larger partitions tend to outperform tables with many smaller partitions.

As you learned earlier, ZORDER BY works in tandem with the OPTIMIZE command. However, you cannot combine files, however, across partition boundaries, and so Z-order clustering can only occur within a partition. For unpartitioned tables, files can be combined across the entire table.

You cannot use ZORDER BY on fields used for partitioning.

Columns used for partitioning are always moved to the end of the table unless the partition columns are explicitly defined in the column specification (the name and data type for each column) when creating the table.

Unless you define a Delta Lake table partitioning columns referencing the columns in the column specification are always moved to the end of the table.

Conclusion

In this chapter, you learned about different techniques to store and organize data and the significant effects that it can have on how data is read and retrieved during operations. As the types of data points captured continue to grow, along with the sheer volume of data, tables will continue to get larger and larger. Performance tuning on large datasets has been, and always will be, considered a good strategy and best practice. Understanding the Delta Lake features that enable this will help significantly reduce overhead.

I talked about the small data files problem, the impact that can have on performance, and how it can be solved using compaction strategies, including optimal file consolidation using OPTIMIZE. After you OPTIMIZE a table’s files, you can arrange the values within those files using ZORDER BY, which allows you to leverage data skipping more effectively through file statistics collected. You can take advantage of data skipping one step further and even further reduce the amount of data that needs to be read during operations. You can partition Delta tables and break the data into distinct parts.

Using these Delta Lake features together can reduce the amount of non-relevant data that needs to be read, and improve performance, especially as the number of files in your datasets grows. In the next chapter, you will learn how Delta Lake leverages old data files to allow you to version your data and travel back to a certain point in time.

¹ Git Repo Location: /chapter05/01 - Compaction

² Git Repo Location: /chapter05/00 - Chapter Initialization

³ Git Repo Location: /chapter05/01 - Chapter Initialization

⁴ Delta Lake documentation https://docs.delta.io/latest/optimizations-oss.html

⁵ ALTER TABLE Delta Lake documentation https://docs.delta.io/latest/delta-batch.html#change-column-comment-or-ordering

⁶ Git Repo Location: /chapter05/01 - Compaction

About the Author

Bennie Haelen is a principal architect with Insight Digital Innovation-a Microsoft and Databricks partner. As Principal architect with Insight, Bennie’s primary focus areas are Modern Data Warehousing, Machine learning, AI, and IoT on various commercial cloud platforms. Bennie has overseen many Data + AI projects in different application domains such as health care, the public sector, oil and gas, and financial applications. Bennie has architected and delivered real time streaming Data Lakehouse applications with Databricks, Spark Structured Streaming, Delta Lake, and Microsoft Power BI for various application domains. Bennie brings a wealth of practical experience in implementing secure, enterprise-scale Data Lakehouse-based solutions to support business intelligence, data science and machine learning. Bennie has also been a frequent speaker at Databricks events at Microsoft Technology Centers around the country, and was a speaker at the Data+AI 2021 summit.

Dan Davis is a cloud data architect with a decade of experience delivering analytic insights and business value from data. Using modern tools and technologies, Dan specializes in designing and delivering data platforms, frameworks, and processâ to support data integration and analytics for on-premises, hybrid, and cloud environments on an enterprise scale.