YT YouTube2Text

YouTube2Text → Transcripts → End-to-End Spark Streaming Project | Databricks + Kafka + PostgreSQL + Real-Time Dashboard

End-to-End Spark Streaming Project | Databricks + Kafka + PostgreSQL + Real-Time Dashboard — Transcript

by DataBeli · 41,939 words · 5,864 segments · language en · Watch on YouTube

Open in transcriber Summarize with AI Download SRT/VTT

Full transcript

0:00Hi, I'm Narendra Kumar. I a senior data
0:03architect with 12 years of experience in
0:05IT. I'm a certified datab brick solution
0:08architect champion as well. And in this
0:11video, we are going to build an end
0:13toend streaming data project. We will
0:16mainly build it in databicks free
0:18edition. And to bring in streaming data,
0:21we will set up our own Confluent Kafka
0:24account for free. We will send realtime
0:27data to our topic. Then process that
0:29data in real time and generate uh
0:32immediate email alerts based on the
0:34detected fraud. And apart from this
0:36Kafka streaming source, we will also set
0:39up a file streaming source of our own.
0:42And then we will use a third source as
0:44Postgress SQL that will be a static
0:47source so that we can understand how can
0:49we join streams, how can we join static
0:52data with streaming data. So all those
0:55scenarios we will cover and we will also
0:58build an dashboard that also will
1:00refresh in near real time and apart from
1:04covering all these things hands-on. We
1:06will also cover the theoretical part
1:08also so that our basic is strong. We
1:11will understand all the streaming
1:12concepts the Kafka architecture the
1:15stateful and stateless streaming
1:18processing how we can handle late
1:20arriving data using watermarks. how to
1:22use tumbling and sliding windows and
1:25then do aggregations on the streaming
1:27data. So all these concepts we will
1:30clear and whatever data code and other
1:33files that I will be using for this
1:35project those I will I have uploaded
1:37them on GitHub so that you can also do a
1:40hands-on on your own and be confident
1:42with it. I also provide a link to this
1:45presentation as well. And to help you
1:47further, I have prepared this PDF
1:49document with all the project details
1:52including the details that you can
1:53directly put it in your resume. You can
1:56use this document as a revision document
1:58before going to any of the interview. I
2:01have created four other projects as well
2:03related to batch data processing, AI and
2:06data modeling. I have also created an
2:08interview series and there are many more
2:11videos related to data engineering. So
2:13feel free to check them out and if you
2:15want to stay tuned for more such videos
2:18you shall hit the subscribe button and
2:20if you are already a subscriber then
2:22let's get started.
2:24Let's start by understanding what
2:26exactly is streaming and how it is
2:28different from batch processing. So
2:31streaming it is the continuous injection
2:33processing and anal analysis of realtime
2:36events which enables low latency
2:40analytics and immediate decision making.
2:43What it means is that uh in streaming we
2:45continuously ingest new data. We process
2:48it and analyze it in real time so that
2:51we can make any decision with minimum
2:53latency in subseconds or seconds. And
2:57how it is different from batch
2:58processing is that in batch processing
3:01we collect data over time. We keep
3:04collecting the data maybe for 1 hour or
3:06sometimes even for a day. Then we
3:09schedule the jobs on daily or hourly
3:11basis which process all the data which
3:14is accumulated over that period of time
3:16and then we write it to a storage or we
3:19do whatever we want to do. Whereas in
3:21streaming uh we don't store the data
3:24over period of time. we move it as and
3:28when it is being generated from the data
3:30source till our analytic system. So even
3:34a single event which is being generated
3:36we keep pushing it. We don't wait for
3:39any more events to be gathered so that
3:42the latency is minimum. So the data it
3:45is coming continuously without any delay
3:48and then we process it in real time as
3:51and when it is coming which provide
3:53instant results and insights. So if we
3:56visualize it in terms of timeline. So
3:58let's say if the data is being generated
4:01here each dot represents a data point.
4:04So in batch processing we collect the
4:06data over uh some period of time then we
4:09process that batch wherein in streaming
4:13we process each and every event as and
4:15when it is coming. And the examples of
4:18batch processing are like daily sales
4:20reports, monthly billing, payroll
4:22processing or historical data analysis.
4:24These kind of examples come under batch
4:27processing where in stream processing it
4:29comes like a realtime fraud detection or
4:32live dashboard or if you want to monitor
4:35IoT sensors in real time. If users are
4:38clicking on website we want to monitor
4:41that in real time or stock price
4:43monitoring in like uh real time. So all
4:47these examples are of stream processing.
4:50So batch processing it is based for uh
4:54huge volume of data where latency is
4:56acceptable where in stream processing it
4:59is for low latency and wherever we need
5:01any immediate decision
5:03and in our demo we will be using datab
5:06bricks which supports both batch and
5:08stream processing.
5:11So now as we understand both types of
5:13processing next let's understand a use
5:15case we will pick up that use case and
5:18then using that use case we will build
5:20it step by step and along the way we
5:23will understand each and every concept
5:25of streaming as well as batch
5:27processing.
5:29So the background for this fine guard
5:31use case fine guard is like a financial
5:34guard. So the background for this
5:36company is that it processes millions of
5:39credit card transactions daily which
5:41comes from ATMs, point of sales
5:43terminals and other online payment
5:46gateways. And this bank it receives
5:48continuous fraud intelligence from
5:51internal fraud teams also and external
5:53security partners also. they get some uh
5:56like data in in some form like JSON or
6:00other files which tells like which cards
6:03are marked for fraud and all and then
6:07like the problem is that the delayed
6:09fraud detection this can result in
6:11significant financial loss and poor
6:13customer experience. So the problem they
6:16are facing is that whenever they get
6:19continuous transactions and fraud
6:20related data they are not able to act
6:23immediately and they are not able to
6:25detect the frauds immediately in current
6:27scenario. So this is the background.
6:30Now the core problem is that the
6:34fraudulent transactions those must be
6:36identified within seconds. So that is
6:39why it has to be some processing in real
6:42time. And then fraud indicators they
6:44arrive from multiple sources like live
6:47transactions, some watch lists are
6:49there, some customer master data is
6:52there which all these come from
6:53different sources. So currently there is
6:55no way to combine all the all that data
6:58and the traditional batch processing
7:00this cannot react quickly enough to
7:03prevent fraudulent activities. So if we
7:05do a batch processing here hourly or
7:08daily basis it is not sufficient because
7:10frauds has to be detected immediately
7:12and then decision or the action has to
7:15be taken immediately.
7:17Then the solution approach is this. So
7:20we are going to ingest realtime credit
7:23card transactions from Kafka. So they
7:26get the transactions in Kafka and that
7:30that we are going to ingest using
7:31managed streaming pipelines. Then we are
7:35going to also continuously process fraud
7:37watch list updates that we got get as a
7:40JSON files. This is also streaming data.
7:43We will enrich it with transactions and
7:45also with the customer and merchant
7:47reference data and after that we will
7:50apply streaming joins window-based
7:53analytics and we will detect frauds in
7:55real time to generate realtime alerts
7:58and operational dashboards. So this is
8:00the solution we are going to build in
8:02this video end to end. So as we
8:05understand the overall use case now
8:08let's understand the architecture step
8:10by step how exactly we are going to
8:12build this. So we will start by setting
8:16up a confluent Kafka source of our own.
8:18We will set up the source. We will send
8:21some credit card transactions to it and
8:24then we will receive those transactions
8:26on our side. So we will consume those
8:29transactions from Kafka using spark
8:32streaming. Uh using this we will ingest
8:35that streaming data and then we will
8:38process that data. For processing we
8:40will use lakeflow spark declarative
8:43pipelines. Previously these were also
8:45called as delta live tables. Now they
8:48call it call call this as lakeflow spark
8:50declarative pipelines. So using this we
8:53will process uh the real time data and
8:56along with the the streaming data from
8:59Kafka which come as events we will also
9:02ingest some streaming files. So these
9:05will be in JSON form. These will be the
9:08watch list data that comes which tells
9:11like uh which of the cards has been
9:13marked as fraud and there shall be no
9:16transactions coming from Kafka for
9:18those. So that those streaming files
9:20also we will generate on our own
9:22continuously. Then we will ingest them
9:25in real time using autoloader.
9:28And after ingestion like we will process
9:31them also using same uh declarative
9:34pipelines. And along with these two
9:37streaming sources we will also ingest
9:40the data from Postgress SQL which will
9:42be the customer master data and then
9:45this data we will ingest from lakeflow
9:48connect. So this will be a batch
9:50injection because this customer related
9:53information it is kind of static. It
9:55doesn't get updated like in seconds. So
9:58this data also we will bring in here.
10:01After that when we process the data
10:03using these lakeflow spark declarative
10:06pipelines that time we will build the
10:08standard bronze silver and gold
10:11architecture for realtime data
10:13processing. We will see what we do at
10:15each and every stage as per the industry
10:18standards.
10:19And when we build these layers, we will
10:21be building all these in the delta lake.
10:23We will be building the streaming tables
10:25as delta tables.
10:28And along the way, the processing that
10:30we will do that will include stream
10:33stream join that will be joining this
10:35Kafka stream with the streaming files.
10:38It will also include stream static join
10:41that will be joining this information
10:43with the batch data that we are
10:45ingesting. And we will also detect the
10:48frauds in real time. And once we have
10:52processed all that then on the
10:55consumption side whenever we will detect
10:57a fraud in real time we will send
10:59realtime alerts as emails. So we will
11:02see that immediately how custom emails
11:04will be sent and then we will also build
11:07the dashboard which will refresh every
11:10every minute and it will provide the
11:12updated uh data up to date data for each
11:15and every minute
11:18and once we have built everything which
11:20is running fine end to end then we will
11:23orchestrate it end to end using lakeflow
11:26jobs and we will run it end to end in
11:29real time for an hour or so. so that we
11:32will see the streaming data flowing
11:34throughout the system and for governance
11:37of course we will use unity catalog
11:39which is ready to be used within datab
11:42bricks.
11:43So now as we understand the overall
11:46architecture and what exactly we are
11:48going to build. So let's start building
11:50this. We will start by building our
11:52actual Kafka source. But before we go
11:55ahead and build it, uh first let's
11:58understand the key terminologies in
12:00Kafka and how does it exactly work so
12:03that when we create it, you are able to
12:05relate all the terms and all the items
12:07which are there. So this is the standard
12:10Kafka architecture and this architecture
12:12it is followed by most of the streaming
12:15services as a standard. So on on the
12:19left side we always start with
12:21producers. So producers are like the
12:24data generators wherever the actual data
12:26is being generated in real time. It can
12:29be a web application, mobile
12:31application, any micros service even
12:33from databases and then there can be
12:36many more. So producers they publish the
12:39messages and where do they publish the
12:42messages is with to the Kafka cluster.
12:46So we set up a Kafka cluster where all
12:49the messages coming from the producers
12:52are stored and within the cluster we
12:56have something called as brokers. So
12:58brokers they are the actual Kafka server
13:01that stores the data. So these are like
13:04the actual nodes where the the the
13:07streaming data is stored and uh it is
13:10like uh managed and how it is managed is
13:14in terms of topics. So topic it is like
13:17a logical channel for messages. So if
13:21you have worked on any database so topic
13:23you can relate it to kind of a table. So
13:27on a database we create a table and then
13:29we put some data into it and then read
13:32the data from it. Similarly in terms of
13:35streaming data we have a topic. So we
13:38have a topic and in that topic we send
13:41some data to that particular topic and
13:44then read the data from that topic. So
13:46the name of the topic is very important
13:49and in one cluster there can be multiple
13:52topics like we have multiple tables in
13:54databases. So that is why this logical
13:58understanding of topic is very
14:00important. We can correlate it as a
14:01table but there is no actual table here
14:04because these are all messages. But
14:06topic is the logical channel where we
14:09send and receive the messages. And
14:12because these streaming systems they can
14:15also be a very high uh they can also
14:19receive very high volume of data. So
14:22they also have to support parallel
14:24processing and they have to support
14:26massive volume of data which can be uh
14:29like there shall be incoming and
14:31outgoing of messages in real time. So
14:35for that what they do is to support the
14:38parallelism within a topic there is they
14:41create multiple partitions. So within
14:44one topic we can have multiple
14:46partitions which we configure while
14:49creating the topic which that we will
14:51see when we do that. So within one topic
14:54we have multiple partitions. So whenever
14:57a publish producer is sending message to
15:00a topic it can go to any of the
15:03partition. So if there are like multiple
15:06producers data coming in or even if it
15:09is coming from one the data will be
15:11distributed across different partitions
15:14so that it can be written in parallel
15:16and it can be read in parallel. So to
15:19consume the data from these topic uh
15:22there is something called as consumers.
15:25So consumers are the entities which
15:28consume the data or which read the data
15:31from these topics.
15:34And the consumer these can again be any
15:37analytic system. It can go to any
15:39databases, machine learning or
15:40dashboard. So there it can go for any
15:43consumption. But the actual consumer
15:46technical consumer it can be any program
15:48programmatic consumer as well which can
15:51read the streaming data and then put the
15:53data into any of these actual consumers
15:56and
15:58the the entities like the that read the
16:02data there can be multiple consumers as
16:06well. So each of the consumer those can
16:09be individual consumer as well or if we
16:12want to read the data also in parallel.
16:15So we can create something called as
16:17consumer group. So within a group if we
16:20have multiple consumers then those
16:23consumers will read the data parallelly
16:26from different partitions and then they
16:29can store their output into a single
16:31system. So if multiple consumers they
16:34belong to same consumer group then they
16:38will read a subset of the incoming data
16:40and then write it into the output. If
16:43any consumer is not part of any consumer
16:45group or uh they are part of different
16:48consumer groups then they read like the
16:51data parallelly but they read their own
16:54copy of the data and the numbers that we
16:57see here within a partition these are
17:00called as offset. So whenever let's say
17:04a first message will come from the
17:06producer it will go to let's say the
17:08partition zero it will go into zero
17:10position the next message can go to the
17:14zero position in partition one zero
17:16sorry partition one the next message can
17:19go here and the next message after that
17:21it will go to this first position first
17:24position first position like that so
17:27that is how the data is distributed
17:29across these partitions ideally it is
17:32distributed evenly like I explained but
17:34practically there can be couple of
17:36offsets into one partition then to other
17:39so practically things are little bit
17:41different but overall things are in
17:43parallel so whenever a consumer read the
17:46data from these partitions they read
17:49based on these offsets so first it will
17:52let's say read the offset zero then
17:54offset one offset two offset three like
17:56that and internally it keep track also
18:00up to which offset that it has read the
18:03data so that next time when it is
18:05reading the data again continuously then
18:08it knows up to which point it has
18:10already read the data and from that
18:12point onwards it will start. So this
18:15offset is also very important uh concept
18:19to know so that we know how exactly it
18:22tracks how much data it has read and
18:24from where it is getting started.
18:27So these are all the core concepts of
18:30the Kafka architecture. I hope these are
18:32clear. Just to summarize, producers they
18:36produce the data then they publish their
18:38messages into a topic. A topic is the
18:41logical entity like a table in a
18:43database. And within a topic the data is
18:47parallely stored as partitions and
18:49within partition the data is stored
18:51position wise one event per position.
18:54That position is called as offset and
18:57the actual storage of data within the
18:59Kafka cluster is stored into brokers
19:02which uh which are like the nodes or the
19:05server systems where we connect and read
19:07the data and then to read the data out
19:12whatever is reading the data those
19:14entities are called as consumer.
19:16Consumer can be individual or there can
19:18be group of consumers as well which can
19:21read the subset of the data and then the
19:24data can be sent into any output system
19:27or whatever we want to do.
19:30So now as we understand each and
19:32everything now we will go ahead and set
19:34up the actual Kafka and where we will
19:37set up is in the confluent. So Confluent
19:40it is the name of the company which was
19:43founded by the same people who built
19:46Apache Kafka when they were in LinkedIn.
19:48So in 2014 there were some people in
19:50LinkedIn who actually built Apache Kafka
19:53opensourced it those those same people
19:56they founded this company Conflint and
19:59that is why whenever it comes to
20:01streaming this Confluent Kafka is the
20:04most famous and reliable one. So this is
20:07the one we are going to use.
20:10So to start setting up Confluent Kafka
20:13what we will do is we will simply search
20:16for the term Confluent
20:18and the first page that will come will
20:20be the official Confluent website. So we
20:23will click this try Confluent for free
20:27and once we do that this is our sign up
20:29page. So we can enter our email ids and
20:32all or we can sign in with Google. I
20:35will use this option because I have a
20:37Google account.
20:40Uh then you can select any of the Google
20:43account that you have which you want to
20:45use for the sign up part.
20:49Now it is asking for some permissions
20:51which I will allow.
20:54Now it is asking for some more
20:56information. So company name we can
20:58provide anything. We don't want the
21:00emails. So I will simply click on submit
21:03here.
21:07So after submit this is the page we are
21:09getting. I will simply skip this piece.
21:12Uh this is kind of the survey. So you
21:15can select anything that you want.
21:18So we are exploring it.
21:26So these options these are for kind of
21:28survey purpose. It doesn't matter here.
21:31This says like create your first
21:33cluster. So
21:35uh this cluster name it's not that
21:37important. We can leave it or if you
21:40want to customize it you can do it here.
21:43This region we can change to the nearest
21:45region that is that is there. For me as
21:48I'm in India I will select Delhi but
21:51feel free to choose anything that is
21:52near to you.
21:55After that it is asking for credit card
21:58number but instead of that below we see
22:01a option like do you have a promo code
22:04we will I have a promo code that works
22:07without any credit card details. So we
22:09will click here and you can also use
22:11this same code. I will put this code in
22:14the description of the video.
22:17So this is the promo code which is there
22:20and I'm going to paste it and then click
22:23on save.
22:26So it has been successfully added
22:29and now it says like cluster is already
22:32created.
22:34Now it is saying like do you want to use
22:35sample data or we want to use our own
22:39data. So we will use our own data here.
22:44And now we are getting these options. So
22:47this one I don't exactly remember. Uh
22:49let's try this one if it works. So okay
22:54we don't want to build any client here.
22:57So what we will do is we see here like
22:59we have home environment this was our
23:02cluster. So I will click on this
23:04cluster. Now we are on our cluster page.
23:07And if by any chance you are you forget
23:10this option or you are not able to
23:12figure out you just click on this home
23:14icon here this confluent icon and then
23:17also we shall be able to find our
23:19cluster here. So here we can go to
23:22environments
23:24and here we can see that we have this
23:26default environment with one cluster.
23:29Here we can see that we have one
23:31cluster.
23:33Uh this is our cluster
23:37and if I click this cluster it will take
23:39us to that very same page.
23:43So now as our cluster is ready next
23:46thing we are going to do is we are going
23:48to create a topic in our clusters. So we
23:51will click on this topics then click on
23:54create topic. Here we need to give a
23:57name to the topic and the name that we
24:00are going to give to our topic is credit
24:03card transactions because this is the
24:05data that we will be publishing and then
24:08consuming from here and then this is the
24:11number of partitions like the parallel
24:13parallelism that we were talking about.
24:16So this one we will leave default but it
24:18means that it is going to have six
24:20partitions and this option it says like
24:23in enable infinite retention for this
24:26topic. So this is like by default it is
24:30disabled. So the data it will be stored
24:33for some default period of time but if
24:36you want to store it infinitely then we
24:38can enable it so that the data it is
24:40never deleted once it is published here.
24:44But uh that is not uh advised for
24:48production scenarios. And if we go to
24:51advanced we can even configure what
24:53shall be our retention time. By default
24:55we can see it is 1 week. So uh like that
24:59like the these are some advanced
25:00configurations that we can make but we
25:03don't need to change anything here. This
25:05was just for the knowledge purpose. So
25:07we just need to put our name to the
25:09topic. Rest everything we will leave as
25:12default and then we will click on save
25:14and create.
25:16So now our topic it is successfully
25:19created. That's it. We don't need to do
25:22anything else. This contract and all we
25:24don't need to add it. We will skip it.
25:27So now our topic it is ready to be send
25:30data to this topic and then rece or
25:32receive the data from this topic. For
25:36that what we need to do is we need to
25:39first send some data to this topic and
25:42to send and receive the data we need
25:44connection details obviously right. So
25:47connection details where we can find is
25:49if we go back to this cluster part then
25:53we will need to generate API keys as
25:56well. Uh before API keys we also need
25:58this bootstrap server. So this is the
26:02link to the server so that it can come
26:04come and connect to our cluster
26:08and this link I'm going to copy here and
26:11then paste it to a notepad so that later
26:14on we will use it and we will also need
26:17the API keys. So I will click on API key
26:20and then click on create key.
26:24Uh here we will click on next
26:28and then this is our key and secret. So
26:30both of these we will need. So this is
26:33our key I have copied. So key is this
26:38and then secret or password whatever we
26:41call it is this. So I have pasted here
26:45and then we it says like download and
26:48continue because uh later on these may
26:50not be visible. So it is important to
26:52take a note of these so that we can use
26:55these later on. So it it has been
26:57downloaded here as well. So that's it.
27:01Our keys are ready and these keys like
27:03even I'm not going to hide it but I'm
27:06going to delete my cluster before I
27:08publish the video. So don't spend time
27:10copying mine. So you shall go ahead and
27:13create your own cluster and try out on
27:15your own environment.
27:18So now as we have the all the connection
27:20details next we are going to put some
27:23data to this topic. So for that we I
27:27have created a producer. So let's check
27:29it out. So for that I have created a
27:33Python Kafka producer. So this is the
27:35folder that I will include in the
27:37GitHub. So you will see the folder in
27:40the in the link that I will provide. So
27:44this entire thing we will need to run it
27:46from VS code. So you need to install
27:49like VS studio like sorry Visual Studio
27:52Code which is free. So after
27:55installation we need to simply open it
27:58and then we need to go to file open
28:01folder and we need to open that folder
28:04TFA producer. So we need to select this
28:08particular folder Kafka producer and
28:10click on select folder then it will open
28:14this folder. So I have already opened it
28:16that is why it is showing like this. And
28:19in this folder what we need to do is let
28:22me close all these files first.
28:25Uh there is this environment file. So if
28:28we go over here we will see that there
28:30are options for bootstrap server API key
28:34secret and the topic name. So these
28:36details we need to provide which we had
28:39copied from there. So this is our
28:41bootstrap server. So I'm going to paste
28:44it here. Then this is our API key. So I
28:48will provide the key here. Uh let me
28:51paste it. And this is the secret that we
28:54had brought in. So I will put the secret
28:57here. And topic name I have already
28:59provided. In case if you are using any
29:01other topic then you can provide that
29:04topic. And these other things you can
29:06leave as is. These are some
29:08configuration I made for this code. So
29:10that we can control how many uh
29:13transactions it is generating how much
29:15fraud and total customers and all are
29:17there. These things you can leave as is.
29:20So after putting these details I will
29:23put press control S. So these are saved.
29:27Now we will need to run it. But to run
29:30this environment uh we need to set up a
29:33like virtual environment and then
29:36install these libraries as well.
29:39And all these steps we don't need to do
29:41manually here. So in VS code we can
29:44directly do all these things. So we need
29:47to click this show run commands. Then
29:50this option will come Python create
29:53environment. we will uh click create and
29:57then it says like quick create. So
30:01we will select this option and to use
30:03this option like any of the Python shall
30:06be installed in your laptop. So if
30:08Python is not there that you need to
30:10install first and then come here and do
30:12this. So now it is setting up this
30:15virtual environment. So and now it is
30:18installing the packages as well. So
30:21whenever it is setting up that
30:22environment, so this folder is created
30:24and this is where it will do all this
30:27all the stuff it want to do. So now let
30:30me pause it until this environment is
30:32set up. So now the environment is uh set
30:36up the pop-ups are gone now. So now the
30:40we will be able to produce some data. So
30:43for that I have created these three
30:45files producer normal which will produce
30:48normal transaction data without any
30:51fraud. It will simply send it and then
30:54it will send some fraud transaction data
30:58and then this is another like fraud card
31:01related data. So these things we will
31:03see for now we need to just test if we
31:06we are able to send our data to the
31:08topic or not. For that we will use this
31:11producer normal. py. So we can there are
31:15two ways to run it. We can open the file
31:18and simply click on this run python file
31:21option. So it will automatically run the
31:23file for us.
31:25Uh looks like it didn't start. Let me
31:28try one more time.
31:36So now it is trying to make that
31:38connection and we can see it is saying
31:40the data has been produced. So each and
31:42every line here it is sending some data.
31:45So now I will putt controll C so that I
31:49cancel it because uh some data it has
31:52sent this data we will be able to use it
31:55and later on we will come back send some
31:57more data so that we will see in real
31:59time how we are we are able to read it
32:03but if these statements are coming
32:05properly it means that the data has been
32:07produced and it has been sent to the
32:10topic that we had created there and how
32:13it was able to send it was based based
32:15on this environment details. So on this
32:18server using these credentials it sent
32:21the data to this particular topic. So
32:24now our data is published. It is ready
32:27to be consumed
32:29and we are going to consume the data
32:31using spark streaming. So let's check it
32:34out. So we are going to use uh spark
32:38streaming in datab bricks and datab
32:40bricks it comes with a free edition. So
32:43if you don't have a account in the free
32:44edition simply search for datab bricks
32:47free edition and then you will be able
32:49to uh sign up using any of your email
32:52account for free and then login. It is a
32:55very simple step you can do it. Once you
32:58have done it then we need to go to
33:01login.databicks.com
33:04and then we need to login using our
33:06account. So this is my account which I
33:08have already logged in. So it will take
33:10me to the same one.
33:13So now this is our homepage
33:16and here we need to go to workspace.
33:20So in workspace we will create some
33:21notebook and everything we will whatever
33:24we will create it will go to workspace.
33:26So for that we need to create a folder
33:29first for our project. So we will name
33:32it as fine guard
33:35uh project.
33:40So now our folder is created and we will
33:44click this favorite also because we are
33:47going to work on it for maybe couple of
33:49days. So then once we click this
33:51favorites it will whenever you will go
33:54to workspace then if you click favorites
33:57it will be there. So directly you will
33:59be able to open it. Now we are going to
34:02create a notebook here.
34:07So this is the blank notebook created
34:09for us. And this notebook we are going
34:12to name it as 01 because it is our first
34:15notebook. Then Kafka streaming test
34:21because first we are going to test it
34:23properly. Then we will use it properly
34:25in the spark declarative pipeline.
34:29So to consume the data from that topic,
34:32first thing we need is the connection
34:33details. So for that we need the
34:37bootstrap
34:39uh servers.
34:42So bootstrap servers it is going to be
34:45the same URL which we had copied here.
34:50So this is our URL.
34:52Then we have API key which is our key
35:02and then we have API secret.
35:07So this is our secret
35:16and then finally we have topic. So our
35:20topic it was the credit card
35:22transactions.
35:29So these are all the connection details
35:31that we have. So now using these
35:34connection details we are going to read
35:36some sample data. So we will call it
35:39sample batch.
35:42So first what we are going to do is we
35:44will not read the streaming data
35:45directly but we will read the uh batch
35:49data first from that streaming source so
35:52that first we understand the normal
35:54batch reading and then we will
35:56understand what option is changing when
35:58we convert it to streaming. So for that
36:01we will do spark dot read dot format.
36:05format is going to be Kafka because we
36:08are going to read from Kafka here
36:13and then we are going to provide some
36:15options. So what we will do is we will
36:18put it in a bracket so that we can use
36:21multiple lines here. So this bracket we
36:23will close here. After that we need to
36:26provide dot option and within the option
36:29there will be a key and there will be a
36:31value. So like this we are going to
36:34provide multiple options that are
36:36required to make this connection. So let
36:39me copy paste it multiple times so that
36:41every time I don't have to type
36:44and looks like it was trying to run it.
36:46So for now ignore the error.
36:50So or I can clear the output as well so
36:52that it is not confusing.
36:55So format is Kafka. After that the next
36:58option we are going to provide here it
37:00will be Kafka dot bootstrap
37:06dot servers. So the value part as you
37:10may have guessed already it is going to
37:12be this particular value. So let me run
37:14this cell so that these values are
37:16stored and available. So this this
37:19option it is ready. Now the next option
37:22we are going to provide is subscribe.
37:26And subscribe it is like we we subscribe
37:29to a topic so that whatever is published
37:32to that topic it will be subscribed and
37:35it will be received to us. That is how
37:37it works. So here will be the name of
37:39our topic which is stored here.
37:43So now we have the server details we
37:46have the topic name. Next is pending is
37:50the uh authentication details. how we
37:53provide the key secrets and all these.
37:56So this part is bith tricky not exactly
37:59tricky there is a very standard way of
38:02providing the API keys and secrets here.
38:05So those options I'm going to directly
38:07paste here because remembering them it
38:10is not possible at least for me. So
38:13these are the options we need to provide
38:16and just to remove these red lines let
38:19me just clear it out. So that later on
38:23we will use this dot load also.
38:27So uh let's understand these options. So
38:30this this is all about like providing
38:33the authentication detail. So here we
38:36are saying like which security protocol
38:39we are going to use. So there are
38:41different uh authentication protocols.
38:44So this one is like a simple
38:45authentication
38:47uh security layer. So this is the one we
38:50are going to use. Similarly for you to
38:53use this particular protocol the next
38:56set of options we are providing is like
38:58which mechanism we are using. So here we
39:01are using plane that means the
39:03connection details we will provide it as
39:05plain text and then we also need to
39:09provide something called as J uh JAS
39:12configs and the full form of JAS is Java
39:16authentication and authorization
39:18service. So this is a very specific like
39:21uh authentication mechanism which is
39:24used by Kafka. So we don't need to like
39:27understand each and every point here or
39:30we we can simply put these options and
39:32say that these are required to make the
39:35connection or this is how we make the
39:36connection but I wanted to explain it in
39:39detail. So I explained if you are new
39:42you can just assume that these three
39:43lines are for authentication and then
39:46there is another thing that to provide
39:48these uh JAS config this also has to be
39:52provided in a standard way. So how do we
39:55provide this is uh like this. So this is
39:59a very standard mechanism that we need
40:01to type this exact thing and then we
40:04provide the username as API key this API
40:07key and this is our API secret. There
40:11seem to be some typo. Let me correct it.
40:13So yes this is the secret. So this is
40:16how we provide the API key and secret in
40:19a very standard way and nobody can
40:21remember this. So we just need to paste
40:24this and change our API key and secret
40:27here. Let me run this so that this is
40:29also available.
40:31Uh okay, I didn't run it again after
40:33correction. So let me run again. So now
40:37this is ready. So now our connection it
40:40is ready and we can read the sample data
40:43now. But before we read it, there is one
40:46more option we need to provide. So by
40:48default what will happen is if we run
40:50this now, so from this point onwards it
40:54will start reading the data from Kafka.
40:56But we want to read all the data that is
40:59currently available. So we need to play
41:01around the offset. So for that we need
41:04to provide another option here. And that
41:07option is uh simple like starting
41:10offsets
41:15and the value for these option is like
41:17earliest.
41:20So earliest means like start from very
41:22beginning of the data that is available
41:24in this particular topic. And its
41:26default value is latest. So that
41:29whenever this is starting after that
41:31whatever data is available that data it
41:33will pick up. But as we are reading it
41:35as a batch so we shall provide this
41:38option or we shall provide the exact
41:41offsets from where we want to begin
41:43reading the data. So now all our options
41:46are ready. So let's go ahead and read
41:49our first Kaf data from Kafka. So now
41:52our data frame is created. But this
41:55doesn't mean like connection is
41:56successful because this is a spark
41:58follows d a follows lazy evaluation. So
42:02when we actually read the data that time
42:04we will see. So first let's see how many
42:07records
42:08which records that are available here.
42:11So we will do a simple count sample
42:14batch dot count. It will give us the
42:16count of records which are available.
42:19So here we can see that 61 records are
42:23available into our Kafka topic which
42:26were produced by our producer. So here
42:29it has published 61 records when we ran
42:32it. So those records are available. So
42:35now let's see what exact data is
42:38available. So if we go there and then we
42:41need to what we need to do is we need to
42:44display this data frame to see which
42:46data is available. So we are going to
42:49display this data frame now. So this
42:52this is the data which is coming here.
42:55We can see that this is the key which we
42:57had provided in our producer. It was the
43:00transaction ID but here it is available
43:03as binary. If I hover over here, it will
43:06say that the type of this column is
43:08binary. Later on we will cast it to
43:11string to see the actual value.
43:13Similarly, the value is also available
43:16as binary here. So, this value contains
43:20our actual data. Let me minimize this
43:22left section from here. And apart from
43:26this actual data, so all the data which
43:29is coming from which is being put up or
43:31coming from Kafka is always sent and
43:34received as key value pair. These two
43:36are very standard columns. So we provide
43:39a key and in the value part we provide
43:41the actual data. In the key part we
43:44provide a unique key for each and every
43:46data point
43:48and then we get some standard uh
43:51standard options also standard
43:53additional columns also which come by
43:56default whenever we read data from
43:58Kafka. So those columns are topic which
44:01is the name of our topic then partition.
44:04So if you remember we had six
44:06partitions. So the partition number it
44:09will be from 0 to 5 because it follows
44:11the zero standard and then this is the
44:14offset. Offset what it means is like the
44:17position of the of this particular data
44:21point within that partition. So let's
44:23say from second partition it was read
44:26from the 0 1 2 all these positions.
44:29Similarly from first partition it was
44:32read from these positions. So that is
44:34how these offsets and partitions they
44:37provide exactly from where the data was
44:39being read. And this is the timestamp
44:42column and the timestamp type. It is
44:44like whether the time stamp column is
44:46available or not. This is the actual
44:48timestamp column which is the event
44:51timestamp. So these are the additional
44:53columns that we are getting. So now to
44:56actually use this data we need to
44:59obviously cast these two into uh
45:02strings.
45:04So now to cast this data we will create
45:06another data frame. We can name it as uh
45:10casted or let's say pared
45:13batch.
45:15It is going to be like our sample batch
45:19dot
45:20and then we are going to put a select
45:22statement and within select we will
45:25provide multiple select options here. So
45:28first we will read the column key
45:31column. So this is how we read in pi
45:34path. So call key dot then we use cast
45:39and we want to cast it as string.
45:43So this call this is again a function
45:46which is in there in python. So by
45:49default it is not available. So we will
45:51need to import this. So we directly
45:55don't use import. we use like from so it
45:58is coming here like pispark SQL
46:01functions call so it is automatically
46:03coming using genie I'm just pressing tab
46:06so as we have imported it now it will
46:09work fine so this will cast our key
46:12column to string similarly we need to
46:16cast the value as well so I will simply
46:19oh it is coming automatically so I will
46:21simply press tab so value also it is
46:24casting a string
46:26topic it is not casting. It has that
46:28intelligence that it knows that uh these
46:31are there is no need to cast these. So
46:34for you it may or may not come. If it is
46:37not coming you can type it. But for me
46:39it is coming. So I will simply press
46:41tab. Uh that's all. So now all the
46:44columns has been selected. So and these
46:48two has been casted. So which is the
46:50more important part. So now I will run
46:53it and then I will display this parsed
46:56batch.
47:05So it took couple of seconds and now the
47:07results are here. This time we can see
47:10proper data here. So this is the data
47:12which is being generated from the
47:14producer that I have created here and
47:18this is the data it is passing like this
47:20is the transaction ID, customer ID, card
47:24number, merchant ID and then many other
47:28columns which are there and there is a
47:30transaction timestamp column as well and
47:33out of all these columns the transaction
47:35ID is unique. So that is what we are
47:38using as key column here. So this key
47:40and value are being passed from the
47:43producer that is generating the data and
47:46this is how like when it is converted to
47:48string this is how it looks like and uh
47:52this is the data it is passing the other
47:54columns we were not making any changes
47:57so those are available as it is here
48:01and now let's say if you want to store
48:03this data into a table so what we will
48:06do is we will take our batch and we will
48:09save it as table. So let's say this dot
48:13uh write
48:16dot save as table and then we need to
48:19provide the table name and in the table
48:22name we need to provide a complete name
48:25which is three name space like catalog
48:27name dot schema name dot table name. So
48:30for that first of all let's go ahead and
48:32create our catalog and schema
48:35and for that I will simply duplicate
48:37this tab
48:40and then we will go to this catalog
48:42section here
48:45and from here I will click this plus
48:48icon and then click on create a catalog.
48:51So this catalog we will name it as fine
48:54guard which is the name of our project.
48:59So catalog is created. Now we will view
49:01it. Within this catalog we will create
49:04multiple schemas. So to store our brunch
49:07data we will create branch.
49:11So we will create a schema named bronze.
49:14And then this is where we will create
49:16our bronze tables. And let's go ahead
49:19and create for silver and gold as well.
49:26So this silver and gold we will use them
49:29later on.
49:32Oh sorry silver already we created.
49:35Let's create gold.
49:39So if I open this bronze one and if I
49:42copy this name this it is copied. Now if
49:46we go back here I paste that name. So it
49:49will show catalog dot schema and then
49:52dot we need to give a table name. So
49:55table name let's say it is uh
49:58transactions
50:00batch test because still it is batch
50:04data that we are writing.
50:07So it is complete now and this means our
50:10table it is created now. So let me go to
50:13the tab where we had created the schema.
50:16And if we refresh this from here we can
50:18refresh.
50:22And now we can see our table that is
50:24created. And if you want to check out
50:27some sample data we can click this tab.
50:30We need to select a compute. It will
50:32start and then it will show the data. So
50:35after a few seconds we can see the data.
50:37It is the same data that we saw. So our
50:40table it has been created now and here
50:43in this icon we can see it is a table
50:46and why I'm showing this because we will
50:48see how this looks like when we create a
50:51streaming table also. So if we go back
50:54so this is how like reading the data
50:56from Kafka and then writing it to a
50:59delta table as a as a batch data it
51:02looks like. So it will read entire data
51:04from that Kafka and then it will write
51:07it into a into this table again and if
51:10you run this code again again it will
51:12whatever complete data is available in
51:15that Kafka topic it will read and then
51:18it will write into this table that is
51:20how it will work end to end. Now as we
51:24have understood like how it works for
51:26batch data. Now let's understand how we
51:30can use it for streaming data that is
51:33the main purpose of this video and this
51:36project itself. So I have copied the
51:38code that we had from above to uh create
51:42this batch and in this time instead of
51:45batch or sample we will call it as
51:48streaming data frame or streaming df.
51:53So for that instead of spark dot read we
51:57need to mention spark dot read stream.
52:00So so this becomes a streaming data
52:03frame.
52:04And do you know what other changes are
52:06required to make it as a streaming data
52:08frame? Nothing else. So this single word
52:11it will change it from a batch read to a
52:14streaming read. But this is not the
52:17complete picture. While writing there
52:19are other multiple options we need to
52:21provide. then only it works end to end.
52:24But for the reading part only changing
52:26this from read to read stream it will
52:28create a streaming data frame. So let me
52:31run it. It will simply create a
52:33streaming data frame.
52:35And now as we already know that we need
52:38to transform some of the data and we
52:41need to uh do this transformation here.
52:44So I will copy this and then we will
52:47make the same transformation here. And
52:50before I do that, I want to tell that
52:53this is a streaming data frame. So we
52:55cannot display it. It will throw an
52:58error if we try to display this. So if
53:01you try it out and it fails, uh don't be
53:03worried because uh this is a streaming
53:06data. It is expected to be continuously
53:09reading and then writing. So displaying
53:12uh it becomes problematic. So that is
53:14why uh display on streaming data frame
53:17is not supported.
53:19So how we will come to know whether
53:21things are working? We will write it
53:23into a streaming table and then we will
53:26check out the data there. So this is the
53:29same data frame. So let me change the
53:31name here. So that now it will perform
53:34the same transformation on the uh
53:39on the streaming data frame and then it
53:42will be parsed uh instead of batch.
53:44Let's call this also streaming df.
53:48So now this will become our parsed
53:51streaming DF
53:53and next what we need to do we need to
53:56write this into a streaming table and
53:59the options to write the data as a
54:01streaming table are totally different.
54:03So we will not copy paste the same from
54:06above. We will write on our own. So
54:09first of all we need to provide format
54:12which is delta. By default also it is
54:16delta but it is better to provide that
54:19and then we need to provide output mode.
54:25So output mode for streaming data it
54:27will be append mode because we want to
54:30append the new data as and when it is
54:32coming. In very rare case cases it may
54:35be complete mode also where we want to
54:37replace complete data but for streaming
54:39data like append mode makes more sense.
54:43So this is our append mode
54:46and then uh let me press a new line
54:49here. The next option that we are going
54:52to talk about it will be a checkpoint
54:55location. As you can see it is coming
54:57automatically here. So I will simply
54:59press tab and for now I will remove
55:02these other options as I will explain
55:04them step by step. So checkpoint
55:07location it is a location in our DBFS.
55:11It can be any location which is mounted
55:14or created as a volume. This is the
55:17location where it will store the
55:19progress. So if we run this streaming
55:21data once uh it will read the current
55:24data write it into table. Next time when
55:27it will run again then it need to keep a
55:30track like up to which offsets it has
55:33read. From there onwards it will read
55:35the next time. So to store that progress
55:38it needs a checkpoint location and we
55:42will specify this location. So to
55:44specify this location we need to create
55:47a volume and to create a volume we will
55:50go to our catalog tab. So this is our
55:53catalog. The fine guard is our catalog
55:56and then uh we will create a new schema
55:59here with the name source.
56:05And this source is our new schema where
56:08we will create a volume. So when we will
56:12create this volume, we will name it as
56:15transactions. We can give it any name.
56:17It doesn't matter. And you will notice
56:19that here we are creating a managed
56:21volume. So this volume is managed by
56:23datab bricks itself. And we can also
56:26create a external volume which can point
56:28to a Azure data lakeink storage or S3 or
56:32Google storage anywhere. So the actual
56:34location can be on any storage account
56:36as well. Here we will use managed volume
56:38for simplicity.
56:42And now our volume is ready. And this
56:44volume is like a directory. So in this
56:47in this uh location or volume we will
56:50create a directory or folder that will
56:53be our checkpoint.
56:56So this will be our checkpoint location.
56:59So now I will copy it from here. so that
57:02it will copy this complete path. Then we
57:05will go back to our notebook and we will
57:07replace this path. So now this is where
57:10it will store the checkpoint location
57:14and before we see other options let me
57:16wrap it into brackets so there's so that
57:19these red dots go away and it is not
57:24going away. Looks like I forgot to
57:26specify the right stream.
57:29So uh here while reading we specify like
57:33read stream and while writing we specify
57:36data frame streaming data frame dot
57:38write stream and then dot format and
57:41other options
57:44and then the next option that we are
57:47going to provide it will be
57:50uh dot trigger
57:52and in trigger like there can be
57:54different options.
57:56So uh the option that we are going to
57:59use is available now
58:03equal to true. I will explain other
58:06options also.
58:08So this is the option that we are going
58:10to use. So in terms of trigger apart
58:13from available now
58:16there there is like another trigger also
58:18supported which is uh processing time.
58:23So in the processing time we need to
58:25provide some time as well. So for
58:28example here we specify that 10 seconds.
58:32So what this will do is uh it will keep
58:35running the streaming job and it will
58:37process the data every 10 seconds. So
58:41whatever duration we specify. So after
58:44that many seconds or minute also it will
58:47process the data. So then it will
58:49process the data in micro batches. So
58:52all it will sim it will be similar to
58:54batch processing but because this is
58:56doing every second or every 10 second it
58:59will kind of become a streaming data. So
59:02this is the processing time trigger and
59:05in datab bricks free edition in this
59:07compute which is serverless it will say
59:10that this trigger is not supported on
59:12this compute because this is something
59:14like currently they have not allowed but
59:17if you have like a paid datab bricks uh
59:20then it will allow it. So there that is
59:22why this one we will not use. We will we
59:25will use like a continuous processing
59:27that is allowed when we will go to the
59:30uh declarative pipelines but here as a
59:33spark streaming it is not supported. So
59:35the trigger which is supported is
59:37available now and what this does is uh
59:41instead of like running it continuously
59:44it will run it only once. So available
59:47now equal to true what it means is that
59:50load all the new data that is available
59:53now. So if you run it for the first time
59:55it will load everything that is
59:57available. If you run it for the second
59:59time it will check on this checkpoint
1:00:01location what data it has read
1:00:03previously and that point onwards it
1:00:06will read the new data that is available
1:00:08now and then stop. So that is how this
1:00:12trigger will work.
1:00:14And then the final option will come that
1:00:17is to table where we specify the table
1:00:20name that we want to create and the
1:00:23table that we want to create we it will
1:00:25be in our fine guard only fine guard
1:00:28branch and then transaction uh streaming
1:00:31test. This is the name we can provide.
1:00:33It has suggested automatically because
1:00:36we had created transactions batch test.
1:00:38So this time we are creating streaming
1:00:40test
1:00:42and this it covers all the options and
1:00:45when we are writing streaming data it is
1:00:47better to capture it into a another
1:00:51variable because whenever it is writing
1:00:54any streaming data we get a query id as
1:00:57well. So if I print it
1:01:01uh we can print a message like
1:01:04query ID is this. So let's see if it is
1:01:09providing it automatically with
1:01:13query id
1:01:15and then uh it is providing normally it
1:01:19is available as dot id simply
1:01:24and that's it. So this ID it is
1:01:27sometimes available for debugging and
1:01:29tracking purpose. So because this is
1:01:31available I want it to show otherwise
1:01:33this is completely optional only this
1:01:35much code it will run fine. So this it
1:01:39completes our uh streaming writing code
1:01:42and now the code becomes end to end. So
1:01:45we are reading the streaming data
1:01:48here we are creating a stream we are
1:01:51performing some transformation which is
1:01:53casting operation then we are writing
1:01:56the stream. So now let's run this code
1:01:59and we will see it running end to end.
1:02:04So now it is saying query has been
1:02:06started. It is currently running
1:02:09and this is the query ID I was talking
1:02:11about. So this is this ID is unique for
1:02:15every time it is doing some processing.
1:02:18And now we can see that the processing
1:02:20has been completed. So let's check out
1:02:22our data now.
1:02:33And here we can see the data. It has
1:02:36been written successfully. It has 61
1:02:39rows which are there available in the
1:02:41producer. And if we go to the uh schema
1:02:45here which was the bronze schema. And if
1:02:48I refresh it from here,
1:02:55we can see that our table has been
1:02:57created. And this table also it is not
1:03:00marked as streaming table. Uh why?
1:03:03Because these are marked by datab bricks
1:03:06inbuilt functionalities not by uh spark
1:03:10streaming. We will see the difference.
1:03:11We will use the uh declarative pipeline.
1:03:15But this is where our table has been
1:03:16created.
1:03:18And now let's explore the checkpoint
1:03:20location also so that we understand how
1:03:23the progress has been tracked. So if we
1:03:26go to the volume which we had created
1:03:28the transactions volume this location we
1:03:30had specified for checkpoint and here we
1:03:33can see that it is storing some metadata
1:03:36commits offset source related
1:03:39information. So different different
1:03:40information it is storing and the the
1:03:43one that where we are interested in it
1:03:46is the offset because we want to see how
1:03:48it is storing the offset. So this is a
1:03:50small file. I will simply download it
1:03:54and then if I go to my downloads and if
1:03:57I open this file
1:04:00in a notepad or something then this is
1:04:03where we will see the offsets being
1:04:05stored. So this is our topic name and
1:04:09then these are the offsets. So what it
1:04:11means is like from zero partition it has
1:04:14read 11 records. One partition it has
1:04:17read seven records. Second like
1:04:20partition number 2 3 4 5. So 0 to 5 six
1:04:24partitions and it has read these records
1:04:27like 11 7 6 11 9 and 17. If we sum these
1:04:32up it will come up 61. So that is how it
1:04:36now it knows that from each of the
1:04:38partition how many records it has read
1:04:40and next time whenever it will read the
1:04:42data. So it will read the data this
1:04:45point onwards from each of the partition
1:04:47and then it will know that this is the
1:04:49new data. So now we will try to write
1:04:53some new data and we will see how it is
1:04:55being put up. So I'm back in my VS code
1:04:58and earlier I showed the option to run
1:05:01it from top like if you open and run it
1:05:03from here. The second option I was
1:05:05talking about is from here. So if I
1:05:08press the up key it will show the script
1:05:10which was run previously which is this
1:05:13producer normal and by default it use
1:05:16the complete paths here. And if you
1:05:18don't want to use this simply what we
1:05:20can do is we can say python and then the
1:05:24name of the file. So in our case it is
1:05:26producer fraud not fraud like producer
1:05:31uh normal this is the file we want to
1:05:33run. So like this also we can run it. So
1:05:36we will put some if put a few extra
1:05:39records it is putting. Now I will press
1:05:41Ctrl C so that it is cancelled and we
1:05:45will see if now it is reading these new
1:05:48records only. So now I will go back to
1:05:51my code and then I will run our
1:05:54streaming write again
1:05:58and we are not running the reading part
1:06:00again because spark it runs a DAZ
1:06:03operation. So it will create it will
1:06:06store only the reference and the
1:06:08progress is stored here. So now it shall
1:06:10be able to check based on those
1:06:12checkpoint location and it will read
1:06:14only the new data. So now it is
1:06:17completed. So previously it has it had
1:06:2061 rows. Now let's see how many rows it
1:06:23has.
1:06:25So this time we can see now it has 79
1:06:28rows. So the extra 18 rows which were
1:06:31written here those has been read and
1:06:34those has been appended here. So that is
1:06:37how using this trigger type available.
1:06:41Now it read only the new data and then
1:06:44it wrote into this table. So this is how
1:06:47like the streaming processing looks end
1:06:49to end in terms of spark streaming and
1:06:52this spark streaming it is not specific
1:06:54to data bricks it is like apache spark
1:06:57streaming. So whatever platforms they
1:06:59support Spark there we can use it like
1:07:03uh we have uh Synapse we have MS fabric
1:07:07and even if you use opensource spark
1:07:10there we can use it but as we are using
1:07:13datab bricks and datab bricks they have
1:07:15the people who built spark so on top of
1:07:18this spark streaming they have provided
1:07:21some advanced and additional
1:07:22functionalities that we can use for
1:07:26spark streaming and batch processing.
1:07:28both but in a easier way with some
1:07:31advanced settings advanced options that
1:07:33we get and those things we get when we
1:07:36use the lakeflow
1:07:39lakeflow spark declarative pipeline.
1:07:42This is the name they have given.
1:07:44Previously it used to be called as DT
1:07:47delta live tables but now they call it
1:07:49as lakeflow spark declarative pipeline.
1:07:53So when we will use that we will see
1:07:55that many of these options will be taken
1:07:58care automatically by datab bricks and
1:08:00we will need to write fewer code we will
1:08:02get data quality checks capabilities and
1:08:06we will get it more managed way where we
1:08:08can monitor in a better way. So these
1:08:11are some advantages we will get if we
1:08:13use the uh lake floor spark declarative
1:08:16pipelines. That is why now as you
1:08:20understand the base how opensource or
1:08:22the normal spark streaming works. So now
1:08:25we will use the same thing in lakeflow
1:08:27spark declarative pipeline and see what
1:08:30advantages we will get there and then
1:08:32there we will build it completely and we
1:08:35will build the rest of the project there
1:08:37itself.
1:08:39And before we do that uh we will now we
1:08:42will store these credentials also
1:08:45securely because when we work on
1:08:47production projects we never store
1:08:49credentials like this. So we store these
1:08:52credentials in a secret scope where we
1:08:55store them once we can use them anywhere
1:08:58in a secure way. So first let's go ahead
1:09:01and store these credentials in a secret
1:09:03scope
1:09:05and for that I have created a notebook
1:09:08already because there we need to use
1:09:10secret scope APIs and all and normally
1:09:13these are done by the admin people or
1:09:17senior architect people but I wanted to
1:09:20show the proper use case of credentials
1:09:22so I created this notebook the same can
1:09:25be used by you. So here first we need to
1:09:29provide a name of our secret scope where
1:09:31all our secrets will be stored. The name
1:09:34we are going to give is fine guard dash
1:09:37scope. So I will simply run it and you
1:09:40can use any other name if you want. And
1:09:43then because we will be creating the
1:09:45secret scope and secret using APIs
1:09:48because we creating them through the UI
1:09:50it is not there in the free edition. So
1:09:53we will be using the API and to do that
1:09:55we will need token for that. So this
1:09:58cell it will automatically uh take our
1:10:01current user generate a token and store
1:10:04them into the variable. That is what it
1:10:07will do. And here we are saying that the
1:10:09backend type it is datab bricks. It can
1:10:12be Azure key volt or the other
1:10:14cloud-based services as well. But here
1:10:17we are we will be storing them within
1:10:19datab bricks itself.
1:10:21And after that when we run this
1:10:24particular cell it will create the
1:10:26secret scope. I have already created
1:10:28mine when I did it previously. So
1:10:31probably it will say that it failed to
1:10:33create because it is already there. If
1:10:35you are creating for the first time it
1:10:37will create for you. So now as the
1:10:39secret is created sorry scope is created
1:10:43then we need to create our uh secret
1:10:46inside this secret scope and there we
1:10:49want to store these uh server topic key
1:10:53and secret details. So let me copy them
1:10:56from our other notebook. So these are
1:10:58the details we will be specifying there.
1:11:02So I'm pasting it in into a new cell so
1:11:05that I can copy paste them and replace
1:11:07these older values that I had
1:11:10previously.
1:11:14So yeah, this is our secret. Sorry, that
1:11:18was the key. This is the secret
1:11:24and topic name it is the same only. So I
1:11:26don't need to replace it. But we need to
1:11:28provide it for sure. Now I will delete
1:11:31this cell. So now what it it is doing is
1:11:34it will take these four values and it
1:11:37will create a JSON out of it. In the
1:11:39JSON it will store them as bootstrap
1:11:42server topic API key and API secret and
1:11:46then it will create this as a JSON
1:11:49string.
1:11:51So it is created. So if I try to print
1:11:53this,
1:11:56it will print it as the JSON and that is
1:11:59what we will store it as a secret this
1:12:01JSON
1:12:03and to create a secret we have this API
1:12:06called secrets dot sorry secrets /put
1:12:10and there we are providing our scope
1:12:12name the secret key name this key is the
1:12:16name of the key Kafka connection details
1:12:19I'm providing it as a hardcoded values
1:12:21here and the same key we will use later
1:12:24on to retrieve these secrets and the
1:12:27value part it is this JSON that I showed
1:12:29here so this is the payload we are
1:12:33providing we are making the API whole
1:12:36and then it will create our secret so if
1:12:39I run it will say it is created
1:12:42successfully I had previously also
1:12:44created but it is replaced when I ran it
1:12:47again this for secret part it doesn't
1:12:49throw an error so now It is uh like the
1:12:53secret is created inside the secret
1:12:55scope with these details and it is
1:12:58ready. And if you want to confirm if it
1:13:01is working properly. So this is the
1:13:04command we use like dbuttils do.cses.get
1:13:08that we will use in our code also to get
1:13:10these details and then the name of the
1:13:13secret scope and the name of our key.
1:13:16then it will retrieve the data and that
1:13:19JSON it will parse it and then print it.
1:13:23So it has run successfully and we can
1:13:25see that it is able to retrieve the data
1:13:28that JSON from the secret and then it is
1:13:31able to print that in production
1:13:33projects never print this. I just wanted
1:13:35to print to showcase you that these
1:13:37values are coming properly. So now our
1:13:41secrets has been stored successfully
1:13:43into the secret scope. So from there we
1:13:46will use it and then we have a next cell
1:13:49also where we will put a secret about uh
1:13:52Gmail API key. This we will cover later
1:13:55on when we will come to email
1:13:56notification piece. But for now like
1:13:59only using these secret is enough for
1:14:02us. So as we have stored these secrets
1:14:04next let's understand how we can read
1:14:06these secrets and then use in our code.
1:14:09So for that we will go back to our
1:14:11original notebook. I will uh comment
1:14:16this cell so that it will not be used
1:14:18anymore. I will add a new cell so that
1:14:21when you try you can still use it and
1:14:24then here we will use like Kafka
1:14:28connection JSON. So this one we will be
1:14:32reading it from our secret. So dv utils
1:14:35dot uh secrets dot get and the name of
1:14:40our scope it it was the fine guard dash
1:14:43scope fine
1:14:46guard dash scope this was the name and
1:14:49the name of the key it was uh the kafka
1:14:54connection detail so let me copy paste
1:14:56that so that there is no typo in that
1:14:59one so this is the name of our key
1:15:08and here for the key part also we need
1:15:10to specify that this is a key then it
1:15:13will work. So now it will read that JSON
1:15:17as a string and then we need to parse
1:15:19that JSON.
1:15:24So what we will say is like kafka
1:15:28we will call it as config this will be
1:15:32JSON
1:15:33dot loads and it is automatically coming
1:15:37up. So I will press tab here. I will
1:15:41remove these for now. So now we can say
1:15:45that like the Kafka config is there. And
1:15:48now we will need to read these one by
1:15:51one. And this Kafka config it is kind of
1:15:55a dictionary now. And this dictionary it
1:15:58has all these four values. So this value
1:16:01it will be if I paste this and if I type
1:16:05so it is automatically coming. So if I
1:16:08simply press tab it will say that these
1:16:10values are there including the topic as
1:16:13well. So I will press tab again. So
1:16:16topic part it is kind of hard coding but
1:16:19in our case we know that that is there
1:16:21in the Kafka config. So we will read
1:16:24from there itself.
1:16:27So now it will read all those secrets
1:16:30from our secret scope. So this one we
1:16:33don't use this in this in production
1:16:35scenarios. So let me run this so that it
1:16:38is able to read that scope and it threw
1:16:41an error saying that the secret does not
1:16:45exist. So there shall be some typo. Let
1:16:48me double check again.
1:16:51So it is details not the detail. So
1:16:55looks like I was not able to copy it
1:16:57properly. Let me run it again. So now it
1:17:00is able to read it and these connections
1:17:02has been set. Now with this new API key
1:17:05and secret let me set up this JAS config
1:17:09and then batch part we will not run
1:17:11again. We will directly run our
1:17:13streaming part. So this stream has been
1:17:17created now after I run this with the
1:17:19new details. The transformation it is
1:17:23done now
1:17:25and then we will write it into our
1:17:27table. But before that let's write some
1:17:30data again
1:17:32so that we will know that the new data
1:17:34is being written. So I will press up. It
1:17:36will bring in the same command. Then I
1:17:39will press enter. So it will read. I
1:17:41will put press control C. So only these
1:17:45three records it is writing. And then I
1:17:48will go back. I will uh run our write
1:17:52query now.
1:17:54And it is still running. So in free
1:17:57edition sometimes the resources in
1:17:59serverless are not available that is why
1:18:02sometimes it is stuck it takes longer
1:18:04time. So you can interrupt it and run
1:18:06again or you can let it run for some
1:18:09time and looks like this time the time
1:18:11out happened. So I will run it again
1:18:15and if it is taking very long then we
1:18:17will go to the next step and then we
1:18:19will come back here and then check it
1:18:21out. And looks like this time also it is
1:18:24running for longer. So to save my time
1:18:28of recording uh I will we will continue
1:18:30building with the uh
1:18:34spark declarative pipeline and this
1:18:36piece we will check later on but I think
1:18:38you know you believe me that it will
1:18:40work fine. So to build the declarative
1:18:43pipeline we have different options. So
1:18:46one of them it is that from here we can
1:18:49create it. So if I click here, if I
1:18:51click create and then we need to click
1:18:54this ETL pipeline. So it will create a
1:18:57pipeline in that same folder
1:19:01and when it creates it creates a
1:19:03different type of UI. This UI itself it
1:19:06is the lakeflow spark declarative
1:19:08pipeline UI. So from now onwards like
1:19:11spark declarative pipeline we the short
1:19:13form we use is SDP. So whenever I
1:19:16mention SDP it is like spark declarative
1:19:19pipeline. So for this STP we can see
1:19:23this is the pipeline part where it is
1:19:25showing only this pipeline related
1:19:27table. It has created a sorry this
1:19:29pipeline related files and if you go to
1:19:32all files it will show the main
1:19:34workspace. So these were the two
1:19:36notebooks we were working previously and
1:19:38this is the new folder it has created
1:19:40for this pipeline. So if I open it, we
1:19:43will see the same thing that we see
1:19:45here. But while working on SDP, we will
1:19:48see that this pipeline section is more
1:19:51useful or it is like easy to work with.
1:19:55And then this pipeline we will rename
1:19:57it. We will name it as fine guard
1:20:02streaming.
1:20:05This is the name that we will provide
1:20:08here. And this folder name also we will
1:20:11rename it with the same name. So it says
1:20:14rename root folder. We will name it as
1:20:16fine
1:20:19fine streaming
1:20:28and then it has created this
1:20:30transformations folder but we will use
1:20:33our own folder here. So we will create
1:20:36our folders based on our medallion
1:20:38architecture. So first we will create a
1:20:41folder named branch
1:20:45and then this folder we are not going to
1:20:48use. So this folder we will delete it
1:20:52and looks like it will not let us delete
1:20:54because at least one folder shall be
1:20:57there. Uh this other folder which is
1:20:59there. So by default it will be ignored
1:21:01for execution. So to include this we
1:21:04need to click this include folder as
1:21:06pipeline source code then this folder
1:21:10will be considered and this option now
1:21:12it will allow us to delete it and why we
1:21:15need to do that because this is also a
1:21:17good functionality that if some folder
1:21:20or pipeline we are doing for testing and
1:21:22if we don't want to run we can include
1:21:25or exclude them as pipeline source code.
1:21:28So this is the branch folder. Now we
1:21:31will create our file
1:21:34and this file we will name it as uh
1:21:38transactions
1:21:42branch.
1:21:45So I will remove rest of the name and
1:21:47dot py we need to keep it.
1:21:52So now this is where we will write the
1:21:54code for spark streaming or specifically
1:21:57for SDP and we will see how it is
1:21:59different from the normal spark
1:22:01streaming. So to start with the code we
1:22:04will first need to import something. So
1:22:07from
1:22:09pispark we will import
1:22:13pipelines
1:22:17sp
1:22:19in pipelines there is a e as well. So
1:22:22this is our declarative pipeline and
1:22:24this is a decorator as well. So how do
1:22:28we use a decorator is a the rate dp so
1:22:31that it will know it is the uh a a
1:22:35standard like sdp decorator
1:22:38and then dot table and in this table we
1:22:42need to specify the name of the table
1:22:44that we want to create. So it take
1:22:47different values like name and this name
1:22:50it is the name of our target table that
1:22:53will be created. So our target table we
1:22:56want to create it as fine guard dot
1:22:59branch dot transactions. This time there
1:23:03is no test or something. We want to
1:23:05create it with the name transactions.
1:23:08And then there are some additional
1:23:10properties as well. For example comment
1:23:13is there. So there we can mention like
1:23:16what is the purpose of this
1:23:22and here it is comma not a dot. So in
1:23:26the comment we can mention what is the
1:23:28intention of this particular table or
1:23:30this particular file. So we can say like
1:23:34transactions
1:23:36row stream data ingested from Kafka that
1:23:41is what we will store into this table.
1:23:45Now with this decorator we will create a
1:23:48function like div then we can name the
1:23:51functions as like transactions bronze
1:23:54only.
1:23:58And now after this name of the function
1:24:02uh we will say what it will return and
1:24:04there is a standard that it shall be
1:24:06returning a data frame.
1:24:09So this data frame as of now we have not
1:24:12yet imported. So so this also we need to
1:24:15import this like from pispark.sql
1:24:20we need to import dataf frame and it is
1:24:23SQL dot dataf frame. This is how we
1:24:27import it. Now it is imported and here
1:24:30we need to return a proper data frame
1:24:33which is our streaming data frame. So
1:24:36the structure until here it is very
1:24:38standard for SDP and we will see like
1:24:41what advantages we are getting with this
1:24:44proper structure.
1:24:46So now what we need to do is we need to
1:24:48write our streaming code here and for
1:24:51that we will reuse the code that we had
1:24:53written already. So we will go there and
1:24:57first we will copy this code which will
1:24:59read our secrets
1:25:02uh from there. So I will copy paste that
1:25:04same code here. This import piece I will
1:25:07cut it from here. I will keep it at the
1:25:10top because all the libraries we shall
1:25:12keep it at the top. I will select it.
1:25:15Press tab so that the indentation is
1:25:18properly it has to be with some space.
1:25:21This is the standard Python.
1:25:23And after we have our secrets ready then
1:25:27the next thing we need to do is we need
1:25:29to create this uh JAS config. So this
1:25:32code also I will copy from previous
1:25:35notebook and put it here. And next we
1:25:38will need to write our streaming piece.
1:25:41So as we are back in the notebook let's
1:25:44confirm if our previous code was
1:25:46successful. And looks like it it's still
1:25:49timed out. So either the Kafka is not
1:25:52available or as we were trying it for
1:25:55the first time after putting our
1:25:56secrets. So maybe we may have made some
1:25:59typo as well. So let's confirm that
1:26:02piece first. So let me copy this. I will
1:26:05go to our setup secret notebook and we
1:26:08will try to compare if we are we are
1:26:10using the right thing. So this is
1:26:13starting properly. Oh, looks like here
1:26:16we missed a two. So it was 9092
1:26:20but we kept only 909. So now I have
1:26:23added this two and we will add this
1:26:26secret again so that it can connect
1:26:28again. So earlier it was trying with
1:26:30some different port and it was not able
1:26:33to reach there and uh it is saying JSON
1:26:37not defined because it was imported in
1:26:39earlier cell. So let me run these above
1:26:42cells again and secret scope it was
1:26:45already created only secret we will need
1:26:48to create again. So now this JSON is
1:26:51ready. It will print it again. So now it
1:26:54is 9092.
1:26:56Let me put this new secret. Now this new
1:27:00secret has been put. Earlier it was 909.
1:27:03Now it is 92. So now our streaming code
1:27:07shall work fine. Uh let's run it one
1:27:10more time.
1:27:11So this is run. Now let's run the
1:27:14streaming code directly.
1:27:16So this is our streaming code. Let's run
1:27:19it.
1:27:22Uh let's run this as well.
1:27:25And this time it is successful. It ran
1:27:28quickly. Now let's check out our table
1:27:31again.
1:27:32And here we can see now it has 89
1:27:35records. I was expecting like three new
1:27:38records on top of 79 but it has inserted
1:27:4210 records. Let me double check. So I
1:27:45thought it has written these three but
1:27:47looks like this is some line. It has
1:27:48written these seven plus these three. So
1:27:51total 10 records. So we are good. So
1:27:54those additional 10 records has been
1:27:56written means our code is working fine.
1:27:59So now let's take this streaming code to
1:28:01the SDP pipeline.
1:28:05So where it is?
1:28:08Yeah, this is our streaming code. So
1:28:10this is uh reading the stream. So let me
1:28:14copy this from here and then we will
1:28:17take it to our uh SDP pipeline. So after
1:28:21that I will provide it here. So this is
1:28:24the streaming piece
1:28:26and once we have the streaming data
1:28:28frame then what we did was the uh we
1:28:32parsed it we converted these into
1:28:34timestamps. So let me copy this code as
1:28:37well and let me put it here
1:28:45and this import statement I will keep it
1:28:47at the top.
1:28:53So now this piece is also ready and I
1:28:56also want to add a new column here as as
1:28:59the current time stamp as the injection
1:29:02timestamp so that it is available for
1:29:04better tracking. For that we need to do
1:29:07something like f dot current timestamp
1:29:10but uh yeah it is coming here and then
1:29:13we will name it as injection timestamp.
1:29:23So this will be the name and here it
1:29:25will be available as the current time
1:29:27stamp and this f it is the function. So
1:29:30we will need to import that as well. So
1:29:33it is like from pispar
1:29:37dossql dot
1:29:40uh functions
1:29:44as f. So this functions it will be
1:29:47imported as f and f dot current
1:29:50timestamp will be our current timestamp.
1:29:55So you will think like I am just copy
1:29:56pasting the code from there. So what is
1:29:59the advantage? So now comes the
1:30:01advantage part. So as you already know
1:30:05that for writing this stream we had to
1:30:07provide these modes the checkpoint
1:30:10location the trigger type the table name
1:30:13and everything. But that one advantage
1:30:16of this is that we don't need to
1:30:18maintain any uh checkpoint or anything
1:30:22any location nothing simply what we need
1:30:24to do is we need to return this data
1:30:27frame and that's all. So that is how the
1:30:30writing part it has been simplified. How
1:30:33it will work internally is that it will
1:30:36return this data frame. This data frame
1:30:39will be written into this table as a
1:30:42streaming table automatically
1:30:45and the checkpoint and everything will
1:30:47be maintained automatically by datab
1:30:49bricks. We don't need to maintain
1:30:51anything there. And when it comes to to
1:30:53the trigger type so for that we get the
1:30:56trigger type here. So if we go to the
1:30:59settings the pipeline mode it is
1:31:02triggered. So we can schedule it. we
1:31:04also get a continuous mode where it will
1:31:07run it in the actual real runtime way.
1:31:10So that is that is something we will try
1:31:12later on when we will build the silver
1:31:15part and then we will see the end to end
1:31:18streaming data flowing. But for this
1:31:20particular table that's the entire code
1:31:23that we need. So now for this table the
1:31:26code is ready. So we are going to simply
1:31:29go ahead and run this code. And to run
1:31:32that we simply click this run pipeline.
1:31:36And there are other advantages also we
1:31:38get. Oh before that let me select the uh
1:31:41catalog and schema name. So it has some
1:31:44default catalog and schema we need to
1:31:46provide. So it will be fine guard and
1:31:49branch will be the default one. And then
1:31:52we need to run the pipeline. This is
1:31:54where we can also configure the default
1:31:56catalog and schema names. Currently it
1:31:59is waiting for the resources.
1:32:02And this is another advantage that we
1:32:04are getting a good monitoring screen
1:32:07here. We can also apply data quality
1:32:09checks that we will apply in the later
1:32:11state and also connecting different
1:32:14streaming table that is also made very
1:32:16easy when we go with these STP pipelines
1:32:19that also we will see. And now the
1:32:22resources are available. It is
1:32:24initializing the pipeline
1:32:28and looks like it threw an error. If we
1:32:30go it says invalid syntax. If we click
1:32:33this jump to code, it is taking here. So
1:32:36it is saying oh. So we didn't mention
1:32:39the import keyword. So in between here
1:32:42we need to import functions as f. That
1:32:45is the correct that is how we actually
1:32:48use it. Now let me run it again.
1:32:52So and this time it threw another error.
1:32:54It is saying find guard is not found. So
1:32:57again I made a typo. I will correct it
1:33:00and then run it again.
1:33:03And this time looks like syntax wise
1:33:05things are fine. So it is currently
1:33:07setting up the tables. And now the
1:33:09pipeline has been started. And if you
1:33:12are also using the free addition
1:33:14sometimes the waiting for resources it
1:33:16may take lot of time also depending on
1:33:18the time of the day when you are
1:33:20running. So if it is taking lot of time
1:33:23you can stop and try at some other time.
1:33:25That is how it works.
1:33:28And this pipeline it is executed
1:33:29successfully. Now we can see the success
1:33:32here. It ran for 23 seconds. It it wrote
1:33:3689 records in the output. And then uh if
1:33:41you click here view in catalog. So then
1:33:44it will open a new tab and show us that
1:33:47table which is created in the catalog
1:33:49directly.
1:33:51So this is the table which is created.
1:33:53If we go to sample data, we will check
1:33:55out some sample data of this table.
1:33:59And if I scroll on the left side, so
1:34:01this is the icon I was talking about. So
1:34:04now because it was created by SDP
1:34:07pipeline, so now it is marked as a
1:34:09proper streaming table. So datab bricks
1:34:12it will also know automatically that it
1:34:14is a streaming table and internally it
1:34:17is managing its uh checkpoint and all
1:34:19the things internally without mentioning
1:34:22or doing anything.
1:34:24And here we can see the data that has
1:34:27has been written into this table. The
1:34:29key the value the topic partition offset
1:34:34timestamp everything and this is the
1:34:36injection timestamp that we had added
1:34:38from our site. So now this is how our
1:34:42row data looks like and uh next thing
1:34:45what we need to do is we need to parse
1:34:47this value the JSON value and we need to
1:34:51create proper columns so that we can use
1:34:53this data properly. So to parse this raw
1:34:56data we are going to create our next
1:34:59silver layer where we will parse these
1:35:01columns and write them into their
1:35:03individual columns. So let's go ahead
1:35:05and do it. And for that I'm back in our
1:35:09pipeline. Here again I will click on
1:35:12create and I will create a separate
1:35:14folder for silver layer.
1:35:19And here we will create again a new
1:35:22file.
1:35:27And this file we will name it as
1:35:29transactions silver
1:35:39and now the top part it will be almost
1:35:42same. So I'm going to copy it from here
1:35:45and then we will use the same here. So
1:35:49imports and all we will keep it the
1:35:51same. Uh looks like it is gone. Let me
1:35:53paste it again.
1:35:55And then this time the table name will
1:35:57be same but the schema name it will be
1:36:00different because we will create it in
1:36:01silver layer this time. And the comment
1:36:04we will name it as like par
1:36:08and cleaned
1:36:14transactions
1:36:16data
1:36:20and this function also we will name it
1:36:22as transactions silver
1:36:26and in here now we need to write our
1:36:28logic. So first of all we will re read
1:36:31the bronze table as a streaming table.
1:36:34So we will name it as bronze df and then
1:36:37we will select spark dot read dot table
1:36:43so that it will read the bronze table as
1:36:45a streaming table and the name of the
1:36:48table. Oh it is guessing automatically.
1:36:50So this is the name of our table.
1:36:54So this is our bronze streaming data
1:36:56frame. Now we will transform it. So we
1:37:01will say transformed df equals to this
1:37:05bronze df
1:37:07and then we will say dot select
1:37:13and in the select part we will specify
1:37:15multiple things. So first of all we want
1:37:18to parse our JSON right. So for that we
1:37:21will say that we have imported the
1:37:24function as f. So there are multiple
1:37:27functions inside the functions. Then we
1:37:30will say from JSON and here we need to
1:37:33specify the uh column. So we will say
1:37:38our column it is value only like it is
1:37:40guessing.
1:37:45And then we need to specify a schema. So
1:37:49for now I will mention only the word
1:37:51schema and then we need to give it as a
1:37:54alias. we will name it as maybe data
1:37:57like it is suggesting. So but before we
1:38:00proceed further we need to specify a
1:38:03schema. So whenever we are parsing a
1:38:06JSON the best practice it is to specify
1:38:09a schema. So what we will do is we will
1:38:12copy this sample data and then uh we
1:38:16will create this schema. So we will do
1:38:18it using uh genie code. So here we will
1:38:22select this. Then we will press this
1:38:25icon and we will say like create the
1:38:31schema
1:38:32based on the sample
1:38:35data this. So we will provide some
1:38:38sample data and then we will ask it to
1:38:41create that schema for us
1:38:44and looks like it is trying to use like
1:38:47f schema of JSON. So this sample it has
1:38:50provided automatically. And then it is
1:38:53asking to derive the schema from this.
1:38:56But this is not a best practice. So I'm
1:38:59going to reject this change.
1:39:01So I pressed escape because that is not
1:39:04what we want. So I will say like here
1:39:08we will need to be more specific here.
1:39:11So we will say that schema
1:39:16and then we will say like uh create this
1:39:20schema
1:39:26is strct type and strruct
1:39:31fields from the sample JSON.
1:39:35Then the JSON that we had copied let me
1:39:38copy it again. So now we are saying like
1:39:41created a strct type and strct fields so
1:39:44that it will create it in the very
1:39:46standard pi spark way. So this time it
1:39:50is creating it properly. So I will
1:39:52accept this change and now we can see
1:39:55that it has created this fstruct type.
1:39:58This is the standard way of creating
1:40:00schema. Then the field it is creating
1:40:03each and every field like it is a
1:40:05string. It is these all are string
1:40:08amount is a double type. So it has
1:40:10specified each and everything and this
1:40:13is the standard way of creating schema.
1:40:16Now this schema will be put up there. So
1:40:19what it will do is it will take the
1:40:20value column. It will attach this schema
1:40:24and then uh it will point out all these
1:40:27columns properly.
1:40:29So now what it is doing is it is reading
1:40:32this data column is it will create it as
1:40:34a kind of dictionary and apart from this
1:40:37value and this data column. What other
1:40:40things we need is uh this the other
1:40:43column for tracking for example the
1:40:46Kafka topic partition offset and these
1:40:50timestamps. So what we will do is we
1:40:52will say that f dot call and then we
1:40:57will uh take the column or it is taking
1:41:00automatically topic partition offset and
1:41:04then we also need to take time stamp.
1:41:09So that also it is taking time stamp
1:41:11type we don't need it
1:41:15and then we had the column like uh
1:41:20the injection time stamp. So that is
1:41:23something we will take
1:41:28and this column we will need to add
1:41:30alias so that uh we will give them as
1:41:34some meaningful name so that in silver
1:41:36layer these are understood properly. So
1:41:39instead of topic we will say uh Kafka
1:41:42topic
1:41:50and instead of simple partition we will
1:41:52say Kafka partition
1:41:56and instead of offset we will say Kafka.
1:42:00So let's see it is able to like suggest
1:42:02automatically
1:42:12and then uh instead of injection time
1:42:14stamp we will name it as bronze
1:42:17injection time stamp.
1:42:22So this is how we are going to select
1:42:23all these columns. Now this data column
1:42:26is available as a dictionary but what we
1:42:29want is we want each and every value as
1:42:32a separate thing. So for that what we
1:42:35will do is
1:42:38uh so I will press tab for now so that
1:42:41it is selecting all the columns and
1:42:43returning this d this uh transform df
1:42:48and this is how we will return all these
1:42:50columns here the data dotstar what it
1:42:54means is that whatever fields are there
1:42:57with this schema so it will create each
1:42:59and every field inside data as a
1:43:02separate separate column. So all the
1:43:04columns will be extracted and then those
1:43:07will be created as separate columns. And
1:43:10here we want to also add a uh current
1:43:14timestamp.
1:43:16So that we will name it as silver
1:43:18injection time stamp.
1:43:22So this will be additional column. Now
1:43:25we are returning this transformed df. So
1:43:28this is this df. This transform data is
1:43:31being returned as a streaming data frame
1:43:34and then this data frame will be
1:43:36automatically written into this table
1:43:38like we understand how STP works now.
1:43:43So now we will try and run this pipeline
1:43:46see if it is working as expected. But
1:43:49before that we will see that this is the
1:43:52mark. So it means like this file is
1:43:54excluded from pipeline source code. So
1:43:56by default it will not run it. So we
1:44:00need to include this. So I will include
1:44:02this silver folder itself. So that it is
1:44:05included going forward.
1:44:08So now it is included and this will also
1:44:11run now. So now we will go ahead and run
1:44:13our pipeline. So I will click on run
1:44:16pipeline.
1:44:18And here looks like our pipeline failed
1:44:20saying that uh SQL functions has no
1:44:24attribute uh strct field. So looks like
1:44:27this is a like mistake made by Genie
1:44:30code. Sometimes AI can make mistake for
1:44:32sure. So what we will do is we will
1:44:35import it separately like from pispa
1:44:42dossql.types.
1:44:45Uh it is saying like import all these.
1:44:47So I just pressed tab and it imported
1:44:50all these separately automatically. Or
1:44:53simply what we can do is we can also
1:44:55mention star so that it will import all
1:44:58the types but these are not available as
1:45:01f dot. So we will need to remove these
1:45:07or what we will do is we will simply say
1:45:10like uh we will use genie code to remove
1:45:12it. We will say like
1:45:16remove f dot
1:45:21from all places here.
1:45:27So here looks like it has removed it but
1:45:31the accept button it is gone. So
1:45:33sometimes it it may be bit weird that we
1:45:36don't see that button button anymore. So
1:45:39I will put escape for now. So the
1:45:41changes will be ignored. And this time I
1:45:44will try not to scroll. So I will do
1:45:47control I and I will say remove F dot
1:45:52from all places.
1:45:58And this time without scrolling I will
1:46:00first click on accept so that the
1:46:03changes are made. And looks like it is
1:46:05it has made the right changes now. And
1:46:08before we run the pipeline again, let's
1:46:11add one more thing that is about data
1:46:13quality which I just forgot. So when we
1:46:17use this declarative pipeline, so it
1:46:19comes up uh with the uh standard way of
1:46:22defining data quality checks. So how do
1:46:25we define is either rate DP dot
1:46:29expect
1:46:31or drop. So this is the standard word we
1:46:35use like either this condition shall be
1:46:37satisfied or otherwise drop this record.
1:46:41So then we name the we give a name to
1:46:44the DQ rule saying like transaction ID
1:46:48is not null. This is the name it can be
1:46:51any name and this is a valid SQL syntax
1:46:54as well. So the name we will give it
1:46:57maybe before this any other name we will
1:47:00name it as like valid
1:47:04transaction ID.
1:47:07So this is the name of the DQ rule. This
1:47:09is the actual rule. So it shall not be
1:47:13null. That is the expectation we are
1:47:15setting. If it comes anywhere null, it
1:47:17will drop that record.
1:47:20And this is one type of rule. We can
1:47:22define multiple also. For example, oh
1:47:25this is putting automatically. So I will
1:47:27just press tab and it says valid
1:47:30customer ID shall not be null. Card
1:47:32number shall not be null. Merchant ID
1:47:35shall not be null. And then there there
1:47:38is another type of rule as well which
1:47:41says like uh it can be either expect or
1:47:44drop or only expect and not drop. So we
1:47:49don't need to specify anything like not
1:47:51drop. uh but if we specify only expect
1:47:55it will check it uh and in the output it
1:47:58will show that the rule was failed but
1:48:00it will not drop this record that is how
1:48:03it works. So this is how we define these
1:48:05data quality checks and when we run the
1:48:08pipeline we will see how the output
1:48:10comes and in this rule the rule is valid
1:48:13amount and amount shall be greater than
1:48:15zero that is the expectation we are
1:48:17setting here. So this is additional
1:48:20functionality that comes ready to be
1:48:22used along with the spark declarative
1:48:24pipelines.
1:48:26So now as we have defined our dq rules
1:48:28and fixed the f dot that error let's run
1:48:32the pipeline again.
1:48:35So now it has initialized the pipelines
1:48:37and looks like there was no syntax error
1:48:40and now it we will see two transactions
1:48:43here. One for bronze, one for silver. So
1:48:46first it is running the bronze one
1:48:48because both of these are included in
1:48:50this pipeline. So it will always run uh
1:48:54every pipeline sorry every table or
1:48:56every file that is there. And here we
1:48:59can see that when it ran branch so it
1:49:02did not write any output record. And why
1:49:05is that? Because we already ran it there
1:49:08are no new record pushed from our side.
1:49:11So that is why it is not coming. And on
1:49:14the other hand on silver table it has
1:49:16written those 89 records and
1:49:19expectations this is the effect of of
1:49:22these rules that we had defined. We had
1:49:25five rules and it is saying five met. It
1:49:28means that all the five rules were met.
1:49:30If I click this we see this that 89 100%
1:49:35records has been written and all these
1:49:38uh
1:49:40all these rules has been satisfied.
1:49:42There were no failures and this is the
1:49:45action it would take if it was there. So
1:49:47expect or drop it will drop it and
1:49:50amount it will allow it because it was
1:49:52only expect and not the drop part. And
1:49:56if I close this and if you want to check
1:49:58out our table the silver table that has
1:50:01been created. So we will click this icon
1:50:05here it will open this catalog explorer
1:50:08and our silver table. And in our silver
1:50:11table we can see all the individual
1:50:13columns inside the value piece the JSON
1:50:17it has been passed and created as
1:50:19individual columns. So now let's check
1:50:21out the sample data here. And here we
1:50:25can see the silver table. So the data
1:50:27has been passed. We have proper separate
1:50:30column for each of the data point like
1:50:32customer ID, card number, merchant ID,
1:50:35name, amount, everything. It is created
1:50:38as a separate column here and then these
1:50:42Kafka columns has been renamed like we
1:50:45did. This is our bronze injection time
1:50:47and our silver injection time. So all
1:50:50the columns has been created properly.
1:50:53And now we can say that bronze and
1:50:55silver layer for this Kafka stream has
1:50:58been built properly. And now let's try
1:51:01to run some incremental data. But this
1:51:04time we will run it in continuous mode
1:51:07because we are done we have done lot of
1:51:10with this triggered mode. Let's see some
1:51:12realtime data coming in how it comes. So
1:51:15if I click this continuous and click on
1:51:18save.
1:51:19So this pipeline it will start running
1:51:21in continuous mode. Now it will start
1:51:24and never stop it until we stop it
1:51:27manually or something goes wrong. So
1:51:30till then it will keep running. So it is
1:51:33currently waiting for resources. In the
1:51:36meantime, let me show you this pipeline
1:51:38graph. If you click this, we can see
1:51:41that automatically it has detected the
1:51:43dependency that first it has to run the
1:51:46silver sorry bronze uh transaction one
1:51:49and then it will run the silver
1:51:51transaction one. So due to this
1:51:53dependency, it will always run this one
1:51:55first then this one.
1:51:58And like I was saying in free edition it
1:52:00can take time sometimes for waiting for
1:52:03the resources. So let me pause it here.
1:52:07And looks like it has found the
1:52:08resources. It took only around 2
1:52:10minutes. Currently it is initializing
1:52:13the pipeline. And now it is running.
1:52:17So by default it will first run both the
1:52:19tables one after the other. And as you
1:52:22may guess there is currently no data
1:52:25going to the producer. So it will not
1:52:27find any new data and it will pause it
1:52:30after some time. So now it goes into the
1:52:33idle state as there is no data but it
1:52:36doesn't mean it has stopped. It is still
1:52:38running. So without making any changes
1:52:41here let's start producing some data and
1:52:44then we will see some realtime data
1:52:47flowing in. So I'm running our producer
1:52:51script again so that it start pushing
1:52:53new data. It has started pushing. And if
1:52:56we go back here here now we will see
1:52:58that without making any changes it will
1:53:01pick up that data automatically from
1:53:03that Kafka topic and then it will start
1:53:06processing.
1:53:08So here it goes automatically it started
1:53:10the bronze script first. It has started
1:53:13writing output record and then silver
1:53:16also it started automatically and we
1:53:18will see that the bronze part first it
1:53:21is coming here like 100 to 123 then
1:53:24those records are going there as a
1:53:26batch. So as a micro batches it is
1:53:29performing uh the the end to end
1:53:32injection of the data. First it is it is
1:53:35ingesting into our bronze table then
1:53:37into silver table after applying these
1:53:40data quality checks and after the
1:53:42parsing. So currently this is how the
1:53:45real streaming pipeline looks like our
1:53:48data it is going in real time. Uh then
1:53:52this is being read in real time and it
1:53:55is being processed in both the bronze
1:53:57and silver layer end to end. And this
1:54:00pipeline it will keep running until we
1:54:02stop it or we make any change here and
1:54:04there.
1:54:06So I hope now it gives a better sense
1:54:09how the end to end data flow looks like.
1:54:13So first of all I will stop our producer
1:54:16script and these records that it has put
1:54:18up so many records. So it will keep
1:54:21processing them and then uh once it has
1:54:25processed all those records then it will
1:54:28go into idle state. So now after a few
1:54:31seconds it has read all those records
1:54:33and it is again in the idle state. So
1:54:36this is how the end to end streaming
1:54:38data looks like. So now I will go ahead
1:54:41and stop this pipeline because we don't
1:54:44want to keep it running for rest of the
1:54:47development. And the trigger type also
1:54:49for now we will change it to triggered
1:54:52so that we make the changes we add more
1:54:55scripts and then we run it whenever we
1:54:58want to. So here I would like to take a
1:55:01pose and then to to uh add a checkpoint
1:55:06right. So what we have done we have set
1:55:08up our Kafka topic everything properly.
1:55:11Our producer is sending the data in real
1:55:14time. We are getting the data in real
1:55:16time in bronze and silver layer and
1:55:18processing it end to end properly. So
1:55:21what I would suggest is that here you
1:55:23take a pause you do a hands on on all
1:55:27these pieces whatever has been explained
1:55:29so far. Once you are comfortable with
1:55:31all these pieces then you move on to the
1:55:34next part of the demo because that will
1:55:36be on top of it. So once you are
1:55:40comfortable with these pieces then you
1:55:42will be able to understand the next
1:55:44pieces better
1:55:46and if you are already comfortable until
1:55:48here then let's move on. Now let's
1:55:51understand first what we are going to
1:55:53work on next. So so far we have brought
1:55:56in the data from Kafka. We uh tested out
1:55:59Spark streaming and then we have built
1:56:02bronze and silver layer with Kafka data.
1:56:06Now we are going to set up our first
1:56:08fraud detection mechanism and for that
1:56:12we will take one uh simple example of
1:56:15fraud. So for example if our
1:56:17transactions are coming so as a customer
1:56:20if I have set up a transaction amount
1:56:23let's say it is 50k or 100k it is a
1:56:26transaction limit in my account and if
1:56:29any transaction has been performed more
1:56:31than that value that means it is a
1:56:33fraud. So that is the fraud we are going
1:56:36to detect and for that we will need
1:56:38additional data now because our stream
1:56:41our current streaming data it is having
1:56:44a transaction customer id transaction
1:56:46amount and everything but we will need
1:56:49the for that given customer what is the
1:56:52uh transaction limit that I have set up
1:56:54for that kind of data it is not a
1:56:56streaming data the customer related data
1:56:59that is uh considered as batch data
1:57:02because normally that is stored code in
1:57:04a postgrace or any other transactional
1:57:07system. So now what we are going to do
1:57:10is we will prepare our Postgress SQL
1:57:13with as a free addition. We will put our
1:57:16customer data there. We will ingest that
1:57:19data as a batch load using lakeflow
1:57:22connect into bronze and then into silver
1:57:24layer as well using similar lakeflow
1:57:27declarative pipelines and after that we
1:57:29will join both this streaming and the
1:57:32static data. So we will cover this
1:57:34stream static join part and using that
1:57:37we will build the gold layer as well as
1:57:40we will detect the fraud and then we
1:57:43will send the real time alert as a
1:57:46email. So this is the piece we are going
1:57:48to work on next. So for that let's start
1:57:51by setting up our Postgress SQL.
1:57:55So for Postgress there are lot of
1:57:57different options we can use. Different
1:57:59companies they provide their own hosted
1:58:01version of postgrace. One of them a good
1:58:04one is neon. So I normally prefer this
1:58:07one if you follow regularly because here
1:58:10the sign up process is simple using
1:58:12this. It has been made very simple. So
1:58:14if you search for neon postgress this is
1:58:17the link that will come. You shall open
1:58:19it and do a sign up. Here we don't the
1:58:22sign up process is very very simple. We
1:58:24just need to uh like login using our
1:58:28Gmail account or using our email id. It
1:58:30is very straightforward. And once we
1:58:33sign up then we will get a login page.
1:58:36And after login this is like the
1:58:39homepage that we will get. Here we need
1:58:41to go to this project section. And in
1:58:44here we can uh create a new project. And
1:58:47as you can see I'm using the free
1:58:48edition currently. So I didn't add any
1:58:51credit card or anything. So here we
1:58:53simply need to create a new project. We
1:58:56will name it as fine guard which is the
1:58:58name of our current project. Uh rest of
1:59:01it we will leave as default. A region
1:59:03maybe we can select whichever is closest
1:59:06to our location. So after selecting
1:59:08these options we will simply click
1:59:10create and our project it will be
1:59:13created. A database has been spinning
1:59:15up.
1:59:17Uh and here you can see that the project
1:59:20that we just created has been selected
1:59:23and then uh what we need to do is we
1:59:26will need to create some table now. So
1:59:29we will go to table section and here we
1:59:32will see that currently there is no
1:59:34table at all. So to create a table we
1:59:36will go to SQL editor part
1:59:39and to run the SQL query I have created
1:59:42two scripts customer historic and
1:59:44incremental load. So these scripts also
1:59:47these are included in GitHub. So you can
1:59:50directly use these. So I will simply
1:59:52copy the content and then I will come
1:59:55here. I will delete this part and paste
1:59:58it and then run it.
2:00:01So this has been executed successfully.
2:00:04First it dropped the table if it was
2:00:06there. It created it inserted,000
2:00:09records and it did a commit on those
2:00:11records. So now if we simply do select
2:00:15star from customers so this table shall
2:00:19be there. If we run the query we can see
2:00:22that a customer related data has been
2:00:24inserted here. Now so customer ID,
2:00:27customer name, gender, age, city, there
2:00:30are different different details which
2:00:32has been added here. The one which we
2:00:34are interested in is this transaction
2:00:36limit. So for each customer this limit
2:00:39has been configured and uh if the actual
2:00:43transaction is more than this amount
2:00:45then we will send a realtime alert to
2:00:48email id of that particular customer. So
2:00:52if you see here I have added a static
2:00:54email id which is my own email id so
2:00:56that I can show the email that is being
2:00:59sent. So if you want to you shall uh
2:01:01replace this email id in this script. So
2:01:04you can simply uh find that email id
2:01:07this one and replace it with your own.
2:01:10So if I putt ctrl h I can find and
2:01:12replace it with anything. So you shall
2:01:15use your own email id and amount part
2:01:17also currently I I am using this uh 100k
2:01:21everywhere because uh like sending some
2:01:24fraud transaction is easy while
2:01:26generating the data that is why I'm
2:01:28keeping it fixed. So now our table is
2:01:31ready. If we go to the table section
2:01:35then here we will notice our customers
2:01:37table has been created. So this is the
2:01:40table that we are going to ingest into
2:01:42datab bricks. Now for that we will need
2:01:45connection details
2:01:47and for connection details we need to go
2:01:49to settings.
2:01:51Uh looks like it is not here. Let me see
2:01:55if we go to dashboard. Yes here we can
2:01:58see. So if we click on this connect
2:02:00option uh we will see this connection
2:02:02string. So this contains the details
2:02:06that we will need to uh to connect to
2:02:10this uh postgress SQL. So I'm going to
2:02:12copy this snippet by clicking this
2:02:15button. Then I will go to a notepad open
2:02:18a new tab and I will paste it. So this
2:02:21contains like multiple things that we
2:02:23will understand how to use this in datab
2:02:26bricks. So now to ingest the data from
2:02:28here we will go to datab bricks. So this
2:02:32is our datab bricks page and here we
2:02:34will go to the jobs and pipeline
2:02:37section. First I will duplicate this tab
2:02:39so that uh I can still use the current
2:02:42tab and here we will go to jobs and
2:02:45pipelines.
2:02:46And here there are three types of uh
2:02:49jobs we can create. So injection
2:02:51pipeline is the one we will use now to
2:02:53ingest the data. ETL pipeline. This is
2:02:56the Spark declarative pipeline that we
2:02:58created previously. From here also we
2:03:01can create the same. This job it is for
2:03:03orchestration running things end to end.
2:03:06This one we will use later on. So for
2:03:08now I will click on this injection
2:03:10pipeline. And here we get all these
2:03:13options from all these sources we can
2:03:15ingest the data. So the one we are going
2:03:18to use is Postgress SQL. So this is the
2:03:20one we are going to use.
2:03:23So here first of all we are going to
2:03:26create a connection and here connection
2:03:29name we can provide anything like fine
2:03:32guard
2:03:33uh postgress
2:03:36that is the name I'm going to provide it
2:03:39then we need to provide the user
2:03:41password and the host name here so for
2:03:44that we need to go to this URL and get
2:03:47the details so after this double slash
2:03:50this neon owner so This is what is our
2:03:54username. So this is what I will paste
2:03:56here. And after this colon, so this
2:04:00colon is not included in any part. And
2:04:03before this add the rate, this is our
2:04:06password. So I will copy this password.
2:04:10And after this 8 the rate
2:04:13uh before this neon slash. So this is
2:04:16what is our URL. So this thing you can
2:04:19maybe take a screenshot and then refer
2:04:22and then you can like uh use your own.
2:04:25So just remember before colon is the
2:04:28username before read the rate is the
2:04:30password and before neon db not
2:04:33including slash that is our URL. So
2:04:36these things I have pasted. Now I will
2:04:38simply click on create connection.
2:04:42So now my connection is created. It is
2:04:44not yet verified. it will be verified in
2:04:47the next step. So for now it has been
2:04:50selected. I will click on next.
2:04:53Now we need to give a name to the
2:04:55pipeline and this pipeline we will name
2:04:58it as pine guard
2:05:01customers
2:05:03injection
2:05:06and uh
2:05:08then we need to select a catalog. Here
2:05:10we will simply select our fine guard
2:05:13schema also we will select bronze one
2:05:15here. This is for like event log. So
2:05:18internally it tracks some events. So
2:05:20that is what it will do here. Then we
2:05:22will click on create pipeline and
2:05:24continue.
2:05:28So this step it took couple of seconds.
2:05:30I had paused the video in the meantime.
2:05:32Here we need to provide a database name.
2:05:35So database name it is going to be our
2:05:38neon db. So I will put it here and click
2:05:41on this plus icon. Now it will check in
2:05:44this database what all schemas are
2:05:46available and then the list of tables.
2:05:50So it has these three schemas and if we
2:05:52go to our uh Postgress UI there we will
2:05:56notice that our customers table it is
2:05:59available inside the public schema and
2:06:02neon DB is our database. So here if we
2:06:05go inside this public one uh this is
2:06:08where we will find our customers table.
2:06:12So we will select this table. Uh
2:06:15destination table name we will keep the
2:06:17same. We don't want to change it. Cursor
2:06:19column it is for incremental data load.
2:06:22So here normally we use timestamp kind
2:06:24of column which helps while doing the
2:06:27incremental loads.
2:06:29So currently it is fetching the list of
2:06:31columns from source. So it is taking
2:06:33some time. Now it is ready. If you don't
2:06:35want incremental load, we can click this
2:06:37batch snapshot as well. So that every
2:06:40time it will do a full load but we have
2:06:43a incremental column. So we will select
2:06:45this update time stamp and in the
2:06:47primary key we will select customer ID.
2:06:50So now our configurations are ready. We
2:06:52will click next.
2:06:55Here this is the destination selection.
2:06:57So in destination we want to create our
2:06:59table into bronze schema only. So it is
2:07:02already selected. We will click on next.
2:07:06And here it is creating a schedule
2:07:08automatically. So by default it want to
2:07:10run it every 1 hour. We don't want to do
2:07:13that. So I will simply delete it. Uh
2:07:16because initially we will run it
2:07:18manually. Then we will add some
2:07:20orchestration. And this is the failure
2:07:22notification. So on failure it will send
2:07:25an email. We can keep it or delete it if
2:07:28we want. And finally we will click on
2:07:30save and run pipeline. Once I click it
2:07:34then it will create the pipeline
2:07:36completely and it will start running it.
2:07:39And the running UI it will look similar
2:07:41to the spark declarative pipeline UI. It
2:07:44is kind of part of that only. So
2:07:47currently it is initializing our
2:07:49pipeline
2:07:50and after it is uh initialization is
2:07:53done then it has started our pipeline.
2:07:56So it is saying like it is creating some
2:07:58staging view and then it is writing to
2:08:01our customers table.
2:08:03Now the execution it is complete. So it
2:08:06has been completed successfully. So I
2:08:09will click here this icon to see the
2:08:12table.
2:08:16So this is our table and let's check out
2:08:18some sample data here.
2:08:21So this is how the data looks. All our
2:08:23data has been loaded as it is from
2:08:26source to this particular table.
2:08:30So as we have loaded historical data now
2:08:32let's try out some incremental data as
2:08:35well. And for that I have created a
2:08:37separate script. So this script it is
2:08:40going to insert two new records with
2:08:42customer ID 101 and 102 and the customer
2:08:47ID 1 that is what it is going to update.
2:08:50So this person with this customer ID. So
2:08:54his current entry it is uh like it is in
2:08:57Mumbai and Maharashtra and this is the
2:09:00annual income and with this new entry it
2:09:03is going to move from Mumbai to
2:09:07Bangaluru and the salary it is going to
2:09:09increase. So we are going to promote uh
2:09:12this person.
2:09:14So I'm going to copy this entire script.
2:09:17Then we will go to our postgress.
2:09:20There I will paste this script and
2:09:23simply run it.
2:09:25So now those records has been updated.
2:09:29And now we will run our injection
2:09:31pipeline again. So here is our pipeline.
2:09:34I'm going to simply start it again.
2:09:39So now as it is completed, we will go to
2:09:41our customers table and we will simply
2:09:44refresh this.
2:09:45So now we will see that the customer ID
2:09:481 is not visible here. It is because it
2:09:51is probably updated and here it will
2:09:53show only 100 records. So to confirm the
2:09:57data we will have to query it in a
2:09:59notebook. So we will go to our project
2:10:01folder and here let me create a test
2:10:04notebook.
2:10:06And now let's try to query our table. So
2:10:10we will do SQL
2:10:14select star from this table.
2:10:22So here is our data and if you see so
2:10:25now it has,02 record two additional rows
2:10:28has been inserted and one has been
2:10:30updated. So if we sort by customer ID
2:10:33for customer ID 1, we see that the city
2:10:36has been changed to Bangaluru and the
2:10:38income has also been updated. So this
2:10:41row has been updated with the new data.
2:10:44And if we do the descending order, so
2:10:46these two ids are the new ids which we
2:10:49inserted. So these are also inserted
2:10:51into our table. So this is how
2:10:54incremental load it is made very simple
2:10:57uh using lakeflow connect. And here we
2:11:00did not have to write any additional
2:11:02code. We just selected the primary key
2:11:04and the watermark or timestamp column
2:11:07and automatically it is able to take
2:11:09care of everything.
2:11:11So now revisiting our architecture. So
2:11:14we have now set up a postgra SQL in
2:11:17neon. We used the lakeflow connect. So
2:11:20the injection pipeline that we created
2:11:23the technical term for datab bricks in
2:11:25that is that is called as lakeflow
2:11:26connect. we use used that and then we
2:11:30have now brought in our customers table
2:11:32to the bronze layer. So now is the time
2:11:35to bring that table to silver layer. So
2:11:38let's go ahead and do that. So next what
2:11:42we are going to do is we are going to
2:11:43load our customers table into silver
2:11:46layer and that we will do using the
2:11:49spark declarative pipeline only. So here
2:11:52one option is that in this pipeline
2:11:54itself our original streaming pipeline
2:11:57we can add it but we we do not want to
2:12:00run uh that silver injection and that uh
2:12:03customer's injection piece continuously
2:12:06when we run the streaming injection that
2:12:08process will be running as a batch
2:12:10process. So that is why we will not
2:12:13create that within this pipeline and we
2:12:16will create a small separate pipeline
2:12:18for that. So I'm going to duplicate this
2:12:21uh tab and in here I will go back to our
2:12:25old files. So this is the folder for
2:12:28fine guard streaming that we we are
2:12:30already using. So here we will create a
2:12:33new ETL pipeline
2:12:36and this pipeline we are going to name
2:12:39it as
2:12:40uh maybe fine guard uh
2:12:48customers
2:12:50uh silver load because this is what we
2:12:52will be doing here. So this is the name
2:12:55we have given.
2:12:58So looks like I already have a pipeline
2:13:00with this name. So in my case I will put
2:13:03a two here.
2:13:07So uh this is our folder. So this folder
2:13:10also we will rename it
2:13:22customer silver load
2:13:31and then this transformation folder we
2:13:34will rename it as silver like we follow
2:13:36the standards
2:13:38and this This is the default file which
2:13:40is created. So this one also we will uh
2:13:43rename it to customers silver.
2:13:53So this is where we now we will write
2:13:55the code for silver load of customers
2:13:57table.
2:13:59And for the code piece we will copy our
2:14:01code from our previous uh transaction
2:14:04table. So this is the entire [music]
2:14:06code. I will copy it from here
2:14:10and then I will paste the same into the
2:14:13pipeline that we are building. So this
2:14:15time we won't need all these things
2:14:17because it is going to be simple silver
2:14:20load. So we will only need this uh DP
2:14:24and the functions as F and the table
2:14:27name it is going to be customers in
2:14:29silver layer and then comment it can be
2:14:32like passed and cleaned uh customers
2:14:34data we can say
2:14:38and in terms of uh data quality rule we
2:14:41can say like valid customer ID
2:14:46and then customer ID shall not be null.
2:14:49So I will simply press a tab. So
2:14:52transactions one it is not required. And
2:14:55now we will write our main function. So
2:14:58the function name it can be the same
2:15:01customer silver.
2:15:04And this it returns a data frame
2:15:11and this data frame we have not yet
2:15:13imported. So let's import that.
2:15:19So it is suggesting it came
2:15:20automatically and here we can write our
2:15:23code. So we can say like branch df. So
2:15:28here we will read our branch table. So
2:15:31spark dot readstream
2:15:35dot table and the table name it will be
2:15:39our pine guard bronze and customers. And
2:15:42here one thing to notice is that even
2:15:44though this customers table we will be
2:15:46using as a batch load but here we will
2:15:49be reading it as a streaming table and
2:15:52this why we are doing so that the
2:15:54incremental part and all can be taken
2:15:56care automatically for by this streaming
2:15:59properties. So we will be running it in
2:16:01a batch mode like once a day or
2:16:03something. But uh to use those like
2:16:05incremental capabilities and this uh
2:16:08declarative pipelines here we will be
2:16:10reading it kind of a streaming table.
2:16:13And if you notice the customers table
2:16:15that was created automatically. So the
2:16:17type of that table also is streaming
2:16:20table even though we loaded it in a
2:16:23batch mode. So that is how things work
2:16:25in datab bricks. So now our data frame
2:16:28is ready and uh then like now we will do
2:16:33some very basic transformation.
2:16:35So we will call it as transform df
2:16:39and then we will select the required
2:16:41columns or like we will need all the
2:16:43columns. So then we will write the
2:16:46select statement here and in the select
2:16:48we will be writing multiple columns. So
2:16:51the first one obviously it is customer
2:16:53id. So we will say f dot column and then
2:16:57we need to specify customer id
2:17:03and here the hint that it is providing
2:17:05so it doesn't know for sure that these
2:17:08are real columns it is providing them uh
2:17:10just like whatever normal columns can be
2:17:13there in a customer table. So be careful
2:17:15we will need to provide the actual
2:17:17column names only.
2:17:23So here what we will do is because
2:17:25typing all the columns is very boring.
2:17:28So let's use genie code here to help us
2:17:31out. So I'm copying this branch table
2:17:34name. From this right side I will open
2:17:36genie code
2:17:39and then I will start a new thread. This
2:17:42is where we start a new chat. And we
2:17:44will say like in this file read all
2:17:48columns from this table.
2:17:51And here what we are going to do is we
2:17:53we need all the columns as it is only
2:17:56two small changes we will be doing. So
2:17:59the first change it will be about the uh
2:18:02date column. So the column that is here
2:18:05account opening date. This is coming by
2:18:08default as a vare which is a kind of
2:18:10string. So this string we will convert
2:18:13it to a actual date. And then what we
2:18:16will do is we will add a new column also
2:18:18that will be our silver injection time
2:18:21stamp that will be current time stamp.
2:18:23So only these two changes we will need.
2:18:25So the same thing we will tell to this
2:18:27genie code. So now here I will press alt
2:18:31and then enter so that it it comes as a
2:18:34new line. So then we will say first like
2:18:37uh we will say that account open
2:18:41date column shall be converted from
2:18:46string to date and then we will say a
2:18:50new column
2:18:52uh silver
2:18:54injection timestamp
2:18:58shall be added as a new column.
2:19:04And then third point we will say like
2:19:06keep uh rest of
2:19:10all columns as it is. So this is the
2:19:14instruction we are giving. So based on
2:19:16this it will uh do everything
2:19:18automatically. Otherwise we would have
2:19:20to copy paste it copy paste all the
2:19:23column names. So that we don't want to
2:19:25do. It is a very repetitive task.
2:19:29So now it is thinking what it needs to
2:19:31do and it is reading the table structure
2:19:34of our branch table and then it will
2:19:36decide what to do. And now it is trying
2:19:39to make the changes to this file the
2:19:42file which is currently open. And here
2:19:44we can see that it has made all the
2:19:47changes.
2:19:49Uh so I will accept this.
2:19:53Let me accept this.
2:19:55And here let's have a quick look. So we
2:19:58see all these columns are being taken
2:20:00automatically as it is without any
2:20:02change
2:20:04and while I'm checking it is trying to
2:20:07uh do a dry run automatically. So I will
2:20:10stop it for now because I noticed one
2:20:12thing. So let me uh okay it is still
2:20:16cancelling I think. I click this stop
2:20:18button and the thing I noticed is this
2:20:21uh date conversion. So rest of the
2:20:24columns it is taking as it is and it is
2:20:27adding a new column as the current
2:20:29timestamp also. But this date it seems
2:20:32like it is not converting it properly.
2:20:35It is just doing like alias and then to
2:20:37date probably it has not checked the
2:20:40data in which format the data is there.
2:20:43So I I will copy this and then I will
2:20:46select this line here and then press
2:20:48this icon or I can use this shortcut
2:20:51also that you see above control I. So
2:20:54here I will tell that
2:20:57uh this column
2:21:02is in format this format change it
2:21:06properly.
2:21:10So now it has explicitly specified uh
2:21:13how exactly it is going to convert. But
2:21:16when we use genie from here it is kind
2:21:18of a tricky like now if you scroll the
2:21:21accept button is gone again. I there is
2:21:24no way to accept this change. So if I
2:21:27press escape then like uh the it is uh
2:21:31undone. So I will ask it to do the same
2:21:33thing here
2:21:36like this column
2:21:38is in
2:21:40the format.
2:21:42I lost the format. Let me copy again. So
2:21:46this format
2:21:49convert it properly.
2:21:53So now it has made the same change. So
2:21:56this time like if I scroll this accept
2:21:58button doesn't go away. So it is a good
2:22:01thing. And again it is trying to do a
2:22:03dry run. Maybe we will let it do this
2:22:05time. So this change I will accept it
2:22:08because now it is looking good. And here
2:22:11we can see that the dry run it is
2:22:12successful. And whenever it does a dry
2:22:15run it is like it will not load the
2:22:18actual data. It will just try to run it
2:22:20if everything looks good. There is no
2:22:22syntax error. Those kind of things it
2:22:24does. So that confirms that coding wise
2:22:26there is no problem. And this dry run if
2:22:30we want we can also do it from here. If
2:22:32you click this button we can also do a
2:22:34dry run.
2:22:36And now it is just trying to remove one
2:22:38space. So I will let it do it. It is
2:22:41again trying to do a dry run. So let me
2:22:45stop this time. So I will click this
2:22:48stop here. I will stop this also because
2:22:50we don't need this help anymore. So now
2:22:53we are going to go ahead and run this
2:22:55pipeline.
2:22:57So it it is asking again for the default
2:23:00catalog and the schema. So we will
2:23:03select the same pine guard and then this
2:23:06time we can select silver or bronze. So
2:23:08this part it doesn't really matter
2:23:10because in our code we have mentioned
2:23:13the exact table name full name but we
2:23:16need to select some default one that is
2:23:18how it works.
2:23:20So now it is actually trying to run the
2:23:23pipeline and then load the data and
2:23:26looks like it failed this time. It is
2:23:28saying like cannot move tables across
2:23:31arite catalog. So this error it is
2:23:33strange sometimes I get it. So let's see
2:23:36if this diagnose error it is able to fix
2:23:38it or we will find some work around. So
2:23:42it is trying to do some things like it
2:23:44is trying to suggest to remove this
2:23:47catalog and schema name which it it did
2:23:49not work and it is trying to try out
2:23:51different things. For now I will reject
2:23:54this change
2:23:56and then it is suggesting some options
2:23:58here and there. So I will stop it
2:24:00because this error it is strange. Uh
2:24:02previously also I encounter one
2:24:04encountered one time. If you have not
2:24:07encountered that is totally fine. But if
2:24:10you have encountered this so uh directly
2:24:13here we cannot fix this one. I tried
2:24:15previously also. So I what I will do is
2:24:18I will copy this entire code and we will
2:24:21just create a new pipeline. That is what
2:24:23we will give it a try. So I will go to
2:24:26this old files and this is our existing
2:24:29f existing uh folder name. So here we
2:24:33will create a new ETL pipeline. So we
2:24:37will go ahead and create a new ETL
2:24:39pipeline here
2:24:41and let me close this right side.
2:24:46So for now we will not rename anything.
2:24:48We will keep everything as it is. Simply
2:24:51let me paste our code here and let's
2:24:54give it a try. Let's directly run this
2:24:56pipeline. And it is asking for a
2:24:59workspace and schema here. So again
2:25:02let's select our fine guard
2:25:07and this time let's give it a try for
2:25:09brunch because we are reading the data
2:25:11from brunch.
2:25:13So now the pipeline has been started. It
2:25:16is waiting for the resources
2:25:19and it has set up the tables and now it
2:25:22has started the injection process. So
2:25:24like I was telling that error it is very
2:25:26strange probably you won't get it but
2:25:29you if you get it you simply create a
2:25:31new pipeline and just paste your code
2:25:33that is a workar around that I have
2:25:35found so the injection it is complete
2:25:38now and as you noticed the only change
2:25:40was that this time we selected branch uh
2:25:43so let's check if that is really a fix
2:25:46let me open our old pipeline from here
2:25:49once and let's change this one to branch
2:25:52launch
2:25:57and now let's give it a try just to
2:25:59confirm if that is the real solution.
2:26:04So this time also it failed. So like I
2:26:07was doubting this is some internal error
2:26:09in datab bricks. Maybe they will fix it
2:26:12later on or it could be in free edition.
2:26:14Maybe in actual datab bricks I have not
2:26:16seen this. Uh but anyway it doesn't
2:26:19matter much to us. So what it means is
2:26:21that our code is working fine. It is
2:26:24just that uh we just created a clone of
2:26:28the pipeline and there it worked fine.
2:26:30So now we know like this is the pipeline
2:26:33that is working and this is the folder
2:26:35where it did not work. So this one we
2:26:38will move it to trash
2:26:41and this one we will rename it. So now
2:26:44let's give it proper naming.
2:26:48So let me open it again. So now it will
2:26:51become our fine
2:26:55pine guard
2:26:58customers
2:27:00silver
2:27:03injection
2:27:05and then two because I have some other
2:27:08pipeline with that name and this folder
2:27:11also let me give it the same name.
2:27:15So I will go here. I will rename this
2:27:19root folder
2:27:22and it is done.
2:27:26And this piece also let's rename it.
2:27:29This renaming is not that important but
2:27:31just to maintain the standards uh let's
2:27:34do that.
2:27:39so this one also we will rename it to uh
2:27:42customers.
2:27:50So now our pipeline is uh finally ready
2:27:54and we had already run it. The data was
2:27:57loaded properly. So let's check out the
2:27:59data. Now
2:28:02it is open. Let's check out some sample
2:28:05data.
2:28:07And here we can see the sample data. So
2:28:10if we look at the account open date it
2:28:13is now a proper date column
2:28:16and then rest of the things looks good.
2:28:18We have our silver injection time stamp
2:28:20as well. So now this table it is there
2:28:23in silver layer and ready to be used. So
2:28:26if we come back to our architecture
2:28:29so our table is there in silver layer.
2:28:32So now we are going to join this
2:28:34customer's table with the transaction
2:28:36data that was coming from Kafka. So this
2:28:39Kafka transaction data is streaming
2:28:41data. Our customers table it is the
2:28:43static table. It it is not streaming. So
2:28:46we will join this that is why we call it
2:28:48a stream static join. And then we will
2:28:51build a gold layer with the fraud
2:28:53detection and then send realtime email
2:28:56alerts. So let's work on that piece. So
2:29:00now I'm back in our uh streaming
2:29:04pipeline. So the goal layer we will
2:29:06create it in the streaming part because
2:29:08it it will be streaming data note the
2:29:10static data. So for that we will create
2:29:13a new folder first with the name gold.
2:29:16And then inside there we will create a
2:29:19file.
2:29:25And this file we are going to name it
2:29:30uh high
2:29:33value value
2:29:35transactions
2:29:37alert because it will be used to send
2:29:40alert for the transactions where the
2:29:43value is higher than what is configured
2:29:45by customer. So for this we will start
2:29:49again with by copying the core stuff
2:29:55and this time also we will not need lot
2:29:58of stuff. So we will only need these
2:30:01three. So let me remove these other
2:30:03things. And the table it will be created
2:30:06this time in the gold schema. And the
2:30:09table name we will keep it as same as
2:30:12our file name. So let me copy it from
2:30:15here only.
2:30:20So this will be our table name. And then
2:30:24here we will say that uh
2:30:28uh we can say alert details where
2:30:34uh transaction has been performed
2:30:41with value
2:30:43higher than
2:30:47what is
2:30:51configured by customer.
2:30:54So with this this is our target table
2:30:57and now we will define our function. We
2:31:00will give it the same name and this
2:31:03function it will uh return a data frame.
2:31:13So here we will start by reading our
2:31:16tables. So first we will read our
2:31:18transactions table. So that will be
2:31:21spark dot read
2:31:26dot table. So it is giving some
2:31:28suggestion. I will press tab. I only
2:31:31need this table for now. And then we
2:31:34will read our customers table. So that
2:31:38is our spark dot read. This time we
2:31:41don't need to read that customer's table
2:31:44as streaming table. We need it to read
2:31:47it as a normal table. So that is why we
2:31:50are saying read dot table. And this
2:31:53piece it is actually very important
2:31:56because uh joining a streaming table and
2:31:59another streaming table we need to add
2:32:01some additional details that we will see
2:32:04when we will join the transactions with
2:32:06the streaming data coming from files.
2:32:09But uh the joining a streaming data with
2:32:12a static data that is relatively
2:32:14simpler. So using this keyword read and
2:32:17not read stream that is very important.
2:32:20So this customer table it is our static
2:32:23table and transactions table is our
2:32:25streaming table. So next thing we are
2:32:28going to do is join them. So we will say
2:32:32joined df equal to
2:32:36uh so it is doing some suggestion
2:32:38automatically.
2:32:40So I will press tab for now and then I
2:32:43will remove these extra lines. So here
2:32:46it is saying that we can join the
2:32:48transactions and customers
2:32:50based on their customer ID. So this
2:32:53piece it is fine but there is one more
2:32:56thing we need to add here. So by default
2:32:59any join it is a inner join and we can't
2:33:02do directly a inner join of a stream
2:33:05table and a static table because
2:33:07streaming data it is streaming data
2:33:09right it is continuously coming so we
2:33:12don't know like after a few seconds the
2:33:14record may come which is there in
2:33:16customers table that is why by default
2:33:18we never do a inner join with the like
2:33:22on a streaming table so there is there
2:33:25are ways to do it but uh whenever we
2:33:27join between a stream and static table
2:33:30we directly use a left join. So why we
2:33:34do that? So that every record that is
2:33:37coming in transaction that record is
2:33:40being done as a left join in customers.
2:33:42So for whatever transaction is coming
2:33:45for every transaction it will check if
2:33:47there is a record in customers or not
2:33:50and vice versa it doesn't have to check
2:33:52that is why we use left join here and
2:33:55then we will apply a filter to identify
2:33:58if it is a it is like more than the
2:34:01limit. So for that we will say dot
2:34:04filter and then we will mention that it
2:34:07is f dot column to mention the column
2:34:11name. So the column for amount it is
2:34:14amount and then it is greater than f dot
2:34:18column the column for transaction limit
2:34:22it is the transaction limit. So this
2:34:25column it is there in the transactions
2:34:28table. Uh and this column it is there in
2:34:30our customers table. So that is how is
2:34:33combination now it is able to figure out
2:34:36that the transactions where the amount
2:34:39is more than what is currently
2:34:41configured by customer. So it becomes a
2:34:44fraud and this fraud it is detected by
2:34:48this join and then it will be detected
2:34:51in real time and then this
2:34:53[clears throat] fraud it will be marked
2:34:55and stored into our gold table. So to
2:34:58store this data into gold table we need
2:35:01to specify the columns that we need
2:35:03there in our gold table and in the
2:35:06select we will have multiple columns. So
2:35:09we will type it inside these brackets.
2:35:12So the first column we want to create is
2:35:14a alert type. So for that what we will
2:35:17do is so we will concat things. So for
2:35:21that we have a function uh concat ws and
2:35:26in here we will concat like a dash and
2:35:30there what we want to concat is uh like
2:35:34f dot uh lit lit it means like whatever
2:35:38we specify that same string will be uh
2:35:43will be done there. So now it is saying
2:35:46that we can add transaction ID and
2:35:48generate an alert ID. So I will press
2:35:51tab here and delete this piece for now.
2:35:54So now what it will do is it will like
2:35:57wherever we see this lit lit it means
2:36:00this alert this is a keyword that will
2:36:02be taken as it is. So it will be alert
2:36:04dash the transaction ID and the name of
2:36:08that column alert- transaction ID will
2:36:10be alert ID. So this will become our
2:36:13alert ID. now and then we also need to
2:36:17mention a column for alert type. So we
2:36:20will say f dot lit and then we will say
2:36:24like high value transaction. So it is
2:36:27suggesting automatically. So I will
2:36:29press tab otherwise we can type it. So
2:36:32there will be a column alert type which
2:36:34will always say high value transaction.
2:36:37It is for future use if you want to add
2:36:39more alert types into the same table.
2:36:42And then the next thing that we want is
2:36:44alert timestamp. So we will say like f
2:36:48dot current timestamp and it is saying
2:36:50itself it can be there as a alert
2:36:53timestamp.
2:36:55So current time stamp it will be added
2:36:57as a alert time stamp and after that we
2:37:00will bring in our id columns
2:37:05and for that we will not use this f. We
2:37:07will simply say like transactions dot uh
2:37:12transaction id
2:37:17and then we also want the customer id.
2:37:20So we will say transactions dot customer
2:37:24ID
2:37:26and this why we are mentioning this dot.
2:37:29So because for example customer ID is
2:37:31there on both the tables. So it shall
2:37:34take it from transaction and it will not
2:37:36throw an error that it is ambiguous or
2:37:38something. And next what we want is the
2:37:41email id. So customers table that has a
2:37:44column email like I showed you
2:37:46previously that we want it as an alias
2:37:50saying like customer email. So it is
2:37:53more like meaningful that means like it
2:37:56is the uh email id and this email id it
2:38:00belongs to a customer.
2:38:02And then I will press a tab here and
2:38:05transaction amount I need it here. But
2:38:09before that I need the the name of the
2:38:12customer as well. So for that uh the
2:38:15customer table it it contains the name
2:38:18is first name and last name. So we are
2:38:21concaten concatenating them and putting
2:38:24a space in between them that will be
2:38:27called as customer name. And then we
2:38:30have transaction amount. So we need the
2:38:33transaction limit as well that is there
2:38:35in our uh customers table.
2:38:40So transaction limit it has been
2:38:42selected the alias here even if we
2:38:44delete it it doesn't matter because the
2:38:47name is same only and after the
2:38:50transaction limit we need the
2:38:52transaction currency. So transactions
2:38:55dot currency
2:38:59and then we need columns like
2:39:03transactions dot merchant name so that
2:39:06we can know where it has been
2:39:08transaction has been performed and then
2:39:11we have a merchant category as well. So
2:39:15that also we will brought in here
2:39:20and then we will need some other
2:39:22columns. So let me just copy paste this
2:39:25so that this transactions part I don't
2:39:27have to type manually and looks like
2:39:30intentionally I hit some run shortcut.
2:39:32So it is throwing that error for now. We
2:39:35will ignore it.
2:39:40So the next column it will be
2:39:43transaction type
2:39:46then it will be payment
2:39:49channel and then it will be city
2:39:53then country
2:39:55and then we will have a column like
2:39:57whether it is is international.
2:40:00So that is also very important
2:40:04and then we will need two more columns.
2:40:08So one will be transaction timestamp
2:40:17and another one it will be status. So
2:40:20these are all the columns that we will
2:40:22keep into our gold table so that we can
2:40:26use them while sending alerts.
2:40:29So let me delete this extra space.
2:40:33And here I noticed that we had a
2:40:35transaction date. So we don't need a
2:40:37date because we have a time stamp. So
2:40:39I'm going to delete that piece.
2:40:42And then all these pieces whatever is in
2:40:45the joined df let me put this into a
2:40:48bracket so that it is it is it knows
2:40:51like these are the next lines and
2:40:53everything
2:40:56and then this is what we are going to
2:40:58return and write into our gold table. So
2:41:01we will say return this one. So now this
2:41:07code it is ready. Let's give it a try
2:41:09and let's see what all errors we are
2:41:11going to get.
2:41:13So here is our first error that we got.
2:41:15It looks like there is some syntax
2:41:17related problem. So if I expand this, if
2:41:20I click jump to the code. So for this
2:41:22return statement, it is saying it is a
2:41:25syntax problem. So maybe the indentation
2:41:28it it is not proper probably. So what we
2:41:31will do is simply we will click on this
2:41:34edit button here. It will open this
2:41:36right side and we will say like uh fix
2:41:40the indentation problem
2:41:43of this file
2:41:46because we typed this code manually. So
2:41:49uh so uh some of the pieces are not
2:41:52aligned properly. In Python it is
2:41:54mandatory. So let genie fix that for us.
2:41:58So it has made the changes for us. So we
2:42:00are going to accept it
2:42:04and now let's give it another try. So
2:42:07let's run our pipeline again.
2:42:10So this time the execution has started
2:42:13and it is running. Let me close the
2:42:16right side. Let me bring this into
2:42:18picture. So now we can see the silver
2:42:21bronze transactions is completed. There
2:42:24is no new data. uh silver also completed
2:42:28and high value transaction alert the one
2:42:30gold table we created just now that is
2:42:33also completed but output records is
2:42:35zero and why is that because currently
2:42:38there is no uh fraud data in our uh
2:42:43stream that is coming so that is why
2:42:45there is no data and why is that because
2:42:48when we were pushing the streaming data
2:42:50in Kafka we did not push any
2:42:53uh fraud related or any corrupted data.
2:42:56So let's do that first. So now I'm back
2:42:59in the VS code our producer part and if
2:43:03I press up. So so far as you know we
2:43:05were running this Python normal. py it
2:43:08is sending the normal transaction data
2:43:10note any fraud related data. So to
2:43:13control the fraud related data I have
2:43:15created a separate file that is fraud
2:43:18transaction. So that is the one we are
2:43:20going to run now.
2:43:23So here what we will do is we will
2:43:26mention like FR and it is saying fraud
2:43:29card but not the fraud card. We need to
2:43:32run the fraud transaction. Fraud card it
2:43:35has a different purpose. We may or may
2:43:37not need later on. We will see that
2:43:39part. But for now we want to send some
2:43:41fraud transaction. So now if I press
2:43:44enter.
2:43:47So it has sent that fraud transaction.
2:43:49And the way I have designed this file is
2:43:52that it will send only one record so
2:43:54that we don't end up detect detecting so
2:43:57many frauds and then uh sending so many
2:44:00emails. So for now it will just send one
2:44:03fraud data and then it will exit. That
2:44:06is how it will work. So now we have one
2:44:09fraud uh transaction to work with. That
2:44:12is good enough. So we will go back to
2:44:14our pipeline and simply run our
2:44:16pipeline.
2:44:19So this time hopefully we will get our
2:44:21first fraud. Let's see.
2:44:24So the bronze table it has detected the
2:44:27new record and it has loaded it from
2:44:29Kafka silver also it has been loaded
2:44:32successfully and yes in our uh fraud or
2:44:36alert table we have got our first record
2:44:39now. So let's check out this record. I
2:44:42will click on this view in catalog
2:44:44button.
2:44:46So here is our table. Let's check out
2:44:48that sample or one record that has been
2:44:51created here.
2:44:54So here is our record. This is the alert
2:44:56ID that we created. Alert dash the
2:44:59transaction ID, alert type, the
2:45:02timestamp, transaction ID, customer ID,
2:45:04customer email, the email part like I
2:45:07was uh talking earlier. This is very
2:45:09important. Using this ID, we will send
2:45:12the email. This is the name of the
2:45:14customer, the amount, the limit. As we
2:45:17can see this is the limit set by
2:45:18customer. This is the transaction
2:45:20amount. That is why it has been
2:45:22considered as a fraud. And then there is
2:45:25this additional information that is also
2:45:27useful about this fraud.
2:45:30So now that this fraud has been detected
2:45:33and our gold layer table streaming table
2:45:36will get it in real time. The next step
2:45:38is to send the alert email in real time.
2:45:42So now let's work on that piece.
2:45:45And for sending any custom email uh we
2:45:48have like different uh options that we
2:45:51can use. We can use any of the SMTP
2:45:54services that are available and for this
2:45:57project we are going to use the Gmail
2:45:59one because it is very easy to set up. I
2:46:02already have a Gmail account. So for
2:46:04that what we need to do is we need to
2:46:07simply say like Gmail SMPP
2:46:11API key. So we will need a API key that
2:46:15we are going to use. So this is what we
2:46:18will search.
2:46:20And here
2:46:22uh I don't see that link. I think in
2:46:26Gmail it is called as app key in terms
2:46:28of Google. That's a different
2:46:30terminology.
2:46:32And yes this is the URL. So it is like
2:46:34my account.google.com.
2:46:36Then app password. So this is called as
2:46:40app password not like a API key. It it
2:46:43is kind of same thing but this is Google
2:46:45terminology. So if I click this then we
2:46:49need to select our Google account which
2:46:52we want to use for sending the email. So
2:46:54I have selected my account. Now it is
2:46:57verifying. I will uh cancel it or let me
2:47:02see if it works.
2:47:07Yes, it is working. So now I am on my
2:47:10app passwords screen. So this is where
2:47:13we are going to create a new app
2:47:15password. So here uh we need to give any
2:47:19name maybe we can give it give it the
2:47:22same name fine guard
2:47:28project and then click on create.
2:47:32So this is our API key or app password
2:47:35that has been created. So I'm going to
2:47:38copy it and uh you shall create your own
2:47:41because this one I'm going to delete
2:47:42before I upload this video. So don't try
2:47:45this one. So let me save it to a notepad
2:47:48first.
2:47:51This I have saved here and I will click
2:47:53on done. So now this is ready.
2:47:56So first of all before sending the
2:47:59actual email we will first confirm if
2:48:02the app password that we have created is
2:48:04it really working or not and we will put
2:48:07it in our key uh in our secret scope not
2:48:10the keyword. So let's confirm that for
2:48:14that we are we can use this test
2:48:16notebook that we had created or we can
2:48:19create a new notebook as well.
2:48:24So in this view it doesn't seem to be
2:48:26allowing us. So simply I will select
2:48:28this drop-down and click on workspace
2:48:31because this is the pipeline view. So
2:48:33normally there we can work on the stuff
2:48:35related to uh these the spark
2:48:39declarative pipelines. So now I am back
2:48:42and let me find my project. This is my
2:48:45project and this is our notebook.
2:48:51So here let me bring in some code
2:48:53related to email testing first. So this
2:48:56code I have prepared it previously. We
2:48:59don't need to type it completely because
2:49:01it is a test code that we are going to
2:49:03use. So this notebook I have included in
2:49:06the GitHub link that I have that it is
2:49:08there. You can use this same code. So
2:49:12here simply what we are doing is this is
2:49:14a body that we are preparing. It is a
2:49:17HTML body. So the format is HTML. We
2:49:20don't need to worry about it. It is just
2:49:22sending a test email from datab bricks
2:49:25like that. It is saying then this is the
2:49:27subject of the email that we are we have
2:49:30hardcoded here. And this is where we are
2:49:33getting the API key from the secret
2:49:36scope. So this is the name of that key.
2:49:39So far we have not stored that. So it
2:49:41will not work. So simply what we will do
2:49:44is first maybe let me run this cell so
2:49:47that this piece is ready and for this
2:49:50API key for now we will provide the
2:49:53hard-coded value and later on we will
2:49:55put it in the secret scope. So I have
2:49:58provided this value here and now this
2:50:01one uh this is another piece you need to
2:50:04be careful. So this is my email id. So
2:50:07using which I created the API key. So
2:50:10here you need to replace with your email
2:50:12id then only it will work and this is
2:50:15the Gmail API key this same one that we
2:50:18just pasted here and this is to email
2:50:21where we want to send the email. So this
2:50:24can be the same email ID or a different
2:50:26one. It is simply the ID where we want
2:50:28to send the email. And now I will run
2:50:31it.
2:50:34So it says email sent successfully. So
2:50:37let's confirm if we have received the
2:50:39email.
2:50:40So if I check my email, this is the
2:50:42email that I have received. This is the
2:50:44from email id. This is the to email id
2:50:48and this is the test email. So if we
2:50:51have received this email, what it means
2:50:53is that our API key is working fine and
2:50:56this sending of email as a testing it is
2:50:59working fine. So we can use this
2:51:01functionality now. So now instead of
2:51:05like again providing this code to our
2:51:08streaming pipeline, I'm going to copy
2:51:11this code as well. But before doing
2:51:14that, we will put this API key in secret
2:51:16scope as well. So that we follow the
2:51:19best practice. So I will copy it from
2:51:22here. And let's go to the setup secret
2:51:24scope.
2:51:26And then if we go below, so this is the
2:51:30last piece that earlier I was talking
2:51:32about.
2:51:34So it will be in the last cell. So this
2:51:37is the name of the key and this is the
2:51:39secret value. So I'm going to replace it
2:51:42with the new one that I just uh created.
2:51:46So now let's store it. Uh let's see if
2:51:49it is running or I need to run above
2:51:52cells. Probably I will need to. Yes,
2:51:55secret scope name is not defined. So
2:51:58this is the secret scope name that we
2:52:00have.
2:52:02Now let me run it.
2:52:11So now it is saying datab bricks
2:52:12instance. So yes, here it is. So let me
2:52:16run this cell. Then this one is about
2:52:19creating the scope. So probably this
2:52:21time it will work.
2:52:25Yes, now it says like this has been
2:52:27created successfully. So now to confirm
2:52:30if this is working through secret scope.
2:52:33So I'm going to delete this cell that is
2:52:35has hardcoded value and I'm going to run
2:52:38this cell. So now it has read this API
2:52:42key from this secret scope. And let's
2:52:45give it another try by sending the same
2:52:47email again. And it says it is
2:52:50successful. So the same email has been
2:52:52sent again. And this confirms that this
2:52:54API key from the secret scope is working
2:52:57fine. Now to integrate this piece, I'm
2:53:00going to copy this code. This one cell
2:53:03of the code is enough. So I have copied
2:53:05it. Why? Because we are going to use
2:53:08Genie code here to integrate this
2:53:10functionality into our pipeline. So this
2:53:13is our streaming pipeline. So I will
2:53:15open in open it in the pipeline editor
2:53:18now. And with this we are back in our
2:53:21pipeline.
2:53:24So here uh now what we are going to do
2:53:27is we will use genie code and we will
2:53:31need to give it instruction. So let's
2:53:34give that. So what we will provide there
2:53:37is uh probably let's type it in a
2:53:40notepad so that it is easier to
2:53:42understand and then we can paste it
2:53:44there. So because I have copied the code
2:53:47here. So let me first paste that code so
2:53:50that I don't lose it and then we will
2:53:53say that this is the table that contains
2:53:56our alerts.
2:54:00Okay. So we will say that the table
2:54:04this one
2:54:06created by this table has been created
2:54:09by this file. So let's mention this also
2:54:12so that it can read the file if
2:54:14required.
2:54:16You will say contains the
2:54:20fraud related information where
2:54:23transaction has been
2:54:26performed
2:54:28with higher value than provided by
2:54:34customer.
2:54:37And then what we will say is that like
2:54:40this
2:54:42table contains
2:54:45information about
2:54:47transaction and other columns
2:54:51along with an email column. So we will
2:54:55mention that
2:54:57I want to send email to this ID in real
2:55:04time.
2:55:07by reading data from the table
2:55:12in in real time or as streaming.
2:55:17And then we will say that for sending
2:55:21the email
2:55:23I want
2:55:26to use the below
2:55:29uh sample code.
2:55:33So integrate
2:55:35this code and then we have our code. So
2:55:39what it will do is we are giving it the
2:55:41requirement what we want to do and we
2:55:44are also giving it this sample code
2:55:46saying that uh this is the example or
2:55:49sample code that we want to use to send
2:55:52the email. So now let's me let me copy
2:55:55this entire thing. Let's go back there
2:55:57and let's ask Genie code to do it.
2:56:02So now it is thinking. Let's see how
2:56:04much it is able to do.
2:56:09So it has prepared some thoughts. It is
2:56:11currently reading the pipeline details
2:56:15and now it is considering different
2:56:17things.
2:56:20And now it is creating a separate file
2:56:24looks like with the name notify.
2:56:27So looks like it has done its
2:56:29processing. So let's try to understand
2:56:31that. So here is the file it has created
2:56:35with the name email notifier
2:56:38and first let's accept these changes so
2:56:40that it is easier to read.
2:56:43So how it is saying is that what it does
2:56:45it reads the streaming data from this
2:56:48table. It processes each micro batch in
2:56:51real time as new alert arrive and then
2:56:55it send the email to the customer. So
2:56:58here we can see it has created multiple
2:57:00functions here. So like we had dp table
2:57:05and here it is using like a different
2:57:07function like dp dot for each batch sync
2:57:12and then we will find another function
2:57:15here at the below. So it is about append
2:57:18flow. So it has created two functions.
2:57:22So let me ask it to like simplify this
2:57:25so that we can understand it in little
2:57:27bit little bit better way and also what
2:57:30it is saying is that next steps it is
2:57:32saying add your SMTP code. So this code
2:57:35it has provided and it is asking us to
2:57:38add here and uh
2:57:42this Gmail API key also it is reading
2:57:44from the session because we didn't uh
2:57:47provide the code for the secret scope.
2:57:50So we will provide that and uh currently
2:57:54uh first let me explain like what it is
2:57:57working and how we can make it simple.
2:58:00So the flow it starts from here. So it
2:58:04says append flow. So this is a standard
2:58:07function or decorator. What it will do
2:58:10is whatever stream it is reading from
2:58:13this function it will send that stream
2:58:15to this target. So the stream here this
2:58:19is our table. So it is going to read
2:58:21this table streaming data. Every record
2:58:25or couple of record is a micro batch
2:58:27that it will get it will read from here
2:58:30and send to this target. So that is what
2:58:33is this append flow. So append means
2:58:35keep reading more data and keep sending
2:58:38to this sync. And where is this sync is
2:58:41that uh here is our sync. So the data or
2:58:45the micro batches will be coming from
2:58:48from there to this function and it will
2:58:51be in micro batches. It can be single
2:58:53single record or a like a couple of
2:58:56records in one group also can come up.
2:58:58So that is where we have this four each
2:59:01batch sync function. So for each of the
2:59:03batch let's say in one batch we got two
2:59:06three alerts in a single uh batch. So
2:59:09then it will perform that operation. So
2:59:12automatically when the data will come
2:59:14there it will it will be received as a
2:59:17data frame and it will have a batch ID
2:59:20also. So that is internally
2:59:22automatically that will work and for
2:59:24this batch this data frame. So this is
2:59:27the email part and then uh from that
2:59:31data frame it is collecting all the
2:59:33data. Then each and every row it is
2:59:36processing here and then it is sending
2:59:38the email by preparing the email body
2:59:41and everything. So first let's ask it to
2:59:45like make some changes so that the code
2:59:47looks bit clean. So we are going to uh
2:59:51open our send email file again. Let's
2:59:53see this time it is opening here.
2:59:57And we will tell that API key is
2:59:59available here. So here we will say
3:00:02point number one that take the API key
3:00:06from here.
3:00:09And second point we will say I want
3:00:15to segregate the email sending part from
3:00:21for each
3:00:23batch. So create a new function for
3:00:27this.
3:00:32So it shall be able to do that
3:00:34automatically. So let's open that file.
3:00:37Let's wait for it to do those changes.
3:00:40And now what it is saying is it is
3:00:42creating the file and it is saying it
3:00:44will create a placeholder for the SMTP
3:00:46code. So why it is doing this is it is a
3:00:49safety measurement from the uh from the
3:00:53genie code side because it don't want to
3:00:55call external APIs automatically. So for
3:00:58now what I will do is I will stop it
3:01:00here and I will say that I have tested
3:01:04SMTP and it is safe.
3:01:08Usel integrate
3:01:13complete code. So that it it is
3:01:16confident that the code is fine to
3:01:18integrate
3:01:19and now it is saying that user has
3:01:21confirmed. So now it will integrate that
3:01:24code completely.
3:01:26So now what is happening is it has
3:01:29provided the code and it is saying that
3:01:31you shall we shall manually copy paste
3:01:33the code because due to this external
3:01:35API its internal mechanism it is not
3:01:38allowing it to uh replace the code. So I
3:01:41will copy the code from here and then I
3:01:45will select everything and paste it.
3:01:50So here I don't know why it is showing
3:01:51red. So I will reject this piece here.
3:01:58And then again I will delete it and
3:02:00paste one more time. And now the code
3:02:03shall be complete. So let me close the
3:02:06right part. Let's understand the code
3:02:08one more time. Hopefully this time it
3:02:10will be simpler. So this is the same
3:02:13append flow. So it is reading our
3:02:15streaming data as alert and sending to
3:02:18this target email notifier sync. And
3:02:22then yes sync part it is simple now. So
3:02:25this is our sync and in the sync it is
3:02:29receiving the data frame and the batch
3:02:32id. This is the from email id which is
3:02:34hardcoded because uh using this email id
3:02:38only we generated our app password. So
3:02:40from id will be this only. The app
3:02:43password it is getting it from the
3:02:46secret scope. So it is doing it uh
3:02:50through like a difficult way. I would
3:02:52say it this code looks strange to me. So
3:02:55what I will simply do is I probably it
3:02:58will work but I will simply remove this.
3:03:00I have not seen this earlier. This way
3:03:02it works fine.
3:03:05Then uh it is collecting all the rows
3:03:08from our data as rows and then it is
3:03:12doing some processing row by row. So for
3:03:15each of the row in the from the table
3:03:17whatever it is reading it is first
3:03:20preparing a dictionary like the alert
3:03:23ID, customer name, transaction amount
3:03:25all these things it is reading
3:03:28and then it is uh it is like uh
3:03:32preparing a subject. So this is the
3:03:34subject that we will get when we get the
3:03:36alert and in the body it is calling a
3:03:39function like create alert email body
3:03:42and it is passing this dictionary with
3:03:45all these details that is where it has
3:03:47like now more modular code so that we
3:03:50don't see a complex code for this email
3:03:53and get confused that part we will go
3:03:55through but first let's say here it is
3:03:57creating the body and then it is sending
3:04:00the email and it is sending the email by
3:04:03passing the email id, subject, body,
3:04:06from and the app password. So then it is
3:04:10saying how much how many emails it has
3:04:12sent successfully and how many has been
3:04:14failed so that it can print it. So this
3:04:17is how it is sending the email and to
3:04:20create the body part it has created a
3:04:22separate function that we can see here
3:04:26and send email also it is a separate
3:04:28function. Earlier I thought it is the
3:04:30Gmail function but for the body part it
3:04:33is receiving that dictionary and from
3:04:36that dictionary it is reading all that
3:04:38different data and preparing a HTML well
3:04:41formatted message. So what exactly is
3:04:44this we will see when we get the email
3:04:46we don't need to understand each and
3:04:48everything here because this is HTML if
3:04:50it looks good it will we will be fine
3:04:53with it. If you want to make any
3:04:55changes, we can ask Genie and it will be
3:04:57able to make changes for us. So that is
3:04:59how it is preparing this email body. And
3:05:02then this send email function which is
3:05:04the final function. So this is the
3:05:07function that is there. And here we are
3:05:09getting all those details and this is
3:05:12where the the email code that we had
3:05:14provided it is integrating that. So it
3:05:17is calling the Gmail SMTP server and
3:05:20then finally sending our email. So that
3:05:23is how this code has been integrated. So
3:05:27overall if we understand the flow. So it
3:05:30will read the streaming data. It will uh
3:05:33send the micro batches to another
3:05:35function that from those microbatches it
3:05:38will read rowby row data. It will
3:05:41prepare subject body and all and then it
3:05:44will send the email. That is how the the
3:05:46flow looks like. And for this piece what
3:05:50we are going to do is uh because it is
3:05:52not creating any table it is sending an
3:05:54email. So for that purpose we will
3:05:57create a separate folder for this one so
3:06:00that we manage it separately. So we will
3:06:02name it as alert. So let me create this
3:06:05folder and let me move this file inside
3:06:09that folder.
3:06:14So by default it is not included as
3:06:17execution. So let me include this folder
3:06:19first so that the file will get executed
3:06:23when we do that. So yes now it is
3:06:26included and it will run this file when
3:06:29we run the pipeline. So with this our
3:06:32code is ready. So let's try to run it
3:06:35and let's confirm if it is working fine.
3:06:38The expectation is that this one record
3:06:40that it had written that same record it
3:06:43will call this file and send the email
3:06:46notification. So let me run it.
3:06:50The execution it is complete now. And
3:06:54looks like like this email notifier sync
3:06:56it worked well and output record is one
3:07:00which means it has probably sent our
3:07:02email. And if you notice so it has not
3:07:05received any new transaction. So other
3:07:08pieces it was not able to process
3:07:10anything just the email part it
3:07:12processed one record because this alert
3:07:14it had it and if you want to see the end
3:07:17to end flow we can click on this
3:07:19pipeline graph. So here we can see all
3:07:23the steps that first the table was
3:07:25loaded into bronze then silver then gold
3:07:28and then the sync part which is for each
3:07:31batch that sent the email and it will
3:07:35look more beautiful when we will combine
3:07:37another stream to it the file streaming
3:07:39later on for now let's confirm if the
3:07:42email has been received properly
3:07:45so if I go to my email id so this email
3:07:49probably it will not be coming to to
3:07:51your inbox, it will go to spam because
3:07:54this uh this Gmail it will know it will
3:07:58see it that coming from a personal email
3:08:01id and saving saying about transaction
3:08:03fraud and all. So this one it will
3:08:05probably go to spam. So check it out
3:08:08there if you are trying it out. So this
3:08:10is where we have received this email and
3:08:13it says like we can report it as not
3:08:16spam but for now I will leave it here so
3:08:18that it doesn't go to my main inbox. But
3:08:22this email has been generated saying
3:08:24that like we detected a transaction that
3:08:27exceeds your configuration limit. These
3:08:30are all the details
3:08:32and then it is saying like if you
3:08:34authorized then no problem otherwise you
3:08:36shall report the authorities and or and
3:08:40all that immediately.
3:08:42So this is how the real time alert looks
3:08:45like. We'll see it working end to end
3:08:47once we configure the other stream and
3:08:49then run everything end to end. So now
3:08:52if we revisit our architecture
3:08:55we are done with this stream static
3:08:57join. We are done with the basic gold
3:09:00layer. Uh although we will be building
3:09:02more into gold pro detection one we have
3:09:05done and then real time alert also we
3:09:08are able to send now. So the main pieces
3:09:11pending now is like bringing in the
3:09:13streaming files for injection through
3:09:15autoloader and do do a stream stream
3:09:18join and then detect another kind of
3:09:20fraud and that also we will send
3:09:22realtime alerts and later on we will
3:09:25build a dashboard as well and at the end
3:09:28we will do end to end orchestration
3:09:30where we will see everything flowing in
3:09:33real time end to end. So now let's get
3:09:36started with this uh streaming piece.
3:09:39But before that I would say like let's
3:09:41use this point as another checkpoint and
3:09:44you take some break you understand it
3:09:47completely how it is working you do some
3:09:49hands on and see it working understand
3:09:52properly then you will be ready to
3:09:54integrate the second stream. So once you
3:09:58are ready then you can uh start again
3:10:00from here. So now let's start with the
3:10:03streaming files part. So to uh to
3:10:07process streaming files, first thing we
3:10:10need is streaming files, right? We need
3:10:12some place where we will get streaming
3:10:14files. So first let's prepare that
3:10:17place. So in our sources, we already
3:10:20have this transactions folder or volume.
3:10:23It is a volume. Here we will create
3:10:25another volume.
3:10:28And this one let's name it as fraud
3:10:31watch list.
3:10:34And why we are choosing this name is
3:10:35because here what we are assuming is
3:10:38that we will be getting some files which
3:10:41are being generated from a system
3:10:44uh which are also like uh monitoring the
3:10:47customer and card accounts. And whenever
3:10:50customer reports that the card shall be
3:10:52blocked or if the card is blocked
3:10:54automatically due to some other fraud
3:10:57related activities then what all cards
3:11:00and other information is being blocked
3:11:02that information we are receiving
3:11:04through these files and if any card
3:11:08let's say if it is marked as that it is
3:11:10blocked or something and if it is marked
3:11:12for fraud for example for this card we
3:11:15shall not be getting any transactions
3:11:17right. So if still we are getting any
3:11:20transaction for that that will be a
3:11:22fraud and that is what we will be
3:11:24detecting through this stream.
3:11:27And here looks like I did a typo. Let me
3:11:29correct it first. So let me rename it.
3:11:36And here first we are going to create a
3:11:38folder where we will be receiving the
3:11:40file. So we will name it as source data.
3:11:45So this is where we will be generating
3:11:47some files and then uh those files we
3:11:51will be processing in in uh real time.
3:11:55And to generate these files I have
3:11:57prepared a notebook that will generate
3:12:00the files automatically and keep sending
3:12:03to this location in real time.
3:12:06And that notebook I have kept in this
3:12:08folder that I will be uploading in
3:12:10GitHub. Inside this folder there is a
3:12:13CSV file and then this is the notebook
3:12:16that will generate that data
3:12:19and in this notebook there are two path
3:12:21that we need to update or confirm if
3:12:24those are fine. One is the path of this
3:12:27CSV file. So it contains some CSV data
3:12:30that it will read one by one line by
3:12:32line and it will prepare JSON file and
3:12:35put it put it into that location. So we
3:12:38need to copy the full path of this file.
3:12:42So I have copied and here we need to
3:12:44provide the full path. So this is the
3:12:47full path and then this is the output
3:12:50path where it will be generating the
3:12:52files. So this path it is the folder
3:12:54that we created just now. So from here I
3:12:57will copy this path and I will update
3:13:00this. So only these two paths we need to
3:13:03update and then we will be good to go.
3:13:06So now I will run this notebook to
3:13:08generate those JSON files
3:13:13and what it will do is it will send a
3:13:16single JSON file for a single record and
3:13:19it will keep sending files with an
3:13:21interval of 5 seconds. So it will send a
3:13:24file wait for 5 seconds then send
3:13:26another file. That is how it will
3:13:29process the data.
3:13:31So currently we can see that one file it
3:13:34has written it is waiting for 5 seconds.
3:13:38Then a second file it has written it is
3:13:40waiting for 5 seconds. So that is how it
3:13:43will keep sending more and more files
3:13:45there. So for now we will interrupt it
3:13:48so that we have some sample files that
3:13:50we will work upon. Later on we will run
3:13:53it again when we will run our entire
3:13:55process end to end. So now if we go to
3:13:58the location that we had specified and
3:14:01if we refresh there so we will see that
3:14:04four files has been created. So if I
3:14:07open any of the file here uh it contains
3:14:10a JSON data and this data it contains a
3:14:13watch list ID the watch type that it is
3:14:16monitoring card and this is the entity
3:14:19ID which is the card number
3:14:22and for this one risk level is high. The
3:14:25reason is that it is for it has been
3:14:28marked as fishing card. So that way like
3:14:31some other details are also being
3:14:33generated here. So this is the data that
3:14:36we will be ingesting now. So these are
3:14:39the uh this is our streaming source now
3:14:41where new files keep coming and to
3:14:44ingest this data we are going to use
3:14:46autoloader.
3:14:48And to use the autoloader also again we
3:14:50have two options either we can write it
3:14:52in a normal notebook or we can write it
3:14:55into our uh uh spark declarative
3:14:58pipelines. So first we will understand
3:15:00in a notebook so that we can understand
3:15:02the functionality of autoloader first
3:15:05then we will integrate it into our uh
3:15:07SDP pipeline. So for that I will copy
3:15:10this path
3:15:12and then I will go to our project
3:15:14folder. We will go back. So this is
3:15:18where we will create that file. So we
3:15:20will create a new notebook here
3:15:24and we will name it as 04
3:15:29autoloader test.
3:15:32So first of all we will create a source
3:15:35path variable here. That will be the
3:15:37path that we have copied just now.
3:15:43And after creating this variable first
3:15:45let's confirm if it is showing the files
3:15:48here. So it is like dbutils dot fs do ls
3:15:54ls command it shows the list of files
3:15:56available into this path. So as this is
3:16:00the variable we don't need to provide
3:16:01the quotes here. Let me run it. So here
3:16:05it is showing that those files are
3:16:07available here. So those four files are
3:16:09showing. So the path is correct.
3:16:12So now we will create a data frame. So
3:16:15we will say like input stream equals to
3:16:20then we will do spark dot read stream
3:16:26and then it is suggesting some options.
3:16:29So I will press a tab here and here.
3:16:33This option we do not need. First let me
3:16:36explain these options. So format cloud
3:16:39files it means we are using autoloader.
3:16:42So cloud files it is the keyword that is
3:16:45used to specify that we are using
3:16:48autoloader that is the standard name.
3:16:50And then we have more options which are
3:16:52specific to autoloader like the option
3:16:55for format. So cloud files dot format
3:16:59our file format is JSON. So that is why
3:17:01we are specifying this. Then we need to
3:17:04provide a schema location. So at this
3:17:07location it will store the current
3:17:10schema of the file so that in future it
3:17:12can compare and evolve the schema if
3:17:15required. So here we need to provide a
3:17:17path where it can store the schema
3:17:20location. So it will be similar to our
3:17:23checkpoint location. So what we will do
3:17:26is so this is where the source data is
3:17:28coming. So here we will create a
3:17:31directory with the name schema and then
3:17:33this we will use as schema location
3:17:37and after that we will provide another
3:17:40option to infer the data types of the
3:17:43file uh so that we don't have to uh
3:17:47specify the uh schema of all the
3:17:50columns. So we will specify
3:17:55infer
3:17:57column types
3:17:59is true
3:18:02and these are all the options that we
3:18:04will need and then finally dot load will
3:18:07be there
3:18:09and in here we need to specify the path
3:18:11that we are going to read.
3:18:15So this is how like we read a we create
3:18:18a stream using autoloader. And here one
3:18:21thing to notice is that when we were
3:18:24doing a Kafka stream test that time
3:18:26first we tried it with only read. We try
3:18:29to read the batch data uh and then we
3:18:32tried the stream part. But in autoloader
3:18:35the normal read operation it is
3:18:37currently not supported. So it only
3:18:39supports streaming way of injection. And
3:18:42this streaming way also we can use it
3:18:44kind of a batch injection as well when
3:18:46we use the trigger type as available
3:18:48now. So it can work in both scenarios.
3:18:51So this will create our input uh
3:18:54streaming data frame.
3:18:57So it is ready now. Next we will do some
3:19:00minor transformation.
3:19:04So let's say transformed
3:19:07df. So here what we will do is
3:19:12uh we will take this data frame and
3:19:14specific to autoloader or anything
3:19:16whenever we read any file in spark uh
3:19:20like outside autoloader outside
3:19:22streaming anywhere then uh we get some
3:19:25additional file related properties also
3:19:28those we can read. So let's say in here
3:19:31select what we will do is so all the
3:19:34existing column we will select as it is.
3:19:37So we will say star
3:19:39and then we will add any additional
3:19:41column. So f dot call and then there is
3:19:45a specific keyword like underscore
3:19:49meta data. So it is coming automatically
3:19:52dot file path and alias is file path. So
3:19:55this will add a new column file path
3:19:58which will store the complete path of
3:20:00the file. So this keyword it
3:20:03automatically get that data. So we have
3:20:05other things also like modify time, file
3:20:08name, whatever we want we can get it.
3:20:11And then also we will get the current
3:20:14timestamp as well. So this one we will
3:20:17call it as injection time stamp.
3:20:26So this one will be a comma here
3:20:30and that's it. So this is the data frame
3:20:33that we are going to write. And as you
3:20:36may have guessed this f is currently not
3:20:39available. So we will need to import it.
3:20:43So from
3:20:46so it is giving it automatically. So we
3:20:48will import it. So this will be our
3:20:51transform data frame and now this one we
3:20:54will need to write it to a table and for
3:20:57that let's copy our code that we used in
3:21:00Kafka streaming because it will be
3:21:02almost the same only.
3:21:06So uh yeah this is the code that we had.
3:21:10Let me copy this
3:21:15and if we go back here.
3:21:17So here this is the transform df. So I
3:21:20will replace it and it will write as a
3:21:24delta table. It will write in append
3:21:26mode and then we need to provide a
3:21:28checkpoint location. So this location we
3:21:31will create another one. So for fraud
3:21:34watch list let's create a directory here
3:21:37which will be called as checkpoint.
3:21:40And at this checkpoint location it will
3:21:42keep track of the files that it has
3:21:44processed in the current run. So that in
3:21:47the next run it pick up uh it it doesn't
3:21:50pick up those files again and it only
3:21:52pick ups the new files.
3:21:54So that is the checkpoint location we
3:21:56will update here. Trigger type so we
3:21:59will use available now only so that it
3:22:02will only run once and then exit and two
3:22:05table. So this table name we will name
3:22:07it as uh fraud
3:22:11watch list
3:22:13then batch test. So this is the name we
3:22:17will provide.
3:22:19Now I'm running it. So it will read
3:22:22those files and write the data into this
3:22:25table.
3:22:27So now the execution it is complete.
3:22:30Let's check out the data in this table.
3:22:41So here is our data. All those four
3:22:43records has been read here.
3:22:46all the columns has been created
3:22:47properly and this is the file path that
3:22:50I was talking about. So for each and
3:22:52every file it is putting the complete
3:22:54path of the file using that underscore
3:22:57metadata dot file path. So this is how
3:23:00autoloader works. So writing part it is
3:23:03exactly same reading part we need to
3:23:05provide this cloud files and cloud file
3:23:08specific options to read the data as uh
3:23:12streaming using autoloader.
3:23:14So now as we understand this using uh
3:23:17normal batch data or through this
3:23:20notebook now let's use the same in the
3:23:23spark declarative pipeline. So for that
3:23:25we will open our same streaming pipeline
3:23:30and here let's go to branch. So this one
3:23:34what we will do is we will clone it so
3:23:36that some of the pieces will be same. So
3:23:39we are going to name name it as
3:23:42road
3:23:46watch list
3:23:50then branch. So this way we are going to
3:23:52clone it.
3:23:58So in here we will not need everything.
3:24:00So we will remove these pieces. The
3:24:03table name we will create it as fraud
3:24:05watch list.
3:24:09And this one it will be like
3:24:12road watch list raw data ingested by
3:24:16autoloader.
3:24:19And this function also we can rename it
3:24:22to fraud watch list branch.
3:24:25It will return a data frame and all
3:24:28these things we do not need. So this
3:24:32streaming df creating this it is simple
3:24:35in case of autoloader. So we will simply
3:24:38copy it from the notebook that we
3:24:40created just now. So let me open that.
3:24:45So this is our streaming DF now. So let
3:24:48me uh read it. So I have copied it. Now
3:24:54let's take it to the streaming df here.
3:24:58So I have replaced it now.
3:25:03And here the source path it is also not
3:25:06available. So directly we will put this
3:25:09path there. So I will copy this and
3:25:13directly put this into this code. And
3:25:16one more thing so this schema location
3:25:18also now as we are using SDP pipeline so
3:25:22it will be maintained automatically. So
3:25:25we don't need to specify this option. So
3:25:27this is one advantage that we get. Rest
3:25:30of the things will be same. So now we
3:25:32have our streaming df and next what we
3:25:36need to do is from the streaming df we
3:25:38need to select the columns that we need.
3:25:41So in our case now we have imported f.
3:25:44So this is another way to specify those
3:25:48and this current timestamp we will still
3:25:50keep it. So uh let me put the other
3:25:54columns that we are going to keep here.
3:25:56So
3:25:58uh let me simply copy paste it so that
3:26:01we can paste those columns. So the
3:26:03columns that we have now are like
3:26:08watch list ID
3:26:10and then we also do not need casting
3:26:13here because the columns are already
3:26:15string. So let me paste this one.
3:26:24And then we have watch list type.
3:26:28And then we had entity
3:26:32ID. Entity type. Sorry, entity type is
3:26:36not there. So let me remove this
3:26:40or I will just use another column that
3:26:43is risk level. Then we have action
3:26:47and then we have reason code, reason
3:26:50description. After that we need a status
3:27:03and then we need effective from
3:27:08and after that we need reported by
3:27:11reported
3:27:13uh source.
3:27:18After that we have country and then city
3:27:26and then we have something special which
3:27:28is rescue data. This one I forgot to
3:27:32talk about. We will talk about it. So
3:27:35this column what it is? It is a column
3:27:37added automatically by autoloader. So
3:27:40the table which was created just now
3:27:43using autoloader. So this is the column
3:27:45like underscore rescued data. So this
3:27:49column uh it is created by autoloader
3:27:51where it will store any data that is
3:27:54corrupted. So in JSON if we get any
3:27:56corrupted data which doesn't fit
3:27:58correctly into these columns. So that
3:28:00data it will put into rescued data. So
3:28:03this functionality is provide provided
3:28:05automatically by autoloader.
3:28:09So it is rescued data and now we are
3:28:12left with only one
3:28:15uh which is our file path.
3:28:20So that one was underscore metadata
3:28:26dot file path
3:28:28and this one we will give it a alias as
3:28:31well. We will name it as source file
3:28:36so that we know it is a source file
3:28:38path.
3:28:41And then we will keep our current time
3:28:43stamp as injection timestamp also. And
3:28:47you may ask like why did we provide this
3:28:49everything in a specific like by putting
3:28:52it one by one. So the reason for that is
3:28:56so even though in our file we had things
3:28:58like watch list ID type in this order
3:29:01but it read this entire JSON and the
3:29:04order of these columns is not
3:29:06guaranteed. So when this table was
3:29:08created so if you look at it it has like
3:29:11action city country so the column order
3:29:14is different that is why we specified
3:29:17each and every column so that we get
3:29:19these columns in the same order.
3:29:22So now this is our parsed streaming df
3:29:25that we are returning and this will be
3:29:28written into our branch table. So with
3:29:30this our brunch uh this uh pipeline it
3:29:34is ready now. So let's run it and let's
3:29:37see this table created. So now I will go
3:29:40ahead and run this pipeline
3:29:46and while this is running so the other
3:29:48thing to notice is that here also for
3:29:50writing part we did not have to specify
3:29:53any checkpoint location because it is
3:29:55taken care automatically by the SDP
3:29:58pipelines.
3:29:59So it has been failed saying like no
3:30:02such field is there file path. So we
3:30:04forgot to put a underscore there. Let me
3:30:07put it now.
3:30:09So here we need to provide it. Now let
3:30:12me run again.
3:30:15And now we have a second error which
3:30:16says like watch list type is not there.
3:30:19So this is actually watch type. It was a
3:30:22typo. Let me correct it.
3:30:26And this time I missed the spelling
3:30:28effective from. So let me correct it.
3:30:33And the thing is it will not tell each
3:30:35and every error in one go. So we have to
3:30:38fix them one by one. And this reported
3:30:41source also looks like I made a spelling
3:30:44mistake. So sorry for all these typos.
3:30:47Uh let me correct them.
3:30:50So now we can see it has written those
3:30:52four records and it is currently running
3:30:54in the continuous mode because after
3:30:57last time we did not change it to
3:30:59trigger type. So I will stop it manually
3:31:02for now.
3:31:03And then I will change the trigger type
3:31:05to triggered so that for the development
3:31:07purpose we can run it only once.
3:31:13So now our table it is ready. Let's
3:31:15check out our branch table for fraud
3:31:17watch list. So here is our table. Let's
3:31:21check out our sample data.
3:31:24So after working for a few minutes looks
3:31:26like the compute is currently not
3:31:28available. It is very busy. So this is
3:31:30serverless type. So what I will do is I
3:31:33will try to copy it in a notebook where
3:31:36the
3:31:38uh serverless like which is not SQL
3:31:41compute the normal serverless compute is
3:31:43used. So there let's see if it is
3:31:45working fine. So we will try to query
3:31:48this table
3:31:51and yes here it is working. So if you
3:31:54also encounter this kind of problem you
3:31:56can try the normal serverless compute.
3:31:59So here it is the data is coming
3:32:01properly. We can see that the watch list
3:32:04ID display level action. So action is in
3:32:07upper case these are in lower cases. So
3:32:09we will convert them to upper cases when
3:32:12we go to silver layer. And then this
3:32:14effective from column currently it is a
3:32:17string. So we will convert it to
3:32:20timestamp when we go there. Other
3:32:22columns those are created properly.
3:32:25So now is the time to load this
3:32:28streaming table into silver layer.
3:32:31For creating the silver uh notebook or
3:32:34silver silver file for fraud watch list,
3:32:37we will simply clone its branch because
3:32:40there are only very few things we will
3:32:42need to do there. So we will clone it in
3:32:44the silver part.
3:32:55So this is the clone file
3:33:00and we will name it as fraud watch list
3:33:02silver.
3:33:08And here the name we will change to
3:33:12silver schema so that it will create in
3:33:14schema silver part. Here we will say
3:33:17like we can say cleaned
3:33:21uh
3:33:22fraud watch list that is the only thing
3:33:25we will give here
3:33:28and then this one also we will rename it
3:33:31to silver
3:33:34and for this streaming DF so we don't
3:33:37need to like
3:33:39uh provide all these things
3:33:42because here we will be reading a table
3:33:44so we will simply
3:33:46provide an option uh dot table and we
3:33:50will read our branch table here. So it
3:33:53is providing as an suggestion as well.
3:33:55So this is our streaming df now or we
3:33:58can name it as branch df
3:34:03and now in this branch df we will say
3:34:07like
3:34:08uh cleaned df or standardized df we can
3:34:12give it any name it doesn't matter.
3:34:15And then
3:34:17uh we will make some standardizations
3:34:19like this watch list. So we need it in
3:34:22the upper case. So we will change it to
3:34:25upper case.
3:34:27And apart from this watch list ID.
3:34:31Uh this entity ID,
3:34:34risk level and action these also we will
3:34:38change them to upper case.
3:34:46So it is done and then current time
3:34:49stamp we will keep it as silver
3:34:52injection timestamp
3:34:54and then rescue data we normally don't
3:34:57bring it into silver layer because it's
3:34:59uh role it is until brunch for debugging
3:35:02and everything and then the existing uh
3:35:05injection timestamp column. So that one
3:35:08we will name it as bronze injection time
3:35:11stamp.
3:35:36So it is done and now we need to make
3:35:39one final change that is about the
3:35:41timestamp column. So if you look into
3:35:44the file that we are ingesting. So this
3:35:47effective from column it is coming in
3:35:50this format. So I will copy this. So
3:35:52this format currently it is being
3:35:54ingested as a string only. We will need
3:35:57to convert it to the actual timestamp
3:35:59column. So this one we will select it
3:36:03and then I will press control I. We will
3:36:06say convert this string
3:36:10from this form
3:36:15to the actual time stamp
3:36:20and it has done it for us. So I will
3:36:22accept this change. So this is also
3:36:25casted properly.
3:36:27And now this cleaned df this is the one
3:36:29we want to return and write it into our
3:36:32silver table. So now let me go ahead and
3:36:35run our pipeline again.
3:36:42So now it failed saying that there is
3:36:45some invalid character. So let let us
3:36:48ask Genie to fix this error. So we will
3:36:51simply click diagnose error. It will
3:36:53check where is the problem and it will
3:36:55fix it.
3:36:57So now it has updated these things. It
3:36:59is trying to do a dry run. So I will
3:37:02accept these changes
3:37:04and here the fix it has done is it has
3:37:07put alias. So looks like this upper
3:37:09function it works only with alias
3:37:12and now I will stop it and the next
3:37:15error that we have is about this
3:37:17metadata file path. So now because this
3:37:20uh column file path itself it was there
3:37:22in the branch. So underscore metadata
3:37:26will not be available. So the actual
3:37:28column name will work now. So this time
3:37:31it is working fine.
3:37:34And here we can see that in our silver
3:37:36table it has written four records now.
3:37:39So now let's check out these these four
3:37:42records. For that we will go to a
3:37:44notebook and then just try to query it.
3:37:54So we will say fine guard dot silver
3:38:01and then dot fraud watch list.
3:38:05So here we can see that those four
3:38:08records has been written properly. The
3:38:10effective from column also it has been
3:38:12converted to a proper timestamp column
3:38:15and also the watch list ID the watch
3:38:19type risk and action these columns also
3:38:22has been changed to upper case now and
3:38:24rest everything also looks fine. So with
3:38:28this uh our fraud watch listed table
3:38:31this is ready in the silver layer and if
3:38:34we look at the pipeline graph now so we
3:38:38can see fraud watch list is being
3:38:39ingested parallelly and then
3:38:41transactions are being ingested
3:38:43parallelly. So now is a good time to
3:38:47join both the streams and find some more
3:38:51fraud detection scenario and then send
3:38:53email. So if we revisit our architecture
3:38:56so we can now say that two streams are
3:39:00now being ingested in bronze and silver
3:39:02and the next part we will work on this
3:39:04stream stream join. This is going to be
3:39:07interesting
3:39:09and before we actually do it there are
3:39:11some two three concepts that we need to
3:39:14understand first so that once we are
3:39:16clear with those concepts then when we
3:39:19do the implementation it will be clear.
3:39:21So let's understand those concepts
3:39:23first.
3:39:24So first of all let's understand the
3:39:26concept of stateless versus stateful
3:39:29stream processing. So so far what
3:39:32whatever we did with the transactions
3:39:34only data. So there we did a stateless
3:39:37stream processing. There what it means
3:39:40is that no memory of past events is
3:39:42required and each event is processed
3:39:44independently. So whenever we received
3:39:47the transactions event from Kafka so
3:39:50each and every event it was processed
3:39:52independently and there was no
3:39:55dependency on other events. So as and
3:39:58when we were getting each event we were
3:40:00processing them. So there was no
3:40:01interdependency among events. So that is
3:40:04why it is called as stateless stream
3:40:07processing.
3:40:08On the other hand we have second type
3:40:10which is stateful stream processing. So
3:40:13this it it need to maintain some state
3:40:17or memory of past events to process the
3:40:20current event.
3:40:22So how it works is new events keep
3:40:24coming. We do some processing and we
3:40:27keep some memory for some time. How how
3:40:30much time and all we can control it. We
3:40:32will see. But we have some memory. So
3:40:34new events they keep coming. We check
3:40:37something in the memory and then we
3:40:38process them and then we write some
3:40:41output. So we need some kind of state
3:40:43for this and that is why these are
3:40:46called as stateful stream processing.
3:40:49And if you take some example so
3:40:52stateless it can be simple filter event
3:40:55uh like we used wherever we we filtered
3:40:57out the transaction amount greater than
3:40:59some value. So if you have some fixed
3:41:02value that can be a simple filter event.
3:41:04If we have a simple map like one to one
3:41:08record transformation we want to do like
3:41:10casting conversion something like that
3:41:13or if you want to enrich it with some
3:41:14static data. So in our case we combined
3:41:18our transaction with the customer static
3:41:21data. So that kind of processing also
3:41:23comes into stateless or if you want to
3:41:26do some simple routing like based on
3:41:28some value if you want to route between
3:41:30different tables or various routes. So
3:41:33for that we don't want any memory. So
3:41:36that is why these examples come under
3:41:38stateless wherein stateful there are
3:41:41different type of example like if we
3:41:43want to do some rolling window
3:41:45aggregation. Let's say if you want to
3:41:47know every 5 minutes how many
3:41:49transactions are coming. So for that for
3:41:525 minutes we need to keep counting and
3:41:55keep things in memory. So that is why it
3:41:57is there. And if you want to do like per
3:42:00user session if you want to capture some
3:42:02detail for some kind of processing for
3:42:05running total counts like how many
3:42:08records or how many events are coming in
3:42:10total. So for that thing or for advanced
3:42:14fraud detection also that we will do now
3:42:16there also we will need a stateful kind
3:42:19of processing. So simply putting on
3:42:23stateless there is no state on stateful
3:42:25we maintain a state in stateless the
3:42:28latency it is less because as soon as we
3:42:30are getting the data we are processing
3:42:32each and every event in a stateful there
3:42:36can be some latency and that will depend
3:42:39on how much memory we are keeping. This
3:42:42one we will see next how do we control
3:42:44it. On stateless the resource usage is
3:42:47less because we don't need memory and
3:42:50single record processing can be there.
3:42:52Here resource usage is more because we
3:42:55need to process multiple set of records
3:42:57together. This is easier to scale and
3:43:00manage wherein stateful it is for
3:43:02powerful use cases. So it is slightly
3:43:06difficult to manage but it is worth it
3:43:08because that is the use case that we
3:43:10need. So this is the core definition of
3:43:13stateless and stateful and to maintain
3:43:16some of the things in stateful like how
3:43:19do we handle late arriving data or how
3:43:21much memory up to how much time we want
3:43:24to keep the memory all these things we
3:43:26need to control. So to control all these
3:43:29things we need something called as
3:43:31watermarking.
3:43:33So watermarking this is a kind of
3:43:36concept and a functionality in
3:43:38structured streaming. So this is what is
3:43:40a watermark. It is a threshold on the
3:43:43event time. So here what we do is we
3:43:46define a timing window like how much
3:43:50late data we can consider when it is
3:43:53arriving late. So we put some threshold.
3:43:56So whatever data is earlier than that
3:43:58threshold that will be considered final
3:44:01and whatever data that is that we are
3:44:04getting after after that threshold uh
3:44:07limit. So that data we will not use. So
3:44:11if we understand it from an example. So
3:44:14let's say we are getting some data and
3:44:16we have configured our watermark is 10
3:44:19minutes. So how it works is so let's say
3:44:23we are getting these different different
3:44:25events on a particular time and the
3:44:27latest event that we have that is from
3:44:2912:30.
3:44:31So if our watermark is 10 minute uh is
3:44:3310 minutes then the current watermark
3:44:36value or the latest allowed limit of the
3:44:39data that will be 10 minutes before the
3:44:42latest data point. So for this example
3:44:45it will be 1220.
3:44:47So whatever data is earlier than 1220
3:44:50that data will be ignored. It will not
3:44:52be considered and if the data is within
3:44:56these 10 minutes that will be considered
3:44:58for processing. So that is how this
3:45:00watermarking it works. And how do we
3:45:04define it? So we define it using this
3:45:07with watermark. We give a column name
3:45:09the the column that we are going to use
3:45:12for watermarking. Then the value whether
3:45:15it is minutes, seconds or hours. So most
3:45:18cases it is a few minutes but it it
3:45:20depends on the use case. But this is how
3:45:23we define it. And then there is one
3:45:26important aspect of it that is sometimes
3:45:28misunderstood that whenever we have
3:45:30defined this 10 minutes. So this 10
3:45:33minute it is considered based on the
3:45:36latest value received of this column. So
3:45:39it is not like 10 minutes before the
3:45:42current time. It is before the latest
3:45:45value of this column. That is how the
3:45:47watermarking works.
3:45:49And these these are some key points that
3:45:52this is used to handle late arriving
3:45:54data. So here we define a threshold like
3:45:57up to 10 minutes of late data we will
3:46:00allow earlier than that like or older
3:46:02than that we will not allow and with
3:46:06this we get a kind of bounded state. So
3:46:09we get a fixed window that we will
3:46:11process and that this much of data we
3:46:14will keep in memory and with this way we
3:46:17get some accurate results because
3:46:19otherwise there is no limit in the
3:46:20streaming data how much data we can
3:46:23process. So now in our use case we are
3:46:27going to join to streaming data. So to
3:46:30join to streaming data defining the
3:46:32watermarks is important. So that
3:46:34whatever data is falling within that
3:46:37watermark values or that window of that
3:46:41watermark that data will be considered
3:46:43for joining. So continuously the data
3:46:46will be coming whatever data is falling
3:46:48within that window that will be joining
3:46:52and then we will be getting the results.
3:46:54And if some data arrives later than that
3:46:56window or some like the value is late
3:47:00value is like less than the watermark
3:47:02and it is received after the processing
3:47:05time then it will be ignored. That is
3:47:07how the join will actually work. So now
3:47:10let's go ahead and implement this
3:47:12scenario.
3:47:14So now to implement that scenario we are
3:47:17back in our pipeline and to join both
3:47:20the streams we will build another goal
3:47:23table which will be doing like join of
3:47:26the streams and then preparing another
3:47:28kind of alert that alert we will name it
3:47:31name it as fraud card alerts. So what
3:47:35how what we are assuming is that
3:47:37whatever data we are getting through
3:47:39this watch list. So all those cards has
3:47:42been marked for marked as inactive or
3:47:46for related to fraud. So whatever
3:47:48transactions we will be getting for
3:47:50those cards. So those we will be
3:47:53considering as fraud and we will be
3:47:55sending an alert. So let's go ahead and
3:47:58join both the streams.
3:48:00And to do that we will uh copy this
3:48:02ready to use code here the one we
3:48:05prepared earlier.
3:48:07So let me copy this much of the code
3:48:11and then we will prepare another file
3:48:13inside this gold one
3:48:18and this one we will name it as
3:48:24road
3:48:26card alert.
3:48:30So let me paste the code here. So this
3:48:33will be same. This one we need to rename
3:48:36it.
3:48:41And this function name also we can name
3:48:44the same.
3:48:46And here as you can see one stream we
3:48:49are already creating for transactions
3:48:51and customers. So this also we will be
3:48:54using it later on so that we can bring
3:48:57in the email id to send the email. Uh
3:49:00but this one we will be using it later
3:49:02on. First we will join the second stream
3:49:04here. So let's bring in the fraud watch
3:49:07list.
3:49:12So it will be similar like spark dot
3:49:16readstream dot table
3:49:23and it will be fine guard
3:49:27dot silver
3:49:30dot fraud watch list.
3:49:34So now both the streams are ready. Next
3:49:37what we are going to do is we are going
3:49:40to add the watermarks to these so that
3:49:43we can perform the join and to add the
3:49:46watermark we will name those as like
3:49:49transactions with watermark
3:49:53and then we will say oh it is providing
3:49:55some automatic suggestion so I will
3:49:58press tab here so for customers we don't
3:50:01need a watermark
3:50:03and this is the syntax of watermark like
3:50:06I was talking earlier here. So this is
3:50:08the transaction column that it will be
3:50:10using and currently it is saying 1 hours
3:50:13but we don't need that much of a big
3:50:16window. We will simply keep it 5 minutes
3:50:21and then for fraud also like the column
3:50:24that is there. So we will be using the
3:50:26effective
3:50:28pro column and then the watermark also
3:50:32we will keep it 5 minutes. In practical
3:50:35scenarios, it can be like multiple hours
3:50:38also because the transaction that will
3:50:40be happening and when it will be marked
3:50:42for fraud. So we want more overlapping
3:50:45for that or in real scenarios it can be
3:50:48even in days. Uh that is also practical.
3:50:51But in our case we will keep it 5
3:50:53minutes so that within 5 minutes
3:50:56whatever fraud card we are getting and
3:50:58whatever transaction is happening for
3:51:00that card that will be considered as
3:51:03fraud.
3:51:05So now we we have like provided a
3:51:08boundary to the streaming data that uh
3:51:12this watermark 5 minutes. So within 5
3:51:15minutes it will keep doing joining and
3:51:17whatever data is matching that it will
3:51:20uh keep providing in our output table
3:51:22and the data that will arrive later than
3:51:255 minutes that will be ignored
3:51:26automatically.
3:51:28So now let's go ahead and detect this as
3:51:31a fraud. So we will say that uh fraud
3:51:36detected equals to we will be doing
3:51:39multiple things here. So first of all
3:51:42obviously it will be a join. So it is
3:51:44providing this as a suggestion.
3:51:48So we will take our transactions with
3:51:51watermark dot join
3:51:57and in this join uh first we will pass
3:51:59the fraud watch list with watermark
3:52:04and then we need to pass the joining
3:52:06column.
3:52:08So we will say that from this
3:52:11transaction part we will be taking the
3:52:14card number
3:52:21and it shall be equal to uh in the watch
3:52:24list it the column name it is entity ID.
3:52:29So when both are matching that means it
3:52:32is a fruit and the join type here it
3:52:35this one we have to take it as inner
3:52:37join. Why? Because once the entry is
3:52:41available on both sides then only it is
3:52:43a fraud otherwise it is not a fraud. And
3:52:47this inner it will work because of our
3:52:49watermark so that it will detect
3:52:51everything within the 5 minutes.
3:52:54So now we have detected our fraud. The
3:52:57next thing we need is a email id so that
3:53:00we can send the email. So we will now
3:53:03join our customer table. And this one as
3:53:06we know so it is not a streaming table.
3:53:08It is kind of a lookup table. So this
3:53:11one we will join customers. And this
3:53:14joining condition we already know it is
3:53:16like customer ID on both sides. So we
3:53:20will take this one dot
3:53:22customer id equals
3:53:27uh then we will say customers dot
3:53:29customer id
3:53:34and then the join type.
3:53:38So join type here uh this one it will be
3:53:41left join because stream directly like
3:53:45uh we shall not do inner join here
3:53:49because like it is a lookup table and
3:53:51when whenever email id is available that
3:53:54we shall join and now our tables are
3:53:58ready tables or the data frame. Next we
3:54:01need to select the columns or the
3:54:03details whatever we need to do here. So
3:54:07this time let me pause it and type
3:54:09everything first. So here I have typed
3:54:12it. So similar similar to our
3:54:15transaction with higher value we are
3:54:17generating a alert ID. This time we are
3:54:19saying it is a like fraud transaction
3:54:22ID- watch list ID and then
3:54:27alert type we are saying it is like
3:54:29fraud watch list match and alert
3:54:32timestamp we are taking as the current
3:54:33time stamp. Then we are taking the
3:54:36columns from like transaction ID,
3:54:38customer ID and customer email also we
3:54:41are taking that we will use to send the
3:54:44email. Uh customer first name and last
3:54:46name we are preparing and then multiple
3:54:49columns we are taking from the uh
3:54:52transactions. So all these columns are
3:54:54direct selection city and all these we
3:54:57are putting an alias to understand these
3:54:59better. And then
3:55:02uh from the watch list side also we are
3:55:04selecting watch list ID, type all these
3:55:07columns and effective from we are
3:55:09renaming this column like watch list is
3:55:11effective from and then these are as it
3:55:14is and then uh city and country also we
3:55:18are marking that this is watch list city
3:55:20and country so that it is not like uh
3:55:24misunderstood with the customer city
3:55:26country etc.
3:55:28So with this all the columns that we
3:55:30want we have selected that and now our
3:55:33table will be created automatically.
3:55:37So now this table it will contain the
3:55:40alert data like we had for transaction
3:55:43alert and the next thing we need is the
3:55:46actual alert. So for that we will need
3:55:49to create this kind of sync and this
3:55:51time we will not type everything. We
3:55:53will use Genie code because it will be a
3:55:55similar activity.
3:55:57So let me open Genie code here and let's
3:56:01start a new thread
3:56:05and the prompt that we need to provide
3:56:07here. First let's prepare that prompt
3:56:09separately so that we can understand and
3:56:12then we will paste it here. So how we
3:56:14will provide it? We will say that we
3:56:17have prepared this file which reads this
3:56:19table and send an email notification.
3:56:22Similarly prepare for this one.
3:56:27So we will copy this name.
3:56:33We will say that
3:56:36there are high value
3:56:42transaction alerts
3:56:47logged in the table.
3:56:52which are fraud
3:56:55and an email is
3:56:59sent using the file.
3:57:03Let me copy the other file name
3:57:14this file name. So we will say this is
3:57:18working fine.
3:57:21I want
3:57:24you to implement or we can say like
3:57:28create a new alert file
3:57:33that read the road data from. Then we
3:57:38will give our table name. So the table
3:57:41name will be this one.
3:57:49this table and send email
3:57:54notification
3:57:55and this table it is not yet created. So
3:57:58let's give it the file name also so that
3:58:01it can understand what all columns are
3:58:03there. So we will say this table
3:58:07prepared by this file.
3:58:10So this is the prompt that we are going
3:58:12to provide. Let's see how much it is
3:58:15able to write for us. So I have typed
3:58:18it. I will press enter. So it is
3:58:21currently understanding all the details.
3:58:23Now it will create a new file.
3:58:26So looks like it has created this file.
3:58:31So currently it seemed to be blank. It
3:58:34is updating the file. And now it has
3:58:37created the file but it is not able to
3:58:40write the data. It is saying that
3:58:42similar safety alert because we are
3:58:45using the SMTP API call. So it has
3:58:48provided the code. So I will need to
3:58:50copy the code from here and then paste
3:58:52it. So now I have pasted the code and
3:58:57let me close this right side. Let's have
3:58:59a look at the code. So the code is
3:59:01similar only. This is our send email
3:59:03function. This is the function that is
3:59:06creating the body of the email with so
3:59:08many details. And then there are these
3:59:11other functions. So this one it is uh
3:59:14reading our fraud alert card sorry fraud
3:59:17card alert table as a streaming table
3:59:20and it will keep appending the data
3:59:22using append flow to this target sync
3:59:26and what is this sync it is this one. So
3:59:28here it will send that data as a data
3:59:31frame and batch ID. It will it is
3:59:34reading all the API key and preparing
3:59:36the app password. Then the data that has
3:59:39been read as a data frame. So it is
3:59:41collecting all the rows. Then it is
3:59:44going row by row here and after that it
3:59:47is preparing the body of the email using
3:59:50the above function and then it is
3:59:52sending the email using above function.
3:59:54So structure wise it is exactly same to
3:59:56the other file and it will send the
3:59:59alert notification.
4:00:01So now this piece has been set up end to
4:00:03end. So now let's see it in action. So
4:00:07what I will do is I will change it to
4:00:10continuous mode so that we can start it
4:00:13and then send our data and see it in
4:00:15action. So I have changed it
4:00:18and now it is starting our pipeline
4:00:22and this pipeline it seem to be having
4:00:24an issue that comes up for a second and
4:00:26then goes away. This is happening due to
4:00:29continuous mode. So for debugging for
4:00:32now let me change it to triggered
4:00:35and let me stop it manually and then
4:00:38let's do our first run as a batch kind
4:00:42of run or one time run.
4:00:45So now we are getting the actual error.
4:00:47It is saying that transaction timestamp
4:00:50it has the invalid type string but it is
4:00:53expecting a time stamp. So probably we
4:00:57have created our transaction timestamp
4:00:59as string type we can see here. So this
4:01:02one it needs to be of timestamp type. So
4:01:06what we will do do is here we will
4:01:08change it to timestamp type.
4:01:13So it has been changed and now we will
4:01:16run again but this time we will not do a
4:01:18run pipeline instead we will do a full
4:01:21refresh. So whenever we are making some
4:01:24changes that can affect our historic
4:01:26data so while development so for that we
4:01:29need to do a full load so that it will
4:01:32recreate the tables. So this time we
4:01:34will do a full table refresh.
4:01:40So this time the error is that we are
4:01:42not returning anything probably it is
4:01:44saying it is returning none type. So
4:01:47let's jump to the code
4:01:50and here we can see that this fraud that
4:01:52we had detected. So we did not put a
4:01:55return statement to this one. So let's
4:01:58put it
4:02:06now. Let's run it again.
4:02:12And this time code wise things looks
4:02:14good. So it has started the pipeline. So
4:02:17first it is running the bronze layer
4:02:19tables. So for fraud watch list in total
4:02:22we have four records only. It has
4:02:23written those and same it has written to
4:02:27silver fraud list as well. And for
4:02:30transactions also it has read all the
4:02:32records again and then it has written to
4:02:35silver layer as well. And then high
4:02:38value transaction alert it has generated
4:02:40one one record and for that it will be
4:02:44sending the email as well. And for fraud
4:02:47watch list it has detected two alerts.
4:02:51So we have two such records based on
4:02:53these four cards that were there. Two
4:02:56transactions has been marked and it has
4:02:58sent the email as well. And if you see
4:03:01my email id so these are the three
4:03:04emails that we have received. This is
4:03:06the similar high value transaction
4:03:08alert. The same email that we had got
4:03:10previously. And this is the new email
4:03:12for that uh fraud card related alert.
4:03:16And depending on the risk, it is saying
4:03:18whether it is high or low risk. And then
4:03:20it is providing all these very detailed
4:03:24uh all the details here. And we can see
4:03:26the formatting also it looks really nice
4:03:29here. This is because of the detailed
4:03:31HTML code that Genie has generated for
4:03:34us. So if you are fine with this then we
4:03:36are good. Otherwise we can ask Genie to
4:03:39customize it as well however we want to.
4:03:43So this is how our stream and stream to
4:03:46stream join is also working fine. And
4:03:49now if we go and check our pipeline
4:03:51graph. So this time it looks like more
4:03:53interesting if I make it like uh bigger.
4:03:57So this is how it will look like. So
4:03:59both the streams are coming until silver
4:04:02layer. For high transaction we are using
4:04:04only one and sending the email. For
4:04:07fraud card card alert we are joining the
4:04:10both and then sending the email alert.
4:04:12That is how it is working. And each and
4:04:15at each and every step it is showing the
4:04:18number of records as well. How many
4:04:20records has been uh generated at each
4:04:23and every stage so that we get proper
4:04:26visibility.
4:04:28So now as we have understood stream to
4:04:30stream join as well. The next thing we
4:04:33are going to understand is how can we do
4:04:36aggregations on streaming data and
4:04:39aggregations again it is it will be a
4:04:41challenging piece because let's say if
4:04:44we want to get number of transactions.
4:04:47So if we uh if we go ahead and do number
4:04:49of transactions like count of
4:04:51transactions. So by default on streaming
4:04:54data it will keep getting keep getting
4:04:57keep adding more transaction and uh it
4:05:00is kind of continuous right so there is
4:05:02no limit when it will stop and give us
4:05:04the count that this is the count. So
4:05:07there also we need to provide some
4:05:09boundary how much data we want or how
4:05:13much after how much time we want to do
4:05:15the aggregation and to to provide that
4:05:19uh time limit we have some
4:05:22functionalities specific to streaming
4:05:24data that is called a window and we have
4:05:27mainly two types of window that we can
4:05:29use. So let's go ahead and understand
4:05:31those.
4:05:33So the two types that we have those are
4:05:35tumbling window and sliding window and
4:05:38this this one we call it as bounding the
4:05:41unbounded stream because using these we
4:05:43provide some boundary. So first let's
4:05:46understand the tumbling window. So this
4:05:50one it is like a nonover overlapping
4:05:52fixed size window and how what what it
4:05:56means by that is let's say here we are
4:05:58getting some events continuously and we
4:06:01define a window of let's say 10 minutes.
4:06:04So 0 to 10 minute it is one window 10 to
4:06:0720 it is window number two 20 to 30 it
4:06:10is third window. So these three are
4:06:13fixed size window which do not overlap
4:06:16with each other and in this case each
4:06:18event belongs to one window. So wherever
4:06:21the event is it will belong to only one
4:06:24window and results are produced when the
4:06:27window closes. So it will keep getting
4:06:29the data keep getting the data whenever
4:06:31it will reach that 10 second then it
4:06:33will provide the aggregation value
4:06:35whatever is required. And the second
4:06:38type of window that we have is sliding
4:06:40window. So sliding window as the name
4:06:43suggest it keeps sliding. So in the
4:06:46similar case if we have this data from 0
4:06:49to 30 minutes. So here also window size
4:06:52is 10 minutes only but we keep sliding
4:06:56this. So for example 0 to 10 minute it
4:07:00is one window. Then 5 to 15 minutes.
4:07:03This is our second window. So this
4:07:05window it seem to be slided by 5
4:07:08minutes. Then uh 10 to 20 is another
4:07:12window. 15 to 25 is another window. Then
4:07:1520 to 30 window comes. So within the
4:07:18same time of 30 minutes here we had
4:07:21these three windows. Here we have five
4:07:23windows because the same window it keeps
4:07:26sliding. So here we can say that window
4:07:28have a has a size and a slide interval.
4:07:32So what it means is that we will need to
4:07:34provide two parameter whenever we
4:07:36configure it. One will be the size, one
4:07:38will be the interval. In case of
4:07:41tumbling, we will only provide the size.
4:07:44And here windows can overlap. And
4:07:46because these can overlap, so whatever
4:07:48event is there that may belong to one
4:07:51window or two windows as well.
4:07:54So in a nutshell, we use tumbling window
4:07:58for simple non-over overlapping
4:08:00aggregation. Let's say if you want count
4:08:03of transactions every 10 minute then we
4:08:06can use this kind of window and let's
4:08:08say if you want more frequent up toate
4:08:10insight let's say we want every 5 minute
4:08:14we want the number of transaction of
4:08:16last 10 minutes then we use this kind of
4:08:19window that so that here it will give
4:08:21these 10 minutes here it will give these
4:08:2410 minutes here it will give these 10
4:08:26minutes so these are more up to date
4:08:28because every 5 minute we are getting
4:08:31the data of last 10 minutes. We can do
4:08:33it like every 1 minute also and window
4:08:36size of five or 10 minutes. So that is
4:08:39how we can customize it. So these are
4:08:41the two types of windows that we can use
4:08:44and do some aggregation. So let's now
4:08:46understand how practically we can do it.
4:08:50So to do that piece we will again create
4:08:52another gold table. So for that let me
4:08:56copy this piece of code from here and
4:08:58then we will create a new file here
4:09:02and this one uh we will name it as
4:09:08transaction count. We will count the
4:09:10number of transaction and then we will
4:09:13say by
4:09:15minute. We will try to do it every
4:09:17minute.
4:09:20So here I think I forgot to put the dot.
4:09:24So it gave a warning. Let me type it
4:09:26again.
4:09:33And then here let me paste the code that
4:09:36I had copied. And then let me copy this
4:09:39name.
4:09:47And then we will create a table with
4:09:50this name. And this function also let's
4:09:52give it the name. And here because we
4:09:55are counting the number of transactions.
4:09:57So this stream only we will be using.
4:10:01So now before we do the aggregation
4:10:03using window here also we will need to
4:10:06specify a watermark so that we can we
4:10:10can tell it how much late data we are
4:10:12allowing it.
4:10:15So that like uh it it will wait for that
4:10:17much of time and then it will do the
4:10:19aggregation.
4:10:21So if we go back there so this same
4:10:24watermark we will add here.
4:10:27So here also we are saying that we will
4:10:30allow the late data by 5 minutes.
4:10:34And now we have everything. So now we
4:10:36will do the actual aggregation. So what
4:10:39we will do is we will say transaction
4:10:46count equals to we will take this
4:10:50watermark value
4:10:52and then do a group by first.
4:10:58So it is providing a suggestion here and
4:11:01in the group by this is where we need to
4:11:04specify our window. So the window that
4:11:07we are going to use is it will be f dot
4:11:10window
4:11:13and then we will mention the timestamp
4:11:16column that we will use for preparing
4:11:19the window and then similar way we need
4:11:22to mention the size of the window.
4:11:26So this is how we will specify and
4:11:30window size this time we will keep it 1
4:11:32minute only. So this will be our
4:11:34tumbling window. So every 1 minute it
4:11:38will do the count of transactions.
4:11:41So so far this is just the group by
4:11:43part. Now let's mention it what we want
4:11:46after doing the group by. So simply we
4:11:49want like f dot countstar
4:11:55and this count star that it will do we
4:11:58will name it as
4:12:02transaction count
4:12:10and after doing the count we want to
4:12:13select what data we want in the output
4:12:16table. So in the select also there is
4:12:19something specific to the window.
4:12:23So when we say f dot column so the
4:12:26window that we are using so we have
4:12:28window dot start and window dot end
4:12:31timestamps that we can take. So I will
4:12:34press tab here and this window dot start
4:12:38this one I will name it as window
4:12:41start as a column
4:12:45and then window dot end that we will
4:12:48name it as window end and after that we
4:12:51will take our transaction count and
4:12:53that's all. So in our target table these
4:12:56will be the three column we will be
4:12:58creating. So we will see a window start
4:13:00time, end time and the count of
4:13:03transactions within that window. And now
4:13:06this uh this is ready. So this name may
4:13:10be confusing for this. So I will name
4:13:13this one as DF because it is a data
4:13:15frame. And this is what we will return
4:13:20so that it doesn't fail this time. So
4:13:22this is what it will create this as a
4:13:24data frame and then this data frame it
4:13:28will be written into this particular
4:13:30table here. So now let's go ahead and
4:13:33give it a try. So I'm going to run the
4:13:35pipeline.
4:13:38And this time also it failed with some
4:13:40error saying like column object is not
4:13:43callable. So if we jump to the code, it
4:13:45is pointing something here. So let's
4:13:48diagnose the error and let's see uh what
4:13:51is the fix for this one. So now it has
4:13:54made the changes. So looks like it made
4:13:57it okay three places. So here we can see
4:14:00it was a minor issue. So just this
4:14:03bracket it has to be right side the
4:14:07group Y. So let me accept this. And now
4:14:10let's see what other change it has made.
4:14:15So it has also made some change in this
4:14:17alert not uh email notifier
4:14:26and here what it is trying to do is
4:14:29earlier this app password and this
4:14:31secrets get it was inside this function
4:14:34so for every batch it was calling it
4:14:37again and again now it is putting it
4:14:39outside so it will be optimized and
4:14:42probably these warnings things like not
4:14:44serializable these also it will be gone
4:14:47because I think it noticed that and then
4:14:49fixed it so that is also good for us so
4:14:52I will simply accept all the changes now
4:14:55and then let's go ahead and run it
4:14:59so this time the pipeline has been
4:15:01started
4:15:06and now it is successfully complete and
4:15:09this time this was the only new table so
4:15:12it has written the five records for that
4:15:15that were there in the last last 1
4:15:18minute and delayed by 5 minutes. That
4:15:20way it has prepared some data for 5
4:15:23minutes and rest of the thing because we
4:15:26did not have any new data. So there is
4:15:28no new data coming in. So let's go ahead
4:15:31and check out this table. So this time
4:15:33we will copy it in a notebook directly.
4:15:38So we will do a select star from
4:15:41pinegard dot gold. Then uh this is the
4:15:46table we have prepared.
4:15:49So here we can see the data. So if I
4:15:52sort it by transaction star time. So
4:15:55here we can say see that uh 654 to 655.
4:16:00This is one window where it got 61
4:16:03transactions. Then 9:44 to 9:45 it got
4:16:0718 transactions. And why we see this gap
4:16:11is because this video I have been
4:16:13recording this over different hours. I'm
4:16:15taking breaks and all. So whenever we
4:16:18send the data within 1 minute that is
4:16:20what is what it is capturing. And here
4:16:22we can see a good window that 116 to 17
4:16:26was one window. Then 17 to 18 was the
4:16:29next window. So this is how for every
4:16:32minute we will keep getting data here.
4:16:35So this is our tumbling window and this
4:16:37is the aggregated value for each of the
4:16:39tumbling window. So now as we understand
4:16:42the tumbling window aggregation
4:16:44completely. So next we will give it a
4:16:47try for the sliding window and that one
4:16:50it is going to be very similar to this
4:16:52and very easy. So simply let's copy it.
4:16:56So we will clone this file and make that
4:16:58minor change.
4:17:01So this one we will instead of simply
4:17:03saying count by minute we will say count
4:17:06by minute and then we will mention
4:17:08sliding window
4:17:14and in here I put a dot. So let me
4:17:17remove this.
4:17:24So now this has been cloned
4:17:29and here also let's create another table
4:17:32with the name sliding
4:17:37window. So this will be our target
4:17:40table. So here also we will be doing the
4:17:43exact same thing but this time our
4:17:45window size we will change it to 5
4:17:48minutes.
4:17:49So let me change it to 5 minutes. Why
4:17:52I'm changing? Because we need a sliding
4:17:55also. So sliding we will keep it for 1
4:17:58minute. So what it will do is every 1
4:18:01minute it will do the aggregation of
4:18:04last 5 minutes.
4:18:07So that is how it will work. For example
4:18:09at like 4 0 it will get the data of last
4:18:145 minutes. At 401
4:18:17it will get the data again for the last
4:18:195 minutes. So 4 minutes those will keep
4:18:22overlapping again and again and by 1
4:18:25minute it will keep sliding. So if you
4:18:28want to understand from this perspective
4:18:30so this window it we have kept it as 5
4:18:33minutes and this sliding time like how
4:18:35much it is going to jump to the next
4:18:37one. That one we have kept it as 1
4:18:40minute and that's all that's all the
4:18:43change we want. So this is the only
4:18:46difference only one additional parameter
4:18:48and it becomes a sliding window. So now
4:18:51let me go ahead and run this pipeline
4:18:53again
4:18:55and the execution has been started now.
4:18:58So while it is running let's have a look
4:19:01at our pipeline graph now. So it looks
4:19:04more expanded now and as we can see uh
4:19:08for for transactions now we are doing
4:19:10multiple thing. So the fraud card alert,
4:19:13the high value transaction alert and
4:19:16also we are preparing two more streaming
4:19:18tables now with the two different type
4:19:21of window-based aggregations.
4:19:24And here we can see that it has already
4:19:26written 21 records now and then it has
4:19:29stopped.
4:19:31So now let's have a look at this newly
4:19:34table that has been created. So let me
4:19:37query this here itself.
4:19:45So this is the sliding window table
4:19:49and here we have like lot more data for
4:19:52the same window that that was for
4:19:54tumbling. So now we have more data and
4:19:57there is a reason because it is a
4:19:58sliding window. So this is what we were
4:20:02talking about. So 50 to 55 650 to 655 we
4:20:06have 61 records. Then 51 to 56 we have
4:20:11again this many records. 52 to 57 again
4:20:14we have this many record. So as we can
4:20:17see it is sliding by 1 minute and the
4:20:19window size it is again 5 minutes every
4:20:22time. So that is how this sliding window
4:20:25works.
4:20:27So these are all the tables we wanted to
4:20:29create using the streaming pipeline. So
4:20:32now what we will do is now let's have a
4:20:35look at at the streaming data flow end
4:20:38to end in real time in all the layers.
4:20:41So for that what we will do is we will
4:20:43start our pipeline in continuous mode
4:20:46and then we will start our streams as
4:20:48well.
4:20:50So it has been started here.
4:20:54So now this pipeline has been started
4:20:57and it is waiting for the data and as it
4:21:00is not receiving any data it will go to
4:21:03the idle mode. So now let's give it some
4:21:06data and see the data flowing end to
4:21:08end. So what I will do is I will go to
4:21:11our notebook section and then we will
4:21:14start pushing the uh fraud watch list
4:21:18data continuously. So let me start this
4:21:21uh run this cell so that it start
4:21:23sending the fraud watch list data.
4:21:27So now it has started sending the fraud
4:21:29watch list data and also parallelly
4:21:32let's start the other stream as well
4:21:35where we were sending the uh transaction
4:21:38related data. So uh this is the fraud
4:21:41transaction. Let's send the uh normal
4:21:44transaction. So that that will be having
4:21:46like load lot uh lot much more data and
4:21:51that is with the name normal.
4:21:55So let me run this
4:22:00and it has also started sending the data
4:22:03in real time and parallelly. Let me go
4:22:07to another tab another terminal here.
4:22:12So here we will also call the fraud
4:22:15transaction. Let's send one fraud
4:22:17transaction as well. So it has sent this
4:22:20transaction and it has exited and our
4:22:23normal transaction it is going
4:22:24continuously and also our fraud watch
4:22:28list data that is also going
4:22:30continuously. Now it is running. So now
4:22:33let's go ahead and look at our pipeline.
4:22:36So here we can now see some real data
4:22:39coming in. So transactions also data is
4:22:42coming. Fraud watch list also data is
4:22:45coming properly.
4:22:47And now the high value transaction
4:22:49alert. So one transaction we has sent as
4:22:52a fraud transaction with high value that
4:22:55has been uh also detected here and an
4:22:59email has been sent for the high value
4:23:02transaction.
4:23:04And similarly for fraud card alert also
4:23:07it has detected like one fraudulent
4:23:09transaction
4:23:10and for that also it has sent the email
4:23:14and for for these window functions
4:23:16currently it is this is in idle state. I
4:23:19think it is currently waiting for that 5
4:23:22minute window then it will start doing
4:23:24the aggregation. So it we will let it
4:23:27run for some time parallelly. Let's go
4:23:29ahead and check our email.
4:23:32So here we can see now multiple emails
4:23:35coming one by one of both the types. So
4:23:38this is how it is running fine end to
4:23:40end now
4:23:42and this is how like end to end flow of
4:23:46streaming data looks like. So now both
4:23:49the streams the data is coming
4:23:51continuously. Even if we go to our
4:23:54pipeline graph, it will also show the
4:23:56same that this is also running this is
4:23:59getting the data this is from bronze to
4:24:02silver we are getting then all these
4:24:04aggregations are happening these frauds
4:24:07are being detected and emails are being
4:24:09sent in real time and this watch list
4:24:12also we are getting continuous data.
4:24:15This join also it is happening like
4:24:17stream to stream join and then sending
4:24:19the email and here the stream to static
4:24:23join is happening and then the uh fraud
4:24:26email are being sent. So this is how an
4:24:29end toend streaming pipeline looks like.
4:24:33So I will let it run for this amazing
4:24:36pipeline.
4:24:38And now let's understand what is left
4:24:40for us to do now. So now revisiting our
4:24:45uh architecture again. So these pieces
4:24:48we have completed all the data injection
4:24:50all the data processing is done and then
4:24:53consumption side email alert part we are
4:24:56done. So the pieces pending are the
4:24:58dashboard and the orchestration. So
4:25:01first we will prepare the dashboard and
4:25:03then we will prepare an end toend
4:25:05orchestration.
4:25:07So now let's go ahead and create our
4:25:09dashboard. So for creating the dashboard
4:25:12we are not going to create it manually.
4:25:15I find it boring and it the genie code
4:25:18it can do lot of things automatically
4:25:20when it comes to the dashboard part. So
4:25:23what we will do is first let's go out of
4:25:26this streaming pipeline. So let's go to
4:25:28the workspace mode
4:25:32and now we will ask the genie code to
4:25:34create a dashboard for us.
4:25:38And this time also instead of writing
4:25:40the prompt here itself we will first
4:25:42type it here. So we will say that create
4:25:46a new dashboard
4:25:51and the dashboard we wanted to create
4:25:53from both our silver and gold tables. So
4:25:57here I want to give it some minimum
4:25:59instruction and then uh come up with a
4:26:01dashboard that we can customize later
4:26:04on.
4:26:05So I will say that step one itself
4:26:09understand all the tables
4:26:12structure from
4:26:16uh fine guard dot silver schema.
4:26:21It shall do this
4:26:25and then what it shall do is it shall
4:26:27also do the same for gold layer.
4:26:33So it shall also understand the tables
4:26:35in the gold schema and then it shall
4:26:39like uh
4:26:42prepare some or we will say come up with
4:26:46some come up with some good APIs
4:26:51and prepare a nice dashboard for me. So
4:26:56this is the prompt we are going to give
4:26:58it so that it can understand all those
4:27:01tables automatically and then create a
4:27:04dashboard for us. So let's see what kind
4:27:08of dashboard does it come up with.
4:27:11So first it is doing a show tables in
4:27:13silver layer so that it can see the list
4:27:16of tables that are available.
4:27:18So looks like it threw an exception that
4:27:21the catalog is not find found. So like
4:27:24always I may have made some typo.
4:27:28Yes. So let me correct that. It is Z U A
4:27:31R D.
4:27:41So now let me create a new thread and
4:27:44give this prompt again.
4:27:48This time it we can see that it has
4:27:50found the tables and now it is trying to
4:27:54understand the structure of each and
4:27:56every table.
4:27:58Now we can see it has created a
4:28:00dashboard a blank one and it will open
4:28:03that and then create many things into
4:28:05the dashboard.
4:28:07So this is our blank dashboard that it
4:28:10has opened automatically and now it is
4:28:13thinking what to do. Now it has come up
4:28:16with some KPIs that it is going to show
4:28:18us
4:28:20and now it is setting up some data sets.
4:28:23So to create any dashboard now it is
4:28:26starting a warehouse. Let me close this
4:28:28pop up here. It will create some data
4:28:31set. So if we go to the data tab we will
4:28:33be able to see whatever it is currently
4:28:36creating.
4:28:38So this one it took quite some time. So
4:28:41I took a break and then I came back. So
4:28:43now I can see that it has completed the
4:28:46things. It has created so many data sets
4:28:48for us. So how it works is this is a
4:28:51data set. First maybe let me close the
4:28:53right side. So here for example from the
4:28:57customers table it is applying a filter
4:29:00wherever risk score is greater than 70
4:29:03and that it is counting as a high risk
4:29:06count customers. Similarly, this is
4:29:08another data set where it is doing a
4:29:11transaction count of today's data only.
4:29:14It is today's data filter from the
4:29:16transactions table. Likewise, it has
4:29:19prepared different different SQL
4:29:21queries. And then it has also prepared
4:29:24this nice page with the visualizations.
4:29:27So from here directly we can understand
4:29:29what it has prepared. So these are the
4:29:31total alerts that has been generated
4:29:34today. So there are so many alerts which
4:29:36are running because the pipeline it was
4:29:38running for very long. After that I
4:29:41stopped it manually and these are the
4:29:44number of transactions that are that has
4:29:46happened today. The data set that we
4:29:48just now saw and this is the average
4:29:52transaction amount. So it is doing
4:29:54simply an average. These are high-risk
4:29:56customer. The one that we saw this this
4:30:00is where it is showing alerts over time.
4:30:03So here if you want to understand this
4:30:05we can simply click it and we can see
4:30:08here which data set is being used here.
4:30:11So this one it is alert uh alert trend
4:30:14over time. So if we go there and if you
4:30:18find the alert trend over time. So this
4:30:20is what it is doing here. So it is
4:30:24taking the alerts from both our tables
4:30:27of the last 24 hours. That is what it is
4:30:30doing. And it is doing a group by of the
4:30:33date and it is taking hour part of it.
4:30:36So for every hour it is checking like
4:30:39how many alerts we are getting. So for
4:30:43today like it only ran I ran it only
4:30:47twice today. So that is why it is
4:30:49showing only two entries. Here it is
4:30:52showing a breakdown like for high value
4:30:55transaction how many alerts are there.
4:30:57Uh so this is the count we have only two
4:31:00because this data we were sending
4:31:02manually for fraud watch list we have so
4:31:05many alerts that were generated
4:31:07automatically
4:31:08here it is showing top 10 customers by
4:31:11the alert count. So this is it is this
4:31:14one it is preparing from the uh by
4:31:17joining the different table the customer
4:31:20table and the other table probably and
4:31:23this is our alert distribution sorry
4:31:25alert status distribution.
4:31:28So here it is taking taking some status
4:31:31column which is always approved this one
4:31:33doesn't seem to be that useful I would
4:31:35say this is our transaction volume by
4:31:38minute. So this is what it is preparing
4:31:41from our aggregated table. So minute by
4:31:44minute it is showing how many
4:31:46transactions has been performed and
4:31:49recently I have stopped it. So the
4:31:51number of transactions has dropped. This
4:31:53is our transaction amount by merchant
4:31:56category. Then this is saying whether
4:31:59the transactions has been domestic or
4:32:01international. These are the top 10
4:32:04merchants by the transaction count. uh
4:32:07this is the country-wise transactions.
4:32:10This is some uh risk level distribution
4:32:12where currently there is no data coming
4:32:14in. Maybe there is some null column or
4:32:16something. And this one it is the
4:32:20payment channel from which payment
4:32:22channel is being done for each and every
4:32:24transaction. So this is a nice dashboard
4:32:26that it has prepared and for any of this
4:32:29if you want to customize it, we can
4:32:31customize it. For example, this one
4:32:34doesn't seem to be useful. So we can
4:32:36simply delete it. We can adjust anything
4:32:39that we want and any of the chart we can
4:32:42customize it as well however we want to
4:32:45from this right side. For now we will
4:32:47not focus on much customizations because
4:32:50our intention was to show a nice
4:32:52dashboard. And this is how the dashboard
4:32:55looks like. This page also we can rename
4:32:58it to anything. We will rename it to
4:33:01fraud monitoring.
4:33:04And that's all. With this we can say
4:33:06that our dashboard is ready. And we will
4:33:09publish this dashboard so that other
4:33:11users they can use it.
4:33:16And now the dashboard has been
4:33:18published. So it is available. So if you
4:33:20click here it will show the published
4:33:23version of the same dashboard.
4:33:26So this is how the published version of
4:33:28dashboard will look like to the end
4:33:30users.
4:33:31So now as we are done with the
4:33:33dashboard, let's see what is pending
4:33:35next. So now we can see that all these
4:33:38pieces we have completed. Only the
4:33:41orchestration piece is pending now. So
4:33:43let's go ahead and do this one. So for
4:33:46orchestration, so there are three pieces
4:33:49to our architecture. So one is batch
4:33:53processing pipeline, one is streaming
4:33:55pipeline and one is this dashboard. So
4:33:58for streaming pipeline we do not need
4:34:00any separate orchestration because once
4:34:02we change the pipeline to continuous it
4:34:05is always running. So there is no
4:34:07separate orchestration required for
4:34:09that. And for dashboard piece we get a
4:34:12schedule option. So this one we get an
4:34:15option how frequently we want to refresh
4:34:17this dashboard. So if we are preparing
4:34:20it for near real time so that we can
4:34:22monitor the transactions at the minute
4:34:25level. So the minimum refresh time we
4:34:27get is every 1 minute. So if we do that
4:34:31and if we create it so every 1 minute it
4:34:34will start refreshing this dashboard. So
4:34:36that is how we can orchestrate this
4:34:38dashboard. And now the only piece that
4:34:42is pending this one I canled it because
4:34:44I don't want to actually do it now. And
4:34:47the only pending piece is the batch
4:34:49processing pipeline. So how do we
4:34:52orchestrate the injection and then the
4:34:55silver pipeline there. So for that we
4:34:58will create a uh orchestration job. So
4:35:01we will go to jobs and pipelines. We
4:35:04will click on this jobs part.
4:35:07So other than jobs the injection and ETL
4:35:10pipeline we already covered. This was
4:35:12the only pending piece. So here we will
4:35:15say customer
4:35:18injection
4:35:22uh source to silver
4:35:30and in this one there are going to be
4:35:32two step. So first we will run a
4:35:35injection pipeline that will inject the
4:35:37data from source to branch
4:35:42and type will be pipeline only
4:35:47and here by default it is showing ETL
4:35:49pipeline only but our pipeline it is
4:35:52injection pipeline.
4:35:54So here we will select our fine guard
4:35:57customers injection pipeline that will
4:36:00ingest the data and then we will save
4:36:03this task.
4:36:05After that we will add a new task. This
4:36:08new task also it will be a pipeline.
4:36:11This time it will be ETL pipeline. So
4:36:14this one it will be bronze to silver.
4:36:20And here we will select the pipeline.
4:36:23And here we will select the pipeline
4:36:25that we had created for customers table
4:36:28bronze to silver injection. So this is
4:36:30the pipeline and we will save it. So
4:36:34that's it. These are the only two task
4:36:36we need to do there
4:36:39and then we can run it manually from
4:36:41here so that it will first ingest the
4:36:43data from source to bronze and then to
4:36:45silver and after testing that we can add
4:36:48a trigger here as well. So trigger type
4:36:51it can be scheduled in this case because
4:36:54we want to run it every 1 day that is
4:36:57how we want to do it. So this is the
4:37:00default one. If you want to select a
4:37:02specific time when we want to run it. So
4:37:04every day it will run at this time. So
4:37:07time also we can select and then if we
4:37:10save so every day it will do a batch
4:37:12injection for this test for this table.
4:37:15So the streaming part it will keep
4:37:17running continuously and batch part
4:37:20whatever latest data is available in our
4:37:22silver customers table that will be used
4:37:24and every day it will it will be
4:37:26refreshing and new data will be used. So
4:37:29that is how the end toend orchestration
4:37:31of this project will look like. So
4:37:34basically this process it will run once
4:37:36a day. Then streaming injection it will
4:37:39run continuously and dashboard refresh
4:37:42that will happen every 1 minute. So then
4:37:45it will be running end to end
4:37:47continuously and uh and all the data
4:37:51that that processing it will happen in
4:37:54real time. So that covers the
4:37:56orchestration piece of our architecture.
4:38:00So now to summarize we have completed
4:38:03each and every functionality in this
4:38:05architecture we have done hands-on. We
4:38:07started by streaming some data from
4:38:09Kafka loading it through bronze and
4:38:12silver layer. Then we brought in the
4:38:15batch data the customer data from
4:38:17Postgress SQL through lakeflow connect
4:38:20and ingested into bronze and silver. And
4:38:23after that we brought in the streaming
4:38:26data using autoloader as well from our
4:38:29volume. And after silver layer we did
4:38:32all these types stream to stream join
4:38:34stream static join fraud detection
4:38:36everything in real time. And then we
4:38:39sent the email alerts and prepared a
4:38:42dashboard. And we did overall
4:38:44orchestration using lakeflow job. And
4:38:47then we uh used Unity catalog to prepare
4:38:51all our tables in Unity catalog with
4:38:54proper governance.
4:38:56So these were all the details I wanted
4:38:58to share in this video. I hope my
4:39:01efforts that I put in to create this
4:39:03video were worth it. So if you think
4:39:06that those were worth it, please put it
4:39:08in the comment section and let it out.
4:39:10And if you have any doubt or if you come
4:39:12across any issue related to this project
4:39:15then also please put a comment and I
4:39:18will suggest that you do a proper
4:39:20hands-on on this project end to end so
4:39:22that you are more comfortable if you
4:39:25want to show it in an interview or
4:39:26something and then only you will not
4:39:29forget anything once you do a practical
4:39:32and you can also get the PDF that I have
4:39:35prepared that you can use as a revision
4:39:37for this project whenever you want to
4:39:39have a quick glance so you can go over
4:39:41the PDF and use it so that you can show
4:39:45it properly in the resume and you can be
4:39:47ready for the interview whenever you
4:39:49want to showcase this project. So that's
4:39:52all for now. I hope you really liked it.
4:39:55So let's stay in touch and let's see
4:39:57what's coming in the next video. For now
4:40:00that's all. Bye-bye. Happy learning.

About this transcript

This page contains the full transcript of End-to-End Spark Streaming Project | Databricks + Kafka + PostgreSQL + Real-Time Dashboard by DataBeli, generated from the public captions YouTube serves with the video. The transcript has 41,939 words across 5,864 segments, with the original timestamps preserved so you can click any line to jump to that moment in the embedded player.

What you can do with it

Use the transcript to take notes, quote the speaker, build a study guide, generate a summary with ChatGPT or Claude via the YouTube Summary tool, or export it as a timed subtitle file with YouTube to SRT. You can also re-open it in the transcriber to choose a caption track available on the video.

Free YouTube transcript tool

YouTube2Text is a free YouTube transcript generator — no signup, no daily limit. Paste any YouTube link and get the full transcript instantly, with timestamps, click-to-jump, caption language selection, AI prompts for ChatGPT, Claude, and Gemini, and exports to TXT, SRT, VTT, or Markdown.