Delta Live Tables or DLT is a framework for building reliable and maintainable data processing pipelines. DLT simplify the hard work of building large scale ETL while maintaining table dependencies and data quality.
Here, our DLT multi-hop pipeline is well visualized and we can see our two bronze tables, customers and orders_raw. They are joined together into the silver table orders_cleaned. From which we calculate our gold table daily_customer_books.
DLT pipelines are implemented using Databricks notebooks. On the pipeline details on the right, we can see the path to the notebook containing the DLT tables definitions.
We can simply click here to navigate to the source code.
Let us explore the content of this notebook to better understand the syntax used by Delta Live Tables.
In this SQL notebook, we declare our Delta Live Tables that together implement a simple multi-hop architecture. DLT tables will always be preceded by the LIVE keyword. Here, we start by declaring two tables implementing the bronze layer. These represent our data in its rawest form. The table orders_raw ingest Parquet data incrementally by Auto Loader from our dataset directory. Incremental processing via auto loader required the addition of the STREAMING keyword in that declaration. The cloud_files method enable auto loader to be used natively with SQL. This method takes three parameters the data file source location, the source data format, which is parquet in this case and an array of Reader options. In this case, we declare the schema of our data. Also notice that we add a comment here that would be visible to anyone exploring the data catalog.
Let us run this query and see what will happen.
As we can see, running at a DLT query from here only validate that it is syntactically valid. To define and populate this table, we must create a DLT pipeline. We will see later how to configure and run a new pipeline from this notebook.
The second bronze table is customer that presents json customer data. This table is used below in a join operation to look up customer information.
Next, we declare tables implementing the silver layer. This layer represents a refined copy of data from the bronze layer. At this level, we apply operations like data cleansing and enrichment. Here, we declare our silver table orders_cleaned, which enriches the order's data with customer information. In addition, we implement quality control using constraint keywords. Here, we reject records with null order_id. The Constraint keyword enables DLT to collect metrics on constraint violations.
It provides an optional On Violation clause specifying an action to take on records that violate the constraints. The three modes currently supported by Delta are included in this table.
DROP ROW where we discard records that violate constraints.FAIL UPDATE where the pipeline fail when constraint is violated.OMITTED when omitted, records violating constraint will be included, but violation will be reported in the metrics.Notice also that we need to use the LIVE prefix in order to refer to other DLT tables. And for streaming DTL tables, we need to use the STREAM method.
Lastly, we declare the gold table, in this case the daily number of books per customer in a specific region. Here it is China.
Let us see now how to use this notebook to create a new DLT pipeline. To do so, start by navigating to the Workflows tab on the sidebar.
Select the Delta Live Table tab. And click Create Pipeline.
Fill in a pipeline name. For example, demo bookstore. For notebook libraries use the navigator to locate and select the notebook with the delta tables definition, the one we have just explored. Under configuration, add a new configuration parameter. Set the key to dataset_path,and the value to the location of the bookstore dataset. This parameter is used in the notebook in order to specify the path to our source data files.
In the storage location field enter a path where the pipeline logs and data files will be stored. We will explore this directory later. In the target field, enter a target database name. The pipeline mode specifies how the pipeline will be run. Triggered pipelines run once and then shut down until the next manual or scheduled updates. Continuous pipeline will continuously ingesting new data as it arrives. For this demo, let us keep it Triggered.
A new cluster would be created for our DLT pipeline. For this, let us choose the cluster mode. For example, fixed size. And we set the number of workers to zero to create a single node cluster. Notice below the DBUs estimate provided similar to that provided when configuring interactive clusters. Finally, click to Create. The pipeline has been created.
Let us select Development to run the pipeline in development mode. This mode allows for interactive development by reusing the cluster, compared to creating a new cluster for each run in the production mode. The development mode also disable retries so that we can quickly identify and fix errors.
Let us now click Start. The initial run will take several minutes while the cluster is provisioned.
Our pipeline successfully run.
Below, we see all the events of our running pipeline, either information, warning, or errors. On the right hand side, we see all the pipeline details and also the information related to the cluster. In the middle, we see the execution flow visualized as a Directed Acyclic Graph or DAG. This representing the entities involved in the pipeline and the relationships between them. Click on each entity to view a summary which includes the run status. And other metadata summary, including the comment we set during the table definition in the notebook. We can also see the schema of the table.
If we select the orders_cleaned table, we can notice the results reported in the data quality section. Because this flow has data expectation declared, those metrics are extracted here. As we can see, we have no records violating our constraint.
Let us now come back to our notebook for adding another gold table and see how this change is reflected here. Let us open the notebook of this pipeline by clicking on the link here.
Let us scroll to the end of this notebook. And add a new cell. We will add a new table similar to the previous gold table declaration. But this time, instead of China, we will filter for France. But let us do something different to see what happens if we remove, for example, the LIVE prefix.
If we run this cell the syntax of the query is correct. However, let us see what will happen in our pipeline. Let us now click start again to rerun our pipeline and examine the updated results. As we can see, this generates an error.
Table or view not found, because we missed the LIVE namespace. Let us correct this. Okay, here we add again the LIVE keyword.
And we run the query. Great. The syntax is valid. Let us re-run our pipeline. Let us re-run our pipeline by clicking start. Our pipeline is successfully completed and we can see now our two gold tables.
It is time now to check what happened under the hood in Delta Live Tables. As we said, the events and information we see in this UI are stored in the storage configuration we provided during configuring our pipeline. Let us explore this directory.
For this, let us create a python notebook and copy the storage location.
Let us see the content of our pipeline storage location. As we can see, there are four directories: auto loader, checkpoints, system, and tables. The system directory captures all the events associated with the pipeline. Let us explore the events file in the system directory. Interesting.
These event logs are stored as a delta table. Let us query this table. So all the events we see in the UI are stored in this data table.
Let us also see what we have in the tables directory in the storage location. Interesting.
Here we see our five DLT tables. Let us come back to our pipeline to grab the database name to query these tables. If we click on any table, we can see the metastore information.
Let us copy this information to query this table. Let us write a select query on our table using the metadata store information. Yes, indeed. The table is exists in the metastore.
And we can see here the 123 records exist in this gold table. Great. Let us end up by turning off our job cluster. To do so Navigate to the compute tab in the left side bar. Click on the job clusters tab.
And terminate this pipeline cluster.
In this lecture, we will talk about Change Data Capture. We will understand what is Change Data Capture, and we will learn how this can be processed in Change Data Capture.
Change Data Capture or CDC refers to the process of identifying and capturing changes made to data in the data source, and then delivering those changes to the target. Those changes could be obviously new records to be inserted from the source to the target. Updated records in the source that need to be reflected in the target. And lastly, deleted records in the source that must be deleted in the target.
Changes are logged at the source as events that contain both the data of the records along with metadata information. This metadata indicates whether the specified record was inserted, updated or deleted in addition to a version number or timestamp indicating the order in which changes happened. Here's an example of CDC events need to be applied on our target table. Notice here. France, for example, has two records, so we need to apply the most recent change. Canada need to be deleted, so we don't need to send all the data of the record. Lastly, USA and India are new records need to be inserted.
Here we see the changes applied on our target table. And of course, we don't see the record of Canada as it has been deleted. Such a CDC feed could be received from the source as a data stream or simply in json files, for example.
Delta Live Tables supports CDC feed processing using the Apply Changes Into command. The command is pretty simple.
Apply Changes Into a Delta Live Table. It is the target table into which the changes need to be applied. FROM a CDC feed table specified as a streaming source. KEYS is where we identify the primary key fields. If the key exists in the target table, the record will be updated. If not, it will be inserted. Next, with the Apply As Delete When condition, we specify that records where the operation field is"Delete" should be deleted. Then Sequence By, where do We specify the sequence field for ordering how operation should be applied. Lastly, we indicate the list of fields that should be added to the target table.
Note here the target Delta Live Table need to be already created before executing the Apply Changes Into command. Here are the features of the Apply Changes Into command.
It automatically orders late arriving records using the user-provided sequencing key. This pattern ensures that if any records arrive out of order, down stream result can be properly re-computed to reflect the updates. It also ensures that when records are deleted from a source table, these values are no longer reflected in tables later in the pipeline. The default behavior for insert and update operation is to upsert the CDC events into the target table. That means it updates any rows in the target table that match the specified key Or insert new records when a matching record does not exist in the target table. Optional handling for delete events can be specified with the Apply As Delete When condition. And We can specify one or many fields as the primary key for a table. The Except keyword can be added to specify columns to ignore. And lastly, we can choose whether to store records as slowly changing dimension, type 1 or type 2. Apply Changes Into defaults to creating a type 1 slowly changing dimension table, meaning that each unique key will have at most one record, and that the updates would overwrite the original information. On the other hand, Apply Changes Into has one disadvantage.
Since data is being updated and deleted in the target table, this breaks the append-only requirements for streaming table sources. That means we will no longer be able to use this updated table as a streaming source later in the next layer.
In this demo, we are going to process change data capture feed with Delta live tables. Let us first pull the latest version of our course materials from GitHub. Go to the Repos tab and click on the main branch. And on the right click Pull. confirm. Before we see our DLT pipeline, let us take a look on the CDC data we will process during this demo.
Our CDC data is delivered in json files. Here, we have a simple notebook to land new data files in our source directory.
Let us run this function to load a new file of the CDC feed. We have now a new json file number 2.
Let us take a look on the content of this file. Here is the CDC data of our books table. As we can see, in addition to the books data, we have two operational columns, row_status and row_time. row_status contains Insert, Update, Delete operation indicating whether the specified record was inserted, updated or deleted. Update and insert operations contain valid entries for all fields.
On the other hand, delete operations contain null values for all the fields other than the book_id. While the second column row_time indicating when this change happened. This row_time will be used as sequence key in our CDC data processing. Let us now switch to our DLT pipeline. Here's our DLT pipeline.
Here, in addition to the previous notebook we saw in the previous demo, we are adding a new notebook to this DLT pipeline. We will see later how we can edit an existing pipeline to add additional notebook.
For now, let us take a look on our new notebook.
In this notebook, we start by creating a bronze table to ingest the book CDC feed.