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