---
title: 'End-to-End Data Engineering Project using Databricks Free Edition | Spark Declarative Pipelines'
source: 'https://youtube.com/watch?v=bIIC44n2Dss'
video_id: 'bIIC44n2Dss'
date: 2026-07-31
duration_sec: 5702
---

# End-to-End Data Engineering Project using Databricks Free Edition | Spark Declarative Pipelines

> Source: [End-to-End Data Engineering Project using Databricks Free Edition | Spark Declarative Pipelines](https://youtube.com/watch?v=bIIC44n2Dss)

## Summary

This video is an end-to-end data engineering tutorial using Databricks Free Edition and LakeFlow Spark Declarative Pipelines (SDP). It walks through building a Medallion architecture pipeline (bronze, silver, gold) with CSV files in S3, applying declarative processing, streaming with Auto Loader, incremental loads, Unity Catalog governance, and Genie AI analytics – all using a free Databricks account.

### Key Points

- **Project Goal and Approach** [00:02] — Build an end-to-end data engineering project using Python, Spark, and Databricks in the transportation domain, using LakeFlow Spark Declarative Pipelines on the free Databricks edition.
- **LakeFlow SDP Benefits** [00:44] — SDP is a declarative framework for batch and streaming pipelines with three benefits: automatic orchestration, declarative processing, and incremental processing.
- **Business Problem at GoodCabs** [01:12] — Regional managers receive late, generic data; Bruce gives an ultimatum to prove the declarative approach delivers faster regional insights.
- **Architecture Overview** [07:11] — CSV files land in S3, Databricks processes them through bronze-silver-gold Medallion layers, Unity Catalog governs access, and Genie powers AI analytics.
- **Free Databricks Setup** [08:21] — Sign up for Databricks free edition without a credit card; use Google or email.
- **Catalog and Schema Creation** [09:15] — Create a 'transportation' catalog and bronze, silver, gold schemas using SQL (CREATE CATALOG IF NOT EXISTS).
- **S3 Bucket and Dataset Setup** [11:44] — Create a globally unique bucket, add datastore/city and /trips folders, and upload the CSV files.
- **Declarative vs Imperative** [16:53] — SQL is declarative (what) while Python loops are imperative (how); SDP lets you focus on what to do.
- **Bronze City Materialized View** [22:38] — A Python function reads S3, adds ingest timestamp and filename metadata, and uses @dp.materialized_view to create transportation.bronze.city.
- **Materialized Views vs Streaming Tables** [28:24] — Use materialized views for dimension tables and streaming tables for fact tables.
- **Bronze Trips with Auto Loader** [51:36] — Use readStream with cloudFiles so only new files are processed; create streaming table transportation.bronze.trips with 356K rows.
- **Silver Trips with Auto CDC** [59:28] — Read bronze using CDF, create staging view with 'expect' validations, then an auto CDC flow keyed on trip_id with SCD type 1 — 50 lines vs 135 imperative lines.
- **Gold Layer SQL Views** [01:11:00] — Join trips, city, and calendar into fact_trips; create city-specific views (e.g., fact_trips_vadodara) for regional managers.
- **Genie AI Analytics** [01:18:04] — Use natural-language queries in Databricks to get insights like highest passenger rating day.
- **Continuous Incremental Processing** [01:19:47] — Switch pipeline to continuous mode; dropping a new file (Dec 27) auto-processes 1.8K records.
- **RBAC Access Governance** [01:25:24] — Use Unity Catalog groups (e.g., vadodara_team) and grant permissions to city views instead of individual users.
- **SDP Benefits Recap** [01:30:54] — Declarative processing reduces code, auto CDC simplifies changes, automatic orchestration removes manual stitching, incremental processing cuts reprocessing.

### Conclusion

The project demonstrates how LakeFlow Spark Declarative Pipelines enable building maintainable, scalable data pipelines with significantly less code, while the free Databricks edition makes it accessible for learners and practitioners to master professional data engineering tools.

## Transcript

end data engineering project using Python, Spark and datab bricks in transportation domain. On this channel we have published bunch of data bricks videos previously where we have covered datab bricks foundations. We also
created end toend project in FMCG domain. The goal of this video is to build this project using lakeflow spark declarative pipelines also known as SDP. SDP is a declarative framework which differs significantly from traditional
procedural way. We will be doing all of these using datab bricks free edition account. Yes, you don't have to even give your credit card details to sign up. This free addition enables students and practitioners to learn professional
data and AI tools for free. As per the official documentation, LakeFlow Spark declarative pipeline is a framework that allows you to create batch and streaming data pipelines using SQL and Python. It offers three primary benefits. Automatic
orchestration, declarative processing, and incremental processing. Now, I understand this may sound like jargon to you, but don't worry, I will explain all these concepts in detail using simple and intuitive language. We will learn
all these theoretical fundamentals while building the project. Okay. So let's straight away jump into the problem statement for our project. Good cap is a fast growing cap service company operating across multiple cities and
regions. As the business scales, regional teams increasingly depend on timely region specific data to run daily operations. During a routine regional review, a serious concern emerges. Regional managers are not receiving data
on time and the dashboards they get are generic and hard to use. Teams are forced to export data and manually rework it for the regions. Bruce Hari responsible for platform outcomes realizes this feedback is coming from
multiple regions. This is no longer a local issue. It is a systematic data platform problem. Bruce calls a review with the data team. He expresses clear dissatisfaction. Data is late, views are inefficient, and innovation has slowed.
From his perspective, stability alone is no longer enough. Tony Sharma explains the current state. The platform relies on procedural Spark pipelines with manual orchestration. While stable, these pipelines are tightly coupled and
slow to adapt to needs. For Peter Pande, this feedback got deeper. This is not just a project failure. It is a matter of his team's credibility and respect. Peter is determined to prove that the data team can innovate and deliver at
scale. Peter proposes a bold shift using data bricks with lakeflow spark declarative pipelines. Instead of manually orchestrating every step, the team will declare data intent and let spark handle optimizing execution plans,
manage dependencies and handle incremental processing at scale. Tony though causes feature and entrusts him with leading the pilot. Bruce remains unconvinced. He gives the team a clear ultimatum. Prove that this new approach
can deliver fast region of insights or the platform direction will change. With the stakes set, the project begins. In this journey, you will join Tony, Bruce, and Peter as they explore how Lake Spark declarative pipelines can transform
regional analytics together. >> Hey Peter, you saw Bruce's email, right? >> Yeah, I saw it. Not exactly the great way to start the day. >> Yeah. I mean the real problem is we can't even convince him technically. We
can't go into the technical depths. It will really bore him. >> Yeah. It will make the problem worse. >> Yeah. So what uh what I want to do is right I want to do something very simple uh and uh I want to give something what
Bruce looks like as a solution which is he wants the regional managers to get the data faster. number one and at the same time it should really look something innovative because what Bruce really thinks is that we are not doing
>> Got it. We can do something that's gamechanging. >> No, no. Uh Peter, let's not spend a lot of time on this. I want this to be rather quicker. So go for something that's a quick win that looks innovative
and at the same time it provides a solution. Okay. >> So more like an experiment. >> Exactly. Right now it feels more like a trust issue than a tech issue. So we need to restore that trust with Bruce by
doing something that reduces the manual work, increases the speed and especially complaining. >> Okay, I think I have something in mind. Instead of redesigning everything, why don't we try declarative approach where
we focus on what to do rather than how to do it. >> Okay. So when you say declarative, did you mean lakeflow spark declarative It's a it's a new feature from data bricks.
>> Yeah, exactly that. We'll do just one focused experiment to see if it helps. >> That sounds good because uh I can see that data bricks is putting a lot of money on on stuff. They are raising uh billions of dollars and I'm pretty sure
they must be doing a good job on this. Yeah, I think uh let's go ahead. I hope Bruce. >> Sounds good. I will start working on it right away. >> Great.
I hope you already got the idea on the problem statement for our project based on the introduction video as well as the stakeholder discussion. But let me add few more details. Good caps is a company similar to Uber and they operate in all
these cities in India. Now city is a dimension table. We are going to use this city dot CSV file in our project. And as you can see they operate in Jaipur, Surat, Kochi etc. And these are the city ids. They also have trips data
and each of these record represents one trip. So for this trip ID uh the fair amount was 337. This was a passenger rating and driver rating. Distance
rating and driver rating. Distance traveled was 25 and so on. Okay. So they regions and this is the trips data. This is our fact table. This is our dimension table. And the management of this company wants to know what is the
average customer rating per city. What are the total number of rides and what is the total revenue. They also want to know the region wise analytics. This is
their main area of interest. And when I say region, each of these cities are regions basically. So let's say there is a region manager or a city manager who is handling let's say city of Surat. They want to know okay how the analytics
look like okay what is my average passenger rating driver rating what is my total revenue in that particular city and so on. Now in terms of technical architecture we will use obviously data bricks with lakeflow spark declarative
bricks with lakeflow spark declarative pipelines and the way it will work is we will provide you CSV files which we are going to uh upload it into Amazon S3. going to uh upload it into Amazon S3. Okay so S3 bucket uh in a real life what
will happen is uh you will have Uber like app on your phone and this app data gets aggregated into some relational database. it can be Oracle or whatever and then from there you will do data extraction and you will store data in
S3. Okay, in our case we will just uh give you CSV files which you can upload give you CSV files which you can upload to S3 and from there we will uh ingest data into data bricks. We are using Medelion architecture where we will
process data through bronze, silver and gold layers. We will be using uh declarative pipelines and you will see the magic of this declarative pipelines as we implement the project. We will use unity catalog for access management and
we will do some analytics using the AI feature. We are not going to worry about building dashboards because that's not the intention behind this project. Now foundations then folks we have another video where we have covered the
foundations. All right. So let's begin the implementation work. Let's go ahead and create an account in datab bricks. All you have to do is click a link down below in the video description and you will come here. This
is datab bricks free edition. They have created this free edition so that learning community can benefit and we truly appreciate that afford. Setting it up is very easy. I'm going to use my Google account. You can use your email
ID as well. So click on continue with Google and I will select my Google account. Say continue and here for personal use get free edition. Click on
continue and start the puzzle. Okay. So I think this is the right answer. Hit submit. After few seconds my account is ready. I'm in datab bricks environment right now. And the first step we are going to
perform here is setting up our catalog and schemas. What we will do is go to workspace. Okay. here and create a directory to store our code and we will
directory to store our code and we will call it let's say project transportation or let's say good caps right whatever and here we are going to create a new notebook that will help us create the
cataloges okay so let me open these cataloges in a new tab and as you can see I have few cataloges workspace system e-commerce and so on and we will
create a catalog for our transportation project. Now I have already created this project setup Python notebook. So instead of writing code from scratch, we can just drag and drop this file here. By the way, you will find all the code
in the video description below. Okay, so let's just import it and and here we will connect to the serverless session. So just click on it and the first line here is taking the catalog name. So let's say control enter. We are creating
a widget here. So let's say if you want to change a name of the catalog, you can change it here. And by using this function, it will uh get that name there. Okay. So let's say when you print this name right here, whatever you have
given in the widget, it will take it. Widgets are useful especially for cases like you want to use a different name for catalog, you want to have same code running for different environments, dev, pro, beta etc. In that case these
widgets are pretty useful. Now let's create a catalog and the way you do that is by running a SQL query create catalog if not exist. Okay, so when you control enter it will create it. All right, this is done. Now if you go here and refresh
you will see this transportation catalog. Let's create schemas underneath. So we're going to have bronze, silver and gold schemas following our Medelian architecture. Let's hit control enter. Schemas are
created. So when you hit refresh icon you will see these schemas. Now obviously these schemas don't have anything in it and sources in these schemas as we move forward in this project.
Next step in the process is to set up the S3 bucket for our data set. As you can see in the architecture diagram, the trip data is generated by mobile app and trip data is generated by mobile app and the data goes to some OLTP database from
the data goes to some OLTP database from where we are running data extraction script and eventually data will land into this S3 bucket. Okay. Now obviously this is data engineering project and we are not focusing too much on OLTP. So
what we are doing is I'm going to give you bunch of CSV files which you and I you bunch of CSV files which you and I both we will upload to this S3 bucket and we will implement the project from this point onwards. Okay. So let's log
this point onwards. Okay. So let's log the S3 account. I hope you all have a S3 account. And here uh you can either type S3 or if you have used it already it will show it on on your homepage. And we are going to create a new bucket. Okay.
So let's call this bucket good caps. That's the name of the company. And the bucket has to be globally unique. So many of you are watching this video. And it's not going to work. It has to be globally unique. So to make it globally
unique, what I'm doing is I am appending some suffix. Okay? So I I'm just saying DP1 double Patel 1. Similarly, you should use your name 1 2 3 right? If you
are Maitra Mishra for example, you can say mm 1. If you are Robert John RJ12, whatever make it unique. Okay, that's a common sense folks. And then let's
create a bucket. Uh everything else is just default. The bucket is now created. just default. The bucket is now created. Here I will create a folder called data store. Okay. So let's create this folder. Create folder. Pretty
straightforward. In this I will create two subfolders for city and trips. Now see if you look at the data set that we have given to you we have city and trips. Two folders. Okay. So let's do that. So create a
Okay. So let's do that. So create a folder call it city. Pretty straightforward. Another folder at the same level. Okay. Another folder at the same level. Okay. And call it trips. Create folder. See,
now you have city and trips, right? So in city you will drop just one file. We have only one file with few cities that we are going to with few cities that we are going to drop it here. Okay. So uh drag and drop.
Yes. So this is the files and folder. You can say upload. You can say upload. Okay. Now see I'm in city and I already Okay. Now see I'm in city and I already see this city. CSV file. Let's go to
trips and in trips we are going to upload all the data from full load. So see we have daily files coming in okay August 1st, second and so on. We'll just
August 1st, second and so on. We'll just select all these files. Okay. And let me select all these files. Okay. And let me click on upload. Select all these files. click on upload. Select all these files. Left mouse click drag and drop. So let's
uh upload all these files and it is some data so it's going to take some time. So let's wait. While my files are being let's wait. While my files are being uploaded in parallel I can establish a
if you have seen my previous datab bricks video we have already done it. Okay. So the process is straightforward. What you do is you click on this menu go What you do is you click on this menu go to catalog and click on external data.
Okay. And here create an external location. location. Okay. Next. And the bucket name is S3 Okay. Next. And the bucket name is S3 uh good cabs DP1. Right? That's my
bucket name. For you it will be different. Generate a new token. Copy this token. Click on launch in quick start and make sure in the same tab you are logged in into your S3. And here datab bricks personal access token.
here datab bricks personal access token. Okay, copy paste that. Okay, I think Okay, copy paste that. Okay, I think this is all same. I acknowledge create this is all same. I acknowledge create stack and it is saying create in
progress. It is going to take some time folks. So, let's wait for a few minutes. After a few minutes, this is complete now. And when I go to now catalog
explorer and click on external data, I will see that connection. Now you click on it. Go to browse. You are seeing this directory data store which has sit in trips. Trips has all these files. This means the connection is set up. Now
datab bricks will be able to pull files from S3. We will be using lakeflow spark declarative pipelines for this project. And for that we need to understand declarative programming. Declarative
programming is a programming paradigm. It is not restricted to data bricks. There are many programming languages and approaches that supports declarative programming. The opposite of declarative is imperative. So let's understand it by
going over a very simple example. Let's say you have this Python code which will filter those users who has age greater than 21. You are running a simple Python
loop and aggregating your results into this result array. Now if you're using SQL the same thing you can do by running this SQL query. Okay. So assume you have database with all the records and you want to filter those who has age greater
than 21. In that case you can write this SQL query. So this SQL query is declarative and this Python code is imperative. In imperative coding you focus on how to do it with great details. I want to filter records
greater than 21 age and I will give all the details. Okay, run this for loop check if condition append result to this uh particular array. Whereas in declarative you focus on what to do then how to do. Okay look at these two
examples you will get an idea. So for our project we will use the declarative this kind of approach where we will focus on what to do rather than how to do it. Now we are going to work on processing
the data in our bronze layer and we will start with this city table. The file is pretty simple here. You have city ID and city name. So what we will do is we have
this data in S3. Correct? See we have it in S3 and we will ingest this in our bronze layer. Now if we were doing imperative programming which is opposite to declarative programming what we would have done is we would go to workspace we
will create a new notebook and we will write a code that will look something write a code that will look something like this where we will have s3 bucket like this where we will have s3 bucket as your source path. Your table name
as your source path. Your table name will be transportation.brity. see transportation which has bronze table. So you'll create which has bronze table. So you'll create a new table in bronze called city and
then you will have a spark data frame which will read the CSV file from this source path which is your S3 bucket. Okay. Uh and we have done this kind of coding before in our datab bricks foundation video. So once again you can
to explain everything in detail because we have covered that. And then the step number two will be to add some metadata columns. Okay, we do it for the purpose
of audit for the purpose of debugging issues. It is important to add this kind of metadata column which is file name and inest date time. And then since this is a bronze layer, we are not doing any transformation etc. We are writing the
transformation etc. We are writing the data as is with metadata columns into df.tright in delta format and target table and target table will be transportation dotbronze
city and then you might set some properties such as enabling change data feed cdf and couple of other parameters. Okay. So we can just copy this code into jupyter notebook run it and it will work. But that will be imperative
programming. Imperative programming tells you how to do things in detail. But here in this project we are using declarative programming. We want to focus on what to do rather than how to
do it. Linkflow Spark declarative pipelines which is SDP makes this thing pipelines which is SDP makes this thing super easy for us. So here I will click on jobs and pipelines. I will show you how to create the declarative pipeline.
You can click on this ETL pipeline. Okay. So let's click on it and let's give it a name. We will call it transportation pipeline. And here we transportation pipeline. And here we will select the workspace. So let's
select the transportation which is your catalog. Then select the schema which is catalog. Then select the schema which is bronze and lakeflow pipeline editor on. bronze and lakeflow pipeline editor on. Okay. So this here through this editor
you can build your pipeline. We are going to start with an empty file. Okay. going to start with an empty file. Okay. So let's click on this and here you will select your project directory which is your project transportation. So let's
hit select. All right. So see I will explain this UI components on the left explain this UI components on the left hand side. What you see is the code that your declarative pipeline will run and by default it will have this
transformations directory which has this default file. Okay. So this file contains nothing right now. So what I'm going to do is here I will add some subfolders. So let's create a folder for bronze.
Okay. So now I have a folder for bronze processing. another one for silver. So the code for bronze, silver and gold processing, I want to keep it in a different folder
different folder and then gold. Okay. And this file we can move it to uh bronze. And let's rename this to city. So let's click on
rename this to city. So let's click on rename call it city do.py. rename call it city do.py. Okay. And here see this is the editor for city.py. So you will write the actual code here and on the right hand
side what you see is the pipeline graph. So as you build your pipeline as you run it either you can run it or dry run it. I will explain what dry run means. You will see this graph being built up. Okay. So for city what we are doing is
let me just write a simple function simple Python function simple Python function where we are uh reading the data from where we are uh reading the data from source path. So what is my source path?
Okay let's see source path is nothing but the path of your S3. Okay and in our but the path of your S3. Okay and in our case in my case it is good caps DP1. In your case, it will be your bucket name/ data store/
city. Right? So see if you copy this and if you say see this is what we have essentially. Okay. So let me just just to confirm and then I have a function function says that load this S3 whatever files are there in this S3 bucket load
files are there in this S3 bucket load them into a data frame and I'm supplying couple of options. So it's a CSV file. Header is true. Infer schema is true. Mode permissive means let's say in my CSV file one of the records is corrupt.
It is not adhering to that CSV format. In that case permissive means just allow In that case permissive means just allow it. The other option is fail fast. Okay. The other option let's say if you do fail fast it will actually crash. The
moment it encounters a bad record it will crash it. But we don't want to do that. We we are being tolerant. Okay. And we want to allow bad record. But what happens to that bad record exactly is specified in this. So you'll have a
column called underscorecorrupt record. And if the record is corrupt, you will take that entire record as a string and put it in this particular column. And the other columns you will keep it as null. Okay. So that is how
you handle bad data. And then you want to add the metadata columns, right? So if you look at our imperative code, we essentially added these two columns. So that's exactly what we are doing here. Okay. Now you will return this data
frame. So essentially you wrote a function which reads data from S3. It adds two metadata column and it returns it. Now you are wondering like okay what's going on? Who calls this function? Well, that's the beauty of
LakeFlow Spark declarative pipelines. It will call this function directly, but it will not call it yet because we don't have the decorator that we need in order to call this. So, let me put this decorator. You might be aware about
Python decorators, right? Uh let's say if you want to time the function, you can say time it something and then you can define time it decorator above it. So, similarly, we are going to use the decorator. Okay. And for that, we will
functions we have used before. But this particular one is something that we will use to define this decorator. So we'll say at DP dot materialized
say at DP dot materialized view and here you are going to define that materialized view. All right. So you will say name is whatever then you will supply other arguments. So let me just uh copy paste it here. So you will
say comment city data row processing. Okay. And the name I'm going to call it transportation.bronze.c city. Essentially we have the catalog
called transportation where we have bronze layer which is a schema and city bronze layer which is a schema and city will be a table in it. Okay. Uh so by will be a table in it. Okay. Uh so by default what SDP will do is it will look
and it will simply execute that function. So when it executes it it will function. So when it executes it it will get the data from S3 into this data frame. It will add these two metadata columns and then this data frame it will
columns and then this data frame it will store it as this table transportation. store it as this table transportation. city it will insert into that table and the parameters it will apply are these okay so it will have change data feed
enabled once again if you don't know what is CDF etc refer to my other datab bricks video the complete crash course on datab bricks uh I have explained all these concepts there okay so I think this is it now you compare this
particular code with this code right it is much shorter I know we are not doing uh too much here. It's a simple code. But when you have a complex code where you are doing lot of things at that time, Leglow Spark
declarative pipelines will save you lot of time. It will uh create a robust data of time. It will uh create a robust data pipeline which will not be prone to uh errors etc. Okay. So let's click on dry
errors etc. Okay. So let's click on dry run and dry run what it does is it will run and dry run what it does is it will just make sure that structure-wise uh logically your pipeline is good. Okay. So let's click on dry run. Right
now it is complete now and it created this kind of graph. So if you click on maximized see this is the pipeline graph. Right now I have only one step which is creating materialized view of
the city. Now the other value for this can be a streaming table. Streaming table is something that we use in our fact data. For our dimension data we will use materialized view. Now materialized view in a traditional
materialized view in a traditional database sense is see a view in a traditional database is just a virtual thing. It will run a query every time you access the data in that view. Uh and on fly it will build that view. But
materialized view is something where it will cache the data to disk. You know it will persist it to the disk. So that is in a traditional database sense. Here it is essentially creating a new delta table called transportation.
City and storing that data. And for dimension table which is city is a dimension table here we will use materialized view. And for fact tables we will use streaming table. Now folks if I go to my catalog right now click on
transportation bronze I don't see a table because we have done only dry run. table because we have done only dry run. Now I would like to actually run it. Okay. So you can run it by clicking on this or you can say run file. So let's
this or you can say run file. So let's click on it and now it will run it. Running means it will now execute this code. Previously in a dry run it was just validating the code. Now it is actually going to S3. Okay. It is adding
these two columns and it is storing it into the this particular table. So see into the this particular table. So see here if you look at it it says output records 10 and you can also view this see city
right and you can view all those records here you can see these two metadata columns okay so that's what this bottom panel is doing so you created this city you can verify it in catalog explorer true when I go to bronze and here I will
true when I go to bronze and here I will see my city. All right. So, as you can see, I have all my records. Hooray. We just created the first uh portion in our
declarative pipeline. As you can clearly see the traditional code would have see the traditional code would have looked bigger and it's you know overall SDP is a better approach because you are just specifying what to do and you don't
bother about how to do it. Of course, transformation is something that is business specific. So, data bricks wouldn't know what kind of transformation you want to apply. Therefore, they provide this uh function
where you can apply your transformation. But other than that, you you just use this uh decorator and your work will be done. After bronze processing, let's do processing in a silver layer. So here I
processing in a silver layer. So here I will create a new file and call it city.py. Okay. Okay. And in this file I will write similar
kind of code. Okay. So I will create a function first of all and that function will simply read the data from bronze table. Okay.
So this is my bronze table and I'm creating a data frame by calling spark creating a data frame by calling spark read dot table and this particular uh read dot table and this particular uh table the bronze table has a column
table the bronze table has a column called inest date time. I will rename that to bronze inest time stamp because the time stamp for silver will be different. So I need to add a separate timestamp column for silver processing.
And by the way, I'm seeing all these errors. So let me import the necessary libraries here. All right. So now this looks good. And you can rename these people have different names of column in different layers. So you can do that
using alias. So here only one column name that I'm changing is ingest date time because if I have inest date time in city silver uh that will be confusing because this one is the inest time stamp for bronze. Okay. And for silver
processing time, I will create a new column called silver processed time column called silver processed time stamp. Okay. And then you will just return DF silver. Once again, I need to add a
DF silver. Once again, I need to add a decorator called materialized view here. And what will be the name by the way? Well, name will be exactly Well, name will be exactly transportation dots silver dot city that
is the name and then you can have comment and some table properties. Okay, comment and some table properties. Okay, I'm enabling change data feed uh quality I'm enabling change data feed uh quality layer and so on. Okay, now let's as
usual let's do a dry run first. And by the way, I want to show you uh if I have any problem what happens. See, let me just controll X. Let me just remove the decorator. Okay. And let me just do a dry run. See this run completed. But I
don't think it will do anything. It will not ingest the data into this particular the new materialized view that I have. Right? So therefore I need that I have. Right? So therefore I need to specify it. Uh here let me just let
let's assume you have some kind of programming error. For example, programming error. For example, I have this error, right? Indentation problem. Uh I mean the IDE the editor itself is
showing you that problem. But when you do dry run, see at that time it will understand the purpose of dry run. It is just to flag any issues. It's just a
mechanism of doing a validation. Okay. So let's fix all the errors. So this looks good. And now let's do a dry run. I think this is running. Okay, I clicked on run pipeline. So this is running directly.
pipeline. So this is running directly. As you can see, it first ran this. Uh we We could have done this run file in which case it will run just this silver processing. But anyways, we ran the entire pipeline. And if you look at it,
what happened is you see this green check mark. Both of it worked. Okay, this one is bronze layer. This one is the silver layer. You can go to catalog and click on silver. And now in the silver layer, you will see this city
silver layer, you will see this city table. See, so you have bronze time stamp and you have silver time stamp. You probably noticed this that it is building this dependency graph. Okay, like this kind of arrow based on the
code that we have written so far. In this code, we are saying that my source table is this bronze city and I'm writing my data to silver. City. So, it will detect all of that and it will
create this particular graph on fly when you run your pipeline. and let's understand some of the theory. So, LegFlow Spark declarative pipeline
is a declarative framework for developing and running batch and streaming data pipelines in SQL and Python. We already used Python for our Python. We already used Python for our city table. Okay. So, we understood the
difference between imperative and declarative programming and LakeFlow SDP supports declarative paradigm. Basically, it's a declarative framework where you can build either batch or a streaming data pipelines using Python or
SQL. Let's go over some of the key concepts. The first one is a flow. So, concepts. The first one is a flow. So, in our case, what we did is we had this transformation logic, right, which we defined using this function and then we
defined using this function and then we had a materialized view which was our had a materialized view which was our transportation. bronze city. So we were reading data from S3 right? So see see this source path S3 is something that we
this source path S3 is something that we referred here that was our source then we uh added these two columns right so that is we can say it's a transformation logic and eventually we wrote data to a materialized view using this particular
annotation. So here the transformation logic is nothing but a flow. Okay. So logic is nothing but a flow. Okay. So this thing right here is your flow. It's a fundamental unit of work that defines how data moves from source to a
destination. Okay. Now there are different types of flows supported in SDP. Materials view. This is something that we already looked at it. Then there is append flow. Auto CDC flow. CDC means
change data capture etc. Materialized view flow is used for batch processing view flow is used for batch processing whereas append and auto CDC is used for stream processing. So this is the example of the materialized view flow.
Uh you will use this particular decorator at dp dot materialized view. for this particular flow where your destination will be a materialized view.
For append flow, it is mainly used for streaming use case uh to uh you know it's a streaming flow for streaming tables and it is insert only. It never
updates existing rows. Now it is best for raw logs, click streams, IoT events etc. you know when you're having too many like logs let's say application
logs it's okay you know even if you are writing duplicates etc so that way you get the speed okay and this append flow is based for all these use cases then is based for all these use cases then there is auto CDC so this flow supports
inserts updates and deletes and it is available only in lakeflow SDP now there is plain Apache Spark declarative pipeline here. This auto CDC flow is not
available. So this is the benefit of using data bricks where you are using lakeflow SDP and you get this support for auto CDC. It will support both slowly changing dimension. Okay, sd means slowly changing dimension type one
and type two. It supports both of them. Materialized view is already something we saw. So when we wrote that city table, if you go to catalog and hover this tool tip you know it will say materialized view. You can also go to
details and check the type. So see here the type is materialized view. So it is essentially a form of unity catalog manage table and it's a badge target. Okay. So it's a basically a managed table supported by Unity catalog and it
can have one or more materialized view flows written into it. So whenever you have materialized view a materialized view flow will write to it and it differs from streaming tables actually in that you always define the flows
implicitly as a part of the view definition itself and that is what we did uh in the case of CT dimension table. Then there is streaming tables. So streaming table is a form of unity catalog manage table that is also a
streaming target for lakeflow sdp. Okay. And it is used for streaming use cases. Uh it can have one or more streaming flows are written into it either append flows are written into it either append or auto CDC. Then there is sync. So sync
is a simple concept. It's nothing but a streaming target for a pipeline and supports delta table. So as a sync you know let's say you have a streaming flow in that your target your destination can be a delta table Apache Kafka topic
Azure event hubs topics custom python data sources etc and all of them you data sources etc and all of them you know are commonly called a sync and it can have one or more streaming flows are written into it. Then there is pipeline.
Data pipeline is a common concept that we all know. Essentially you have raw data and business data as your destination and whatever you do in between is uh can be done through a
pipeline. Okay. So usually what do we do? We have bronze, silver, gold layer transformation. Okay. So this is your pipeline. Data flows through it and it gets enriched. You clean data. You create business ready columns and so on.
This thing, this whole thing is your pipeline. It's a unit of development and pipeline. It's a unit of development and execution in lakeflow spark declarative pipeline. See data pipeline is a common term but in this context uh when you're
talking about lakeflow spark declarative pipeline a pipeline is a unit of development and execution. Now you can run this pipeline on some schedule. So we'll look into it. We'll go to the pipeline and we can schedule it. It can
also run automatically. Let's say when new data arrives, the pipeline is active and data will flow through the pipeline. It can contain one or more flows, streaming tables, materialized views and syncs. Okay, so you can have a pen flow,
auto CDC flow, etc. You can have streaming tables, materialized views, syncs, etc. And this is a nice diagram which captures the essence of all those concepts. See whatever we did for our city dimension table is uh this
particular use case. Okay, the second one. So your batch source was S3 uh table S3 bucket basically city do.t CSV. Then we wrote that that Python function
which was a materialized view flow and the data went into the materialized view which was our city table. Streaming uh use case is something that we'll cover later on. Uh here you can have a pen flow or auto CDC flow and you can uh
have a sync or a streaming table as your stream target. Let us pause for a moment and think about our next step. We have two tables city and trips. City is a dimension table. Trips is a fact table. When
you're doing any analysis, let's say you're building any BI dashboard, etc. Most of the time you need additional dimension table called calendar or a dimension table called calendar or a date table. So let's say if you want to
do analysis on your trips uh aggregated based on quarters or based on let's say weekday versus weekend or let's say monthly etc. In that case this calendar monthly etc. In that case this calendar or a date table will be pretty useful
and we have used this date table in many of our videos on our channel. Okay. If you look at even our past two datab bricks video, you will find that uh we always have calendar or date as one of the dimension table and this table is
something you can generate programmatically. Okay. So let's say if you write this kind of SQL query okay select explode this and this then you get all the days between let's say 1st Jan and 1st February. Right? So
Jan and 1st February. Right? So similarly we will use this uh SQL query in our Python code in our declarative pipeline to generate the dates let's say pipeline to generate the dates let's say between 1st Jan to 31st December okay
the entire year and then we will add additional columns via code we will say okay is this a weekday or weekend we will also extract month quarter and so on okay all of This can be done
on okay all of This can be done programmatically using our Python code. So let's go ahead and in our silver table see this table we are creating as an additional table. So it will go in silver not in bronze because bronze is
all the raw data that you're getting from S3. Okay. So here I will create a new file called calendar.py. calendar.py. Okay. So let's create this. Let's import
some of the important libraries. Okay. And start date and end date is something that we want to receive it via some kind of configuration. So here you can say spark.com.get to get this from configuration. I will
show you how to set up this configuration later on. Okay. And then we can have this function. Okay. So let's define this function calendar. Okay. So it's the same query that we ran here and that will give you all the
dates between start date and end date and it will put it in a data frame. After that you can have bunch of transformation right you can have things such as you can first generate a date key and it is nothing but see you are
creating one column called date and based on that you are deriving all this information. See you created this as date column which is this. Now once you have this kind of date you can create a date key which is the same date
essentially you are removing this hyphen okay so that will be the key for the table then you can have a year column which will be nothing but year extracted from the same date column month quarter
and so on right so you're creating all these columns and you are doing also bunch of other transformation so I'm going to give you this code folks just copy paste we don't need to to write all of it. It's uh pretty straightforward.
We have covered this kind of code in many of our past videos. So here see you are getting day of the month day of week all of that right month name Jan all of that right month name Jan February etc. quarter week is weekend is
week day holiday you're defining some of the holiday right like these are like all Indian holidays and so on and eventually you will get a table which eventually you will get a table which will have all these columns okay so this
is your transformation function but we need to define the annotation as well need to define the annotation as well right so let's add the decorator so this is a materialized view uh and it will store it in transportation dots
silver.cal table. So that view will appear in the silver schema and these are the table properties. Okay, these are the same ones that I think we used are the same ones that I think we used it for the city table. All right, so
it for the city table. All right, so this looks good to me and I would like to do a dry run here. So my dry run failed and the error I'm getting is SQL config start date cannot be found. So see this is the benefit of doing try
run. If there are any logical errors, configuration missing etc. it will point it. Okay. So when I say spark.com.get start date from where exactly it gets that configuration. We have to define that configuration parameter and you can
go to settings and do that. Okay. So click on this cog wheel and scroll down and here you see this configuration. So let's add configuration let's add configuration start date which is 1st Jan 2025
and then end date is going to be 31st December end date is going to be 31st December 2025. Okay. So let's hit save. So now you see in the configuration you see start date and end date. So let's
see start date and end date. So let's close this and let's do dry run again. So this time the dry run finished successfully without any issues. Okay. And you're seeing this calendar uh table here. Now let's run this. So you can run
This will run everything. We don't need to run the entire pipeline. So I'll just click on run file and get my materialized view created for calendar table. After few seconds it completed successfully. You see this green arrow.
By the way, you can click on maximized and see your entire pipeline. Okay. So, we had city bronze processing, city silver processing and we have calendar silver processing. This is independent,
two. So, that's why it is not showing any link between these. Okay. And here if you click on this calendar, it will show you all this. Okay. We can also
check this in our catalog. So, let's go to catalog. Hit refresh and in silver to catalog. Hit refresh and in silver you will see this calendar table.
you have date column then date key year month day of month day of week. See how amazing it is. Now if you want to do your trip analytics let's say for January month what you can do is you can
join your fact table with this date or calendar table and whatever visualization you're using PowerBI click etc. You can you know plot nice visuals which will show you the data based on the time dimension. You can also use
this date table for your AI needs. Okay, it can be used for both AI as well as your BI needs. See here also in catalog view you can see that this table is created and once again this is a materialized view.
Let's now process our fact table. We have one fact table which is called have one fact table which is called trips and we have a trips data per day. So this is the data for 1st August. And the columns we have a strip ID, date,
city ID, passenger type, whether it's new or repeated, distance traveled, passenger rating, and driver rating, etc. I will go here and create a new file in bronze and let's call it trips. Okay, so
trips.py and let's open that file. Okay, so this is the trips.py. pi file and first of all let me import the necessary libraries. Okay, so this is going to be DP1 and I will write a function to read
DP1 and I will write a function to read trips data from S3. And here notice one thing we are using read stream. read stream will uh read the data in a streaming fashion and the benefit of doing this is going to be that whenever
new file is dropped right see in fact data changes often so this is my bucket data changes often so this is my bucket okay and daily a new file gets dropped so once I'm done processing all the files and let's say my pipeline is
running on day-to-day basis okay and let's say next day completes at that time when it drops the new file. Because of this code, it will only process the new file. It will not process all the other existing files. So this is called
an autoloader feature in datab bricks. And the way you get it is by using read And the way you get it is by using read stream function with this cloud files argument. Okay. So this is autoloader. If you want to learn more about
autoloader, you can Google it. You can ask jet gpt um and it will tell you more. But the crux of this feature is that you have let's say 100 files and
you have processed all of them. On the next day when 10 first file 101 file uh gets dropped in that folder through this feature you are only going to process
the new file. Okay. You can ignore the rest of the old files which are processed. schema evolution mode rescue will make sure that if let's say there is a schema changes then this will not fail it will create a new rescue column
so I will show you that column once this is done and then infer column types means based on the data infer the column type whether it's integer string etc and max files per trigger means when you drop the new files let's say if one file
gets dropped it will be processed but let's say if you drop 105 files then 100 file gets processed in one batch like one trigger and the remaining five files will be processed in the other batch. So this is stream processing but underneath
this is stream processing but underneath per trigger it will use this batch of files for doing the processing and then we have this column called distance travel kilometer. Okay. So let me just show you. You see this and this bracket
show you. You see this and this bracket will create issues for us. So therefore we have renamed it to underscorekm. Uh if you want you can uh do a dry run.
Okay let me stop it by mistake I clicked on run pipeline but if you do a dry run you will notice a problem if you don't have this particular line. Okay you can try it out. Okay I tried it. I was getting error and there is a reason I
put this code in. And then we will also have the metadata columns right like the file name the ingest date time and so on. Okay. So this code looks pretty straightforward just to summarize you are doing stream processing and you are
using autoloader feature to process only the new files. Let's say if you're not using autoloader then what you will do is once you process all these files you will create a directory called processed and you have to move those files into
processed directory. So when new file comes in in a parent directory, it can process only the new files. Okay? And that approach is manual. When I was at Bloomberg, we used to use this approach. We were not using data bricks in my job.
So we were manually processing the files and moving it to processed folder. If let's say some of the files get failed, then there will be a failed folder. Okay? So there is process folder, failed folder and there will be files which are
unprocessed. But using autoloader feature you don't have to do all of that. Okay. So it saves you a lot of time from writing all that code. And time from writing all that code. And then here see previously we are using at
dp materialize you right. But here we are going to use dot table. Okay. So this table will create a streaming table and the arguments I will supply are
these. So you are creating uh a streaming table in transportation dotbronze schema. Okay. And the table name is trips. This is the comment. And
these are the arguments. These arguments are similar to what we have seen before. Now you can refer to the documentation if you want to learn more. Okay. So here see this table will create the streaming table.
Streaming read on table. This is the syntax essentially at db.t then you have function name and then readstream.t table or or if you're using autoload or readstream dot format right so this is the syntax that we are using make sure
you are changing this bucket to the bucket that you have now you can do a dry run I don't want to do that I will just uh directly run this file so my run is completed you can see the check mark here and you can see that it inserted
356,000 rows folks we had that much data if you look at this particular bucket. See, we have data for all these months. I think August to December, okay? And each of these CSV file will have some records.
So, if you sum it up all together, you get 356,000 records. And you can see that it is building this pipeline. See these two have dependency, but these are running separately. You can also go here refresh
go to bronze you will see trips now see trips it says streaming table okay so if trips it says streaming table okay so if you go here and if you click in on the details it is streaming table so it is a manage table which is managed by unity
manage table which is managed by unity catalog but the type of this table is streaming table whereas the type of the city table if you see here it is materialized view in the sample data. I
can see all these records see trip ID, date and so on. And you will see this column rescued data. Uh remember I told you about this particular argument.
you about this particular argument. So this rescued data will contain data whenever there is an error. Okay? It will not throw that record. It will show that record here and in the individual column it will show null. Folks, I
recommend that you use AI to understand uh this code as well. So what you can do is copy paste this code in chat GPT and just say hey chat GPT explain this code and it will explain you the purpose of each of these arguments each of these
instructions etc. Using AI to understand the code to write code is going to be very important skill in this new AI era. Okay. So don't get stuck folks. Anytime you are confused about any parameter,
you have AI. AI can explain you really well. Let's move to the next step which is creating the silver processing for trips. So I'll go to silver and create a new file called trips.py.
Okay. So here is the file. Let us import So here is the file. Let us import necessary libraries and then define a function which will read our bronze table. Okay. So I'm reading bronze table
table. Okay. So I'm reading bronze table here. And when you do read stream, it here. And when you do read stream, it will only read the updated records will only read the updated records because change data feed is enabled in
this bronze function. Right? So if you look at the bronze layer processing when look at the bronze layer processing when we created this table we enabled change data processing due to that when I call read stream function on that table it
read stream function on that table it will only read the updated rows because see bronze is just accumulating all the rows you don't want to process the rows which are already previously processed so you should update only or you should
rather process only the updated rows okay and then we probably change names of certain columns. Let's say for our business needs uh I want fair amount to
be called sales amount. See this will happen. Your column names will not be perfect and as you process them through bronze, silver, gold, you may want to bronze, silver, gold, you may want to change the names and you will also take
care of your time stamp. So here silver process time stamp will be the current time stamp and the previous uh time stamp which we had which was inest time stamp is actually bronze inest time stamp. Okay. Now folks we are not doing
many transformations here because the entire purpose of this tutorial is to teach you the lakeflow spark declarative pipelines. So we are focusing more on that. But in real life you can do any number of transformations here. All
right, let's return the silver data frame. Okay, now here what I will do is frame. Okay, now here what I will do is I will create a view. Okay, I will create a view and let's call it a staging view. Okay, so why we are doing
that? Well, please hold on. I will explain. So we are going to use this auto CDC flow and creating staging view which is like a intermediate temporary view where you're doing all the processing is going to help us with uh
many things. So I will explain uh why it is needed but let's create it for now and I will put some additional annotations uh called expect. Okay. So this is a validation basically you are validating that your business date uh
year is greater than 2020. We are processing data more than 2020. If it is processing data more than 2020. If it is less than 2020 then it will not stop the processing but it will log the error and that error we can uh look through the UI
or we can run some SQL query and we can get an idea on the failures. Okay. So if expectation they have these different modes. So we are using expect here which means track the error but it will continue the
processing. If you expect or drop it will track the error but the uh record which has an error it will drop it. If you use expect or fail in that case the moment it encounters the error it will fail the pipeline. So based on the use
case that you have, you can use one or the other. In our case, if there is one record which is invalid, we don't want to stop the processing of the pipeline. So we are using expect and if the error occurs, it will just log it which we can
view through UI or through a SQL query. Okay. So we're checking the driver rating is between 1 to 10, passenger rating is between 1 to 10 because that's a valid range due to whatever error. If it is out of the range, then we will uh
track that error. Okay. So I hope this much is clear. All right. So just to summarize what we are doing is from bronze table we are reading only updated We are just changing names of certain columns. You can do any number of
transformations and that you are storing in a view which is your staging table. in a view which is your staging table. Okay. Now let's create our actual silver table and it's a streaming table. So you will say dp.create create streaming
will say dp.create create streaming table transportation.s.trips. Okay, this is the comment and the properties. So here also we have enabled the change data feed. The comment is clean and validated orders with CDC
observed capability. CDC means change data capture. Change data capture is a important term in data engineering. uh whenever data is changing you want to capture only the rows which are updated either updated or deleted or added.
Okay. So that is basically your CDC. You have created this uh streaming table. Now the next step will be to let your data flow from silver staging to this
table. Okay. And that you can do using the auto CDC flow. So this is how you do it. You'll create auto CDC flow. Your target is your silver table. Source is
this silver staging. Remember this silver staging table and keys is ID. So ID is nothing but a trip ID. See we have just renamed it. So uh what it will do
is in the target table if there is a record with the same trip id it will update it and if there is no record it will insert it. Remember that absert behavior if you have seen my previous datab bricks video
we do the upsert operation which is if the ID is present in the database then update that record override that record if it is not present then insert it. uh if you're writing the code you have to do like you know when match id then
update otherwise insert you have to write all that code but using spark declarative pipeline it becomes very easy see in just one line now you can say okay this is my key and by the way this is an array so you can have
multiple fields also also okay so this is my primary key uh for this if you find a record in the this particular table then update if you don't find it then insert okay it tackles The other issue as well which is out of sequence.
sequence and you want them to be sequenced by this silver processing time stamp. The next argument is slowly changing dimension sedd type one which means if you find a record with the same trip ID then override it. See if I do
sedd type two what it will do is it will insert a new record with start and end date. This is valid for cases like customer. Okay, let's say you have a customer. Okay, let's say you have a customer record and customer changes
their email id or address. In that case, you are not deleting the customer record record and overwrite your address, right? Overwrite customer's address. You will have second record which will say okay customer address is valid from this
date to the other date. Okay. And the previous record will also have start previous record will also have start date and end date. So if you do two it will do the slowly changing dimension type two. Here for trip ID we can just
we are supplying this as one. Now folks let me show you how the procedural code will look like. See here we wrote only these many lines right like 50 lines.
these many lines right like 50 lines. Now let me go to chat GPT and say that below is the code from spark declarative pipeline generate traditional code which is not basically non-declarative code it is also called imperative generate
imperative equivalent. So see you can use declarative programming or imperative up to you but declarative programming will bring the benefit which is it will reduce the size of the code first of all and uh you don't have to do
manual stitching orchestration all of that okay you just specify what you want to do you don't have to worry about how you want to do it so see imperative you want to do it so see imperative equivalent will be this so here you are
creating a table if not exist right you're creating Then you're setting the table properties. Then you read your bronze table. Then you create your silver staging. Okay. You run your validation
and then upsert sedd. See you are doing upsert sd. Okay. Remember this code from other data bricks tutorial. When not match insert when match update. You have to do all of that. Okay. Now see if I copy this code into my notepad
copy this code into my notepad and if I count the lines it's and if I count the lines it's 135 lines whereas here it is only 50 lines and when you write less code using declarative pipeline it is less
errorprone right the probability of making error in a bigger code base is higher usually not all the all the time than probability of you making the error than probability of you making the error in a small code base. Okay, so you now
understand the benefit uh of this declarative pipeline. All right, I think we are done with this. You can do dry run but I'm going to run this directly.
So let's click on run file. Right now it is just preparing a graph. See preparing a new graph. It's going to take couple of minutes and this one is the bronze table. You will see the new pipeline appearing here after this point. You see
now it appeared. So you created this view and you see like the stream is view and you see like the stream is flowing into your streaming table. The view is created and from view view you are streaming your data into your silver
streaming table. See run completed and 356k records were updated. This table we are creating for the first time. But when you do incremental data updates, you will see that it will only observe the
new records. It will not touch the old records. Okay. So let's verify it in our catalog. So I'm just hitting refresh icon here and let's go here. See, now
icon here and let's go here. See, now you are seeing this trips table and you can view the data here also. I think tables. I think this one, right?
So if you click on this here you are seeing the data. See you have trip ID, business date, city ID. You will see all the renamed columns something. We change it to sales amount. See let me just show you this is see
it's changed from this to sales amount. So transformations are working. Okay. And you have bronze in just time stamp as well as silver process time stamp. And you can see this here also see in the catalog view. All right. So we are
done processing our fact table in silver layer. Folks, it is now time to do our gold processing. So far we have city and
calendar as our dimension table. Okay. So these two are in silver layer. These two are dimension table and trips is our fact table. In goal layer we will simply
join these three. Okay. And we will generate a highly denormalized table which will have all the columns from all these tables. Okay. So this is one big
table which will contain all the data. And we have this regional managers who cares about the statistics for their own city. So see we have our business good caps is operating in all these cities. These are cities in India by the way and
we have separate regional manager per city. So let's say for sur the person can be moan okay for indor it can be mira okay okay visaka it can be fatima
mira okay okay visaka it can be fatima and miraa doesn't care uh what's going on in surah the luckno she cares only about indor okay so what we will do is
we'll create city specific view as well okay so here once you have your data from all the cities we are going to generate city specific view. So there will be one view for let's say Indor, there will be another view for let's say
Surat, okay, another view for Vodra and so on. So this way uh the individual so on. So this way uh the individual regional managers can uh hook their BI dashboard with these views. Okay. So there can be a BI dashboard which is
hooked to this view and this BI dashboard will show all the visuals and charts and things like that. Okay. And the manager for indor will be using this
manager for Surat will be using maybe they have different dashboard. Okay. So we can get all that flexibility. So I hope you are getting the idea on what we are going to do in this section which is we will combine all of this to a single
view called trips right gold trips something like that and then we will create individual views per city so let's go to our gold folder and here we
are going to create a file now this time we are going to use SQL by the way we will not use Python okay and let's call this one trips gold. So trips
this one trips gold. So trips gold dossql and you can use AI as well. You can just say generate with AI and you can say okay uh generate this common you can say okay uh generate this common view. I already have that SQL with me.
So I'm just going to copy paste. So see I'm creating a view called gold fact I'm creating a view called gold fact trips where I'm doing a join between these three tables. Okay. So I'm joining trips, city, calendar and pulling
majority of the fields. Okay. So from trips I am pulling ID, business date trips I am pulling ID, business date etc. From city I'm pulling city name. From calendar I'm pulling month, quarter, things like that. So let's do a
dry run. I will click here. Click on a dry run. And let me just maximize this view here. Wonderful. So this is how my pipeline is going to look like. So I'm joining
calendar city trips and I will I will get fact trips. Okay. And this is the previous pipeline if you remember. So city from bronze to city silver. Okay. You see silver in the tool tip. Here in the tool tip you will see bronze. Then
the tool tip you will see bronze. Then here we have trips which is bronze. Then temporary staging view. And then you have trips which is silver. We are joining here. Okay. All right. Now, instead of running this, let me just
create the uh city specific views as well. It's pretty straightforward. So, let's create a new file SQL and this file is for vodra. Okay. So,
and this file is for vodra. Okay. So, trips and see if you know the basics of SQL this should be pretty easy actually. you
will create a view. Okay. So see from flag trips you are creating a view where flag trips you are creating a view where city ID is GJ02. city ID is GJ02. So if you look at this see Vodra city ID
is GJ02. And similarly you can create views for all other cities. Just to save the time I'm going to simply drag and drop those
views. Okay. So I already have those SQL files with me. Okay. So here the way you files with me. Okay. So here the way you import those files is go to gold and click on import and then here we have all the files. I
unselected trips code and trips photo because we have it already. We are going to give you all the code. Check the video description folks. So let's drag and drop this here. Okay. Import
and see all of these things are imported. Now we are ready to run our pipeline. So just click on run pipeline and maximize the view. By the way, if you want to see the execution see you see here it will show you all
the runs. So you can uh click on this as well. Okay. So if you click on this it will open the monitoring page and you will be able to see the details of your pipeline. Okay. So see it started running it. You can see it here as well.
Uh you can see some spinning wheel. See here it's spinning wheel. Uh let's just make it bigger so you can see it finished this uh it it actually finished the entire thing. Okay. Now it automatically
uh detects the dependencies because when I'm creating fact trips coochi I know that I'm using fact trips as my source table. So this graph you know this pipeline graph it will create automatically and and this is how my
entire pipeline looks right now. So let's go to catalog and let's make sure we have the data. Uh-huh. See these are all views by the way. See we have fact trips, Chundigard, everything. See like let's click on
all the cities. Okay. Right now it's showing only Chundigard but this table contains all the data. And if you want to see data for let's say Vodudra, you
click on it, go to sample data and here you see data for only Vodudra. And now the regional managers are going to use these views to build their dashboards or
let's say if they want to build AI solution, they can use this data. Okay. solution, they can use this data. Okay. So our data engineering uh pipeline is So our data engineering uh pipeline is kind of complete and you have the the
good quality data you know good quality massage data that can be used both for BI as well as for AI needs and here by the way I like this feature from data bricks this is genie so let's say I want
to know which day of the week has highest passenger rating right so I can click on it and this is only for vodra by the Okay. So let's say I'm regional manager in Vodudra. I want to know the answer. You can run this query
and it will show me the rating. See Thursday. Okay. On Thursday I'm getting Thursday. Okay. On Thursday I'm getting the highest rating. Okay. So we can ask different questions. I would suggest you ask uh different questions on this,
right? Which month has the highest? You can say okay what is the average driver rating okay and it will generate the query you can run the query and this is
by the way only for vodra so in vodra the average driver rating is 665 now if I go to my trips this is the entire table right so here I can say
what is the average driver rating by city and it will tell me that. See, isn't this amazing? This is so quick. You build your pipeline and right here in the same platform, you can do your analytics using this genie. And this is
called unified analytics folks. Here in a single platform which is data bricks, you are able to do your data engineering, data analysis, data science engineering, data analysis, data science all in one place.
Folks, let's now discuss incremental load. So far what we have done is we load. So far what we have done is we have done a full load where we uploaded have done a full load where we uploaded data all the way till 26th December. See
26th December and you can see that here in S3 also okay and you can verify it by in S3 also okay and you can verify it by the way in your catalog. So let's go here and you can say see in fact trips you
and you can say see in fact trips you can say what is the max date if your server has stopped it will not run your query so I clicked here in this location and I made it up so now it is saying max date is 26 okay so this part looks good
date is 26 okay so this part looks good now here is my pipeline how do you exactly schedule this pipelines There are two ways. You can click on schedule icon and you can schedule it for a specific time. So usually people what
they do is they run it once a day or multiple times a day. And you can say multiple times a day. And you can say okay run this at 11:00 or let's say 11:00 at night. Okay, which is 23. Let's say 11:30. I want to run it at 11:30
where my business is closed every day. And you can use chrome syntax as well. But what we're going to do is not use a fixed schedule. Instead, as soon as the fixed schedule. Instead, as soon as the file is dropped here in my S3, I want
this pipeline to run. So for that, you will click on this cog wheel and you will change this mode. See, right now pipeline mode is triggered, which means when you manually trigger it, it will
run it at that time. But we will make it continuous. So hit continuous save and close it. Now see it is preparing the pipeline for to run it in a continuous mode. And you can check the progress by right clicking on this bar. Okay. And
right clicking on this bar. Okay. And you can go to this page. See. So let's wait until this pipeline is initialized and it is in a continuous running mode. Okay. My pipeline is now running. Okay. You can also go to this page and see it.
See, you see this thing is kind of running now. Let me close this one so that I can see it in a in a maximize mode. Okay. Can view it from here also.
mode. Okay. Can view it from here also. So let's fit to the view and let's close this. Okay. Let's just view it here. Okay. So what you see here is the materialized view. You see they are static so there is no change to it. You
see this activity only in streaming table and that's kind of explains the entire purpose of streaming table which is to detect the change in data and is to detect the change in data and process it. Okay. All right. Now let's
process it. Okay. All right. Now let's go to our S3 and here click on upload and I will upload the file for 27. If you look at that file, see here is the
you look at that file, see here is the file for 27th December, it has around 1.8K records. Okay. So let's drag and drop.
So I'm going to drag and drop the 27 file and hit upload. So it is uploading. Now notice this. Now see this pipeline is running. So it will detect that new file has arrived in S3 and it will process it. You see 1.8 8k records. 1.8K
process it. You see 1.8 8k records. 1.8K records. Now it will come here. Okay. Uh so let's see. See the S3 change. That's why it is processing this as well. Uh because there was an update in uh bucket. See 1.8K. Isn't this amazing
folks? And for the trips, the view of trips uh is static because it's a view. See it's this is a view whereas this is a this is a view whereas this is a streaming table. Okay. Now let's go to
here and let's verify. Okay. So I'm going to say what is my maximum date and run the query. Sometimes see there is a delay. So you may not get uh this right
delay. So you may not get uh this right away. You might have to wait. Um let me just refresh this real quick. Let me ask this question again. this question again. Okay. Reset. What is the max state?
Okay. Reset. What is the max state? run query 27. Isn't this amazing? The pipeline is running continuously and it is uh processing those records. So as an exercise, I want you to drop all these files one by one and just
monitor you know how pipeline is processing it and how it shows the data eventually in your fact table. All right folks, so that's it folks. We are done with this incremental processing. Just to summarize what we did is we changed
the mode of our pipeline to continuous the mode of our pipeline to continuous and we uh dropped the new file in S3. We saw how it is processing it. Okay, just to save the resources you can stop it.
It's important you stop it. If you leave it, it will keep on running. Okay, and you have some free credits. So, it may exhaust your credits. So, I'm going to stop it right now. We have created all these views. The
We have created all these views. The next step will be to enable access of regional managers to the corresponding views. And this access management is important part of data governance. So let's say you have this fact trips vodra
let's say you have this fact trips vodra and let's say the person who is managing and let's say the person who is managing that uh that person's name is let's say Queso. Okay. Okay. So queso is a person who is managing this and you want to
who is managing this and you want to give them an access of this. So what you give them an access of this. So what you will do is you will go here and click on settings. So here you will go to identity and access and click on manage
and here you can add users. So I have this uh different account with this ID called double.codebasics.gmail at Gmail and let's say this person let's
assume he's a queso and queso is the regional manager for Vudodra so now uh what you will do is I think you should what you will do is I think you should click on add new and add that person or
click on add new and add that person or that user ID here okay so see this test that user ID here okay so see this test is added here now you will go to the catalog and click on the corresponding
and click on the corresponding view which is fact trips vodra. Go to view which is fact trips vodra. Go to permissions and here you can grant the access. Right now to make things simple we are granting access to a user.
Usually see in a sophisticated environment most of the companies what environment most of the companies what they will do is they will use rolebased access control RBAC where they will add user to a group. Okay. So let me just I
think let me just show you that. So see here you have uh users. So in user you added this test atgmail.com right? And test three is my current account. Now I
test three is my current account. Now I will create a group and group will be will create a group and group will be let's say word uh team. Okay. Actually let me go here.
Yeah team. that is a group. Okay. And in that group you will add that particular user. So if I click on this I can add members
and in here I can add this particular person. Okay. I can add this particular person. Okay. So now let's go to the catalog and in catalog we will go to
transportation gold vodra go to permissions grant and here vodra go to permissions grant and here you will say vodra team right so instead
you will say vodra team right so instead of granting access to individual you are granting access to a team or to a role So let's say if Queso joins different team let's say they join Mumbai team tomorrow in that case you don't have to
removed from Vodra group they'll be added to Mumbai team but whoever is in Vodra team will always have the access so this is called rolebased access so this is called rolebased access control and in the privileges uh let me
just grant all privileges okay so hit confirm so now see this particular user has this access Okay. Now I will get this email. Okay. Uh to join. So I can
this email. Okay. Uh to join. So I can click on join now. And that way click on join now. And that way uh okay let's hit continue with Google. Click on workspace. Okay. So that way I have joined that particular group. Now
for that user I see transportation uh schema and see I don't see bronze and silver because I have given access to only gold and that to fact trips for. only gold and that to fact trips for. Now I'm able to access it. I can look at
the records and the Vudra regional manager will be able to build dashboards or they'll be able to perform the analytics using this genie you know using this AI. So this access control is an important aspect of data governance
and usually it is done through rolebased access control. I have just shown you you will have teams for different cities and you can enable uh those groups for a corresponding views.
We are done with the project implementation. Just to summarize what we did is we had all our data in S3 using LakeFlow Spark declarative pipelines. We ingested that data into
bronze. Okay. Then from bronze we transformed it into silver and then gold. Okay. So we used this declarative pipelines. We used unity catalog for the access management and then we did some quick analytics using genie. Remember we
used to type natural queries in data bricks and it will transform that into bricks and it will transform that into SQL queries. So that's your AI feature. They they have a separate gen option also. So it's kind of a related AI
features. We didn't build up a dashboard but that was not the purpose of this video. Okay. So we are done with this project and now let's summarize the benefits of LakeFlow Spark declarative pipelines. So the first benefit is
obvious one declarative processing where you focus on what to do rather than how you focus on what to do rather than how to do. As a result you can write much less code. We already saw in the case of trips when we are doing silver
processing using declarative pipeline it was 50 lines of code. If you use normal procedural code, remember we we we talked about this it remember we we we talked about this it was 135 lines whereas the declarative
pipeline code was only 50 lines. See it was only 50 line versus 135. So less code means you can focus on your business logic and you don't have to worry about stitching the workflows etc. Auto CDC auto change data capture API
simplifies CDC processing. Once again when we have this normal procedural imperative code see for CDC you have to do all this logic when not match when match update and so on. Whereas when you are using declarative flow all you do is
specify your key and just create auto CDC flow and that's it. The next point is no need for manual code to handle out of order events. The second benefit is
of order events. The second benefit is automatic orchestration. We saw that as you write declarative code, it will create the pipeline on on its own. We don't have to stitch all these workflows. Okay, so that's a major
benefit and it also gives you efficient retry of transient failures. The third retry of transient failures. The third benefit is incremental processing. Okay, you get incremental processing engine for materialized views and it will
reduce inefficient reprocessing of updates and new records. >> Well, Tony, overall this was a good experiment. The main takeaway is we
wrote much less code compared to our traditional pipelines. >> Okay, that sounds interesting. And instead of managing executions, dependencies and stitching things together, we focused on business logic
using this lakeflow spark declarative pipelines and that made things simpler pipelines and that made things simpler and less errorprone. >> Okay, but uh what is the benefit that you see from the business side? From
business side, it will help the most because the pipeline ran faster, incremental updates worked really well and regional views were ready without and regional views were ready without reprocessing everything.
reprocessing everything. >> Okay. Um, so I'm not sure if I'm uh you see this is fitting in our road map? What I mean to say is like do you see that we can replace all our old pipelines and use this method instead?
>> See, I wouldn't go that far. We just did one experiment and it looked good. But pipeline. >> Okay. Yeah. Yeah. I mean I mean I wasn't expecting you to uh say yes to this anyway. I just wanted to know if there
is a scope for that. But it's it's good because u so these things when it starts it's good that we uh did an early experiment. So now that we can share this to Bruce and then you know uh I hope he will be happy looking at
something what we have done because it has some concrete and tangible outcomes that that we have got and at the same time we have done something uh new and experimental. So I think Bruce will appreciate this. Uh let's let's hear it
>> Okay. I I agree. Let's show it to Bruce and I think it's worth sharing with the team and seeing where it fits going forward. forward. >> Great. Good job Peter.
That's it folks. I hope you enjoyed building this enterprisegrade datab certified, visit our website. We are going to provide a link in the video description. Also share your project on LinkedIn and tag codebasics databicks
myself and Haman. If you have any questions, there is a comment box below. questions, there is a comment box below. Thank you for watching.