Data Engineering with Python and AI/LLMs – Data Loading Tutorial

Key Takeaways

Welcome to this course on mastering data ingestion for data engineering. Learn to tackle common pipeline features like schema changes and API limits by adopting the mindset and practices of a senior platform engineer. This course covers essential techniques including extracting data from APIs, automatic schema management, incremental loading, and orchestrating scalable automated workflows using modern tools. Alexi and Adrian are the data scientists who created this course, and dhub.com provided a grant to make this course possible. Hi everyone, welcome to our course about the basics of data injection for data engineering. My name is Alexe and I have been in the software industry since 2010. Last 12 years I focus on data and creating data pipelines. I'm the founder of data talks club which is a community of people who love data and there we teach precourses. One of these courses is data engineering zoom camp a 10week long boot camp about getting started in data engineering. I am the instructor for the first part of this course. Here we will cover the basics of data engineering. Specifically, we will talk about things like data pipeline, data injection, ETL and I will also talk about extracting data from APIs and putting it to a data warehouse. In the second part, we will talk about DT the standard tool for data injection. It will be covered by my co-instructor Adrian. Let's get started. You've built a data pipeline. The script works. Everyone is happy. The data flows, reports update, everything looks great until one day you come to work only to realize that the job failed. You start on investigating why turned out that the schema changed and now the script no longer works or API requests hit rate limit and your script fails because there are no retries and this leads to entire job failing and the script you wrote using AI is now a spaghetti mess and nobody wants to maintain it and um suddenly the entire engineering team is firefighting instead of creating new data pipelines instead of creating new reports and uh nobody is maybe and nobody trusts your data. If it sounds familiar and you want to avoid situations like that in the future, this course is for you. And the reality is that most engineers can write a script that moves data from A to B. But the scripts are not always reliable. Senior engineers think ahead. They design for resilience, scalability, and automation. In this course, you learn how to deal with situations like that. This course is not about basic ETL. It's about approaching adjustion like a senior platform engineer so that you can build data pipelines that work that are resilient even when things around them change. You already have a picture of what may happen if you ignore best practices. You spend more time fixing than building new data plans. Data breaks unexpectedly leading to trust issues. The management no longer wants to use your dashboards because they don't trust the data in these dashboards. Your team avoids your pipelines because the code there is not reliable. It's not easy to understand. It's not easy to maintain and scaling these pipelines is painful and it requires constant refactoring. So this is not a fun situation to have and senior engineers don't want these situations to happen. They design injection pipelines that work now and in the future even when things change and this is exactly what we will talk about in the course. So this course is for analytics engineers, people who want to ingest structured and unstructured data, handle schema evolution, enable self- service analytics. This is also for data engineers, junior and mid level who want to move beyond creating scripts to building production ready data pipelines who want to automate data injection and scale efficiently. This is also good for analytics managers and architects who want to design reliable, scalable and costefficient digestion for analytics, machine learning and operations. By the end of this course, you'll go from moving data from A to B to engineering injection systems. So you're not just reacting to failures but preventing them. In this course, you learn the following. So first of all, you we will learn extracting data from APIs and handling API challenges. So we will work with back services. So we will need to handle situations like authentication rate limits uh penation in order to extract data efficiently. Then we will also cover schema management and automatic normalization. We will use DT to infer schema from the JSON data. Then we will flatten nested JSON. We'll extract list into children tables. And we will see how to handle schema evolution automatically. We'll also handle we'll also look at incremental data instruction and state tracking. When our jobs need to reprocess the same amount when our jobs need to reprocess the same data over and over again, this is not effective. So we will learn how to load only new or modified records and thus avoiding unnecessary reprocessing. Then we will see how to load data into various destinations. So first we'll look how to store data in duct DB but then we'll move beyond duct DB and store data in data warehouses such as BigQuery, Snowflake or data lake and all of that will happen automatically. Finally we will talk about automating and orchestrating pipelines who will deploy, schedule and maintain injection workflows with tools like Dexter, GitHub actions and chrome jobs. And finally we will also talk about scaling data pipelines efficiently. We will see how to handle large scale data ingestion while optimizing for performance, retries, and parallel execution. By the end of this course, you will not just know data ingestion. You will be able to build an API injection pipeline that automatically detects schema changes, retries intelligently, and scales with demand. If you've been in data science or analytics, you know that you usually have a clean data set that you can use. um somebody prepared this data set like you have a table with all the columns you need without any data issues like you don't have any data duplicates and what you do is just take this data and train a machine learning model on top of that or you build some analytics on top of this data but the reality is that this data does not just magically appear in your data warehouse somebody has to process this data has to make it available and has to clean this data such that you as a data scientist or data analyst can use it for your job. And in this particular case, the data that um data scientists or data analysts use is already well structured. There is an explicit schema and uh we can use this data immediately. So examples could be packet of raw database tables and so on. But usually as we know the data does not just magically appear there. Somebody has to put it there and usually the source data is raw. it's uh not clean. It doesn't have schema. It can be JSON file. It can be CSV files. It can be some row logs from our server. So, somebody needs to process this data, normalize it and um sometimes aggregate and then put it to our table. Here in this course, we will learn how to be the magicians who take the raw unstructured data and turn this into well structured data. So we will be the one making data sets magically appear and in order to do this we need to extract data from various sources APIs databases files then normalize this data by transforming cleaning and defining schemas and finally load this data to the place where it can be used data warehouse data lake or database and we usually call this entire thing data injection. So the process of turning unstructured data into structured data this is what we call data ingestion. So we extract data from a source transport it to a suitable environment and prepare it for the use. So usually it includes multiple steps. So first we extract data from the source. Then we normalize data. Then we clean it. We remove inconsistencies, duplicates. We add metadata such as schema or maybe we need to look something up. And then finally we'll b the data to a data warehouse or some other environment where it's ready to be used. And people who take care of this process are data engineers and they're extremely important in this process. So the data engineers are the architects behind these pipelines. They don't just build pipelines. They make sure that they are reliable, efficient and scalable. So when the data when they know that data scientists and analysts expect the data they make sure the data happens and that the end users can trust this data. Beyond pipeline development data engineers need to think about the use case and select the best storage for keeping the cost low and performance high because every use case might be unique and we need a specific data storage for this specific use case. They ensure data quality and integrity. There could be some duplicates in the data source that need to be handled. There could be some inconsistencies that need to be removed. There could be some missing values that need to be added back to the data set. Then also it's important that only people who are supposed to access the data can access it and the others cannot. So this is uh this is called data governance and data engineers need to also think about that. So they need to implement governance for secure compliant element and well-managed data. So the role of the data engineers is to manage the entire data life cycle from collecting to consumption. Now we want to extract the data. We have two options. We can process it in batches and or we can process it in steaming. When it comes to steaming, when event is generated, we want to process it as soon as possible. So we processing we are processing data continuously as it arrives. In case of batching we let events accumulate and we process it periodically. So could be every 5 minutes, every hour, every day or every week. So batch processing is good for scheduled tasks and um stream is ideal for realtime processing use cases. Choosing the best approach depends on factors like data volume, latency requirements and system architecture. But typically the latency requirement is the most important one. Do we have to process data immediately as soon as it arrives or we can wait for some time before we do this. In this course, we will focus primarily on batch processing as this is the most common way of processing data. We'll briefly touch on streaming. uh but uh the majority of applications in real world data engineering are batch batch processing is best when you can wait for data to accumulate before we process it and we process it in large chunks. So it's more cost efficient and works well for non-timesensitive work codes. So for example we want to update a database uh during the night or we want to generate daily or weekly reports. When it comes to streaming, we want to process data in real time or close to real time with minimal delay. Common use cases for streaming include fraud detection. We don't want to wait till fudsters do anything bad to our system, we want to catch them immediately as soon as possible. And there are many other applications like for example in EoT when we have some sensors attached to our production devices and if something happens to the devices, we want to detect this as soon as possible. And there are of course many many other use cases. And when we want to select between batch and streaming first of all think about latency. So if we can wait minutes, hours, days, then we need to go for batch. If we cannot wait, so we talk about milliseconds or seconds, then it's streaming. And with streaming it's continuous small events while in batches it's usually a chunk of data. So it's way larger. It's not individual events. Use cases for batch include generating reports uh ETL jobs and stream processing is about real-time analytics and event-driven applications. When it comes to complexity, steaming applications are not easy to maintain. So it requires a complex event-driven architecture while bite jobs are easier to manage. And when it comes to cost u usually batch processing is cheaper because we don't uh we already know what kind of servers we need and we can provision the servers just for the time of the job while for stream processing we need uh our applications constantly up and running which results in higher cost for always on processing. There are many tools on the market that can support both bot and streaming. So first there are tools that are specifically designed for steaming or more designed for steaming. So these tools are Apache Kafka, Rabbit MQ, AWS, Kinesis, Google WhatsApp. Of course we can do some batch workloads there but people usually use it more for event driven architectures. And there are tools um for building pipelines tools like Apache Spark, DBT, Flink, Nifi, AWS Glue, Google Cloud Data Flow and DT. So these tools um some of them uh work better with steaming, some of them work better with budge um but uh many of them can handle both. So you can choose whatever tool you want. In this course we will focus mainly on DOT. So this is uh the standard tool for DP injection. So it can automate API instruction, incremental adjustion, schema evolution and it works with both batch and streaming pipelines. Okay, now let's go to the practical part. So we talked about processing data in batch versus streaming approaches and imagine that we want to extract data from an API data source. When it comes to batch we when we use Python we can use the request library to send a get request uh to our API and then get back the response process this and put this results into a data warehouse. So the use case we have uh for this course is um the events that we have from our one of our repositories. Uh this is our course about data engineering data engineering zoom camp. And every time something happens in this repository somebody starts a rep or somebody forks a repo. Uh we see this in uh this endpoint. So the endpoint is slash events right. So these are all the events we have here. Um so we see the fork event. If we scroll down, we will see a watch event. This API is well doumented. We can see all the event types. So for example, watch event is when somebody starts the repository. But we can see that there are many many other events that um we can have here this API that this API returns. So what we want to do is we want to visit the URL and we want to send a request to this URL. Want to send a get request to this URL. it will return this JSON and we want to process this JSON. We'll use the request library. I'm going to start by importing it and then I'm going to send a get request to this URL. First let me name it a URL and then requests.get and then we get the URL. We save this to response. And since we know that this is JSON, we can type response to JSON and then this way we send a get request to this API. We get back the results and we display these results. So we see this is the same data as we have here. I am going to hide it. And this is how we send this is how we get data for batch processing from AIS. We see that the protocol here HTTPS it could be HTTP or HTTPS and we typically send the get request. We get back some responses and then we do something with this responses for example putting them to a data processing them putting them to a data warehouse and building some analytics on top of that. When it comes to streaming it looks a little bit different. So I'm not going to live code it right now. I just want to give you a short example. So here uh instead of sending a cat request, we subscribed we subscribe for events in this team. So usually the URLs that we have for streaming APIs, they start with uh WDSS protocol which is websockets and then you define a listener, you say every time that there is a new event, I want to do something with this event. So could be putting this event to a cafka theme or I don't know reacting somehow to this event. So this is how typically it looks like and uh we will not focus on this part here in this course. Okay, we have seen how to send the get requests to get the data from an API endpoint and how to we haven't processed it yet but we have seen some data that now we can process but that's not the end of the story. So we cannot just send a get request get the data and assume it will work like that forever. There are some challenges with extracting data from APIs and the common challenges are first of all there could be some authentication involved. So when there are some endpoints that are restricted only people who are supposed to access it can can access it. So we need to tell the API that yes we can do this and this usually requires a different authe authentication mechanisms like including API keys and so on. Then the events that this endpoint returns this is actually not all the events. So if we can see that the last event is from the 6th of May from the 6th of March but there were events before that. How can we access that? How can we access this data? So for that we need to uh use the concept of pagenation. We need to be able to access holder data. Then there could be also rate limits. If we start if we start bombarding this API with requests requesting the same data over and over again at some point they will say hey slow down you're sending too many requests. We cannot handle this request from your side. And they will just start returning an error code instead of actually returning the data. So we also need to take into account and if something like that happens we need to slow down um and come back to request and the data again later or could be some network failures for example something between u the computer where the job is running and the server the connection is interrupted and we cannot uh request the data anymore. So then the network the API uh the URL becomes unavailable for some time and we need to retry later to be able to access this data. And finally there could be some memory issues when the API returns so many data points that we cannot process at once. we need to um gracefully handle the situation and uh u perhaps uh introduce chunks uh um and not process all the data at once. So we need to keep in mind all the snakes when we deal with APIs. And um some APIs they actually some of the API developers they are actually kind enough to provide a special URL where we can see what are our current rate limits. So for example the case of GitHub they have a special endpoint rate limit and we can see what are our current rate limits by sending a cat request uh to this um endpoint and then when we do this we can see that uh okay so these are our limits. So we have 60 requests. Uh there are 59 uh remaining requests and the reset of the rate limits will happen at this particular uh time time point. Right? So for example we can then access this rate uh remaining. So this is how we can uh see okay we have 59 remaining requests and our code remaining and for example if we want to if we hit the rate limits and uh we need to handle this situation. So typically when it happens we need to sleep for some time. So for example we can wait for 30 seconds before retrying and uh so for example in code it can look like if remaining equals zero then we sleep for let's say 30 seconds and of course we need to import time for that. So this is how we can handle a station like that when we are running out of um when we're hitting the rate limits. Then in addition to rate limiting we um can come across authenticated endpoints endpoints that require me to prove that I'm actually that it's actually me who is running this request. Um so an example again from API is uh /point user. So if I try to send the request right now to this endpoint then we will see that it's responds with 401 error code. Um so here we actually see if we look at the request it says it requires authentication and here is documentation about that. So this is um this endpoint is not it's not possible to use this endpoint without a proper API key and I already have an API token that I can use. So I went to my GitHub settings and I generated an API token. So this API token is um stored here in the secrets. So I can use this token now and um right now I'm running on Google Collab. So in Google Collab I can use uh this user data to access the token. But if um um you're using it elsewhere, if you're running your code elsewhere, you can use environment variables for that. Um so we have the API token and uh the way we authenticate is we include this token in our um headers. So there's it's authorization and then the standard way of doing this is in including this into the header like that app token. And when we send a get request, we include this um we include this in in the request. So now this is my token and we see all the information about me. This is how we deal with authorization. And the last thing I wanted to talk about here is pagenation. So here um we see that um this is definitely not the last event that um we generated that was generated in our repo. If we want to access the events um before this date, we can um use specify the get parameter page and say okay we want the second page and then this way we can go back to um to see all their events right and this way we can iterate over different pages and like for example it is page 10 um and um yeah we can iterate over all these pages and collect all this data and process as we need this Again the same URL we use a get request to send data to this URL and uh right now if we want to understand what is u okay we send the request to this page I'll send the request to this page so it returns uh some JSON data and we wanted to understand what is the next page for this and they the developers of this API they include this information in uh the headers of the response. So we can see that there's a link header. I can access this header here. We see that for this particular request for this particular endpoint the next page is um this is how we access the next page by adding the parameter uh page equals to and there's also the last page page number 10. Right? So now what we can do is we can iterate uh we can have a loop and then every time we send the request we see okay what is the next page and this way we process all the events until uh yeah we reach the last one and for example if we actually send a request for the last one it looks a bit different. So there is no next link here there is only previous link. So what we can do is we can just check if next is there in the links and if it's there we take the link we send another request and we keep doing this until there is no next. Um then this thing is not actually easy to parse. We can parse it because there is structure but we don't want to parse it ourselves and um the requests library already does this for us. So we can just access links um by using links links there. Um and um yeah let me check a different page. So if we do it page number nine we have the next link. So what we do is we check if next is there among all these links and if it's there next if it's there we just access the URL. So, and we uh keep the rating until we get all the pages we need. So, let us implement that. So, I'm going to have a while loop while true. So, we will do it forever and uh we'll break inside the loop. So here I want to check if next is not in response links then we break loop but if it's there we get the URL and the URL is this is how we actually access it right so in this way we send a get request and we get the new URL and we keep doing this until there is no next link and now we can do something with the response. So for example I can do the length of the data and data would be in this case [Music] response.json. So let me execute this. Yeah, of course I need to start with the event num with uh the first page and we see that it has each page has 30 events uh except the last one which has 22 so and 10 pages in total. So this is how we process uh this is how we handleation in case of APIs. Um yeah so different APIs uh different API prov providers handle the situation a little bit different. So in this case in case of GitHub we get it from the header information from the links section. Some may explicitly return this in the response and we need to adjust our code to each of the API providers. Okay. So now we know how to handle pion and let's say we want to process all this data. We want to take all this data and put this to our data warehouse. So we can run a script like that and simply append this to a list and then at the end once we collect all this data we load this to a data warehouse. This could be problematic if we have a lot of pages. In this case we have just 10 pages but what if we had uh thousand pages and each page contains not 30 uh items not 30 events but uh a million events then we will quickly run out of memory. Our job will crash. So we'll need to do something about that. Instead of collecting everything at once and then processing, we can um do this in chunks. And um since this API already uh kind of chunks the events data, we can for example use one we can process one page at a time. And for that I will take the code we have already written and modify it slightly. So we will write a special function that will call events getter and here when we get back the data we will so this function events getter is actually a generator. So it will return u every time we request a new page it will return this. So instead of accumulating this in a list and returning everything, we use the yield keyword to actually at the moment when the page is ready, we immediately return and the client like whoever is using this function will process this one page at a time. And uh so now I'm just going to use this uh generator here. I'll call it events pages and we can trade over uh these pages and for example uh we do some logic here right now I'm just going to print it u but this is where we do some processing right and now we send a get request for each of the pages and uh instead of accumulating and all the pages we process them one at a time. Okay, so we know now how to request pages. We talked about all the possible issues that might arise and now we are ready to actually do something with this data because we cannot simply take this data set as is and insert it load this into our data warehouse. We actually need to do something cuz in our data warehouse we have tables and this is JSON JSON is unstructured. Um so we have uh nested uh fields here. So for example actor is a nested field and we in order to put this into a table format in order to convert it to a table format we need to uh we call this process uh unnsting adjacent fields. So this is one of the things we can do. Um there are other things. So we often have a station like um we have lists in the quests. So for example here we have a list with topics. This is for a pull request event. We have some topics right? And we need to also be able to do something with this to also um flatten this list. And there are many other things. So we can add types, we can rename columns like all these things um are important when we are converting data from the row format to table format when we want to load this to a data warehouse. And um this is uh let me take a look at one of the events that we have and take page. So this is our event and right now we want to write some code that processes this event and flattens it. So I will call this function uh process event. Let me just print it one more time. So this will be the function that is uh taking the row event and returns an event that we'll later put in in our data warehouse. So we'll start with an empty dictionary and right now uh we first will just um copy some of the fields that do not need any pre-processing. So this is special p ID. So this is this thing uh then uh type. So for ID we simply copy it as is without any changes. Then we do the same with uh type with public and with created that. So we simply take the these events uh we take these fields as they are without any extra pre-processing and return them. And then we want to do something with actor. So actor is a nested JSON field. We need to do something about this. So we want to access ID and let's say login. And here we use the special naming convention. So when we access uh when we create when we normalize the data. So the naming convention is actor ID meaning that actor was a nested field and uh we split we say okay it was a nested field uh and the field inside this actor is ID and in Python this is how we access it right so it's a dictionary inside a dictionary and uh we also want to copy to login and now we simply return the result. Let us quickly test it. So I'm going to use this function on the event. Okay. So this is how we convert uh the original event that we had into a processed event. So I'll call it pro and then of course we can use this function for all the for all the events we have in our batch in our page. So I'll just do a for loop or event in um events page and then let's say we can accumulate all these events into a list that I will call processed events. Okay. Okay. And now at the end we just simply can see what is inside. It contains all the events with that we have but they are already processed so they are uh normalized. And uh here we copied created at as is we didn't do any changes but actually we can uh parse this too. So this is um is in a special it's stored in a special uh ICO date time string and let's say we want to parse this string and store it as a Unix time stamp. So we can do this [Music] um right now we will take our function. So let me find this. We'll take our function for processing events and we'll adjust it. And here yeah I already imported datetime and in case of uh so in case of Python we can use this from ISO format function for parsing uh parsing the data so pared stump And I can now extract the Unix time stamp from this using the time stamp field. Yeah. So when I do something like that and let's we can immediately check how it looks like for our let's sorry okay now we can see that uh the created at field in the process to looks like that. So right now it's a Unix time stamp. Okay. And then there's another thing we talked about. So here we want to extract the these topics and these topics the structure is it's quite nested. So topics leave inside uh we'll see inside repo and then repo is inside base then this is inside uh what um pull request and pull request is inside payload. So this is super nested and uh what makes the matter even more complicated is not all the events actually have this. So we want to be careful and um or what we are going to do right now is I'll just use the uh get function of a dictionary and I'm going to get the payload and if payload the payload field is not there I'm simply returning an empty dictionary and then um from payload it's pull request again empty dictionary if it's not here then base then um repo and then finally topics and if again topics is empty then we return an empty list and let us just uh return topics. Now I can do this for all the data for event infall data uh process event event and uh yeah we will put [Music] them inside processed event list. Uh we have already written this uh but also now we need topics. So I'll call this processed topics. and topics. And what we're going to do is we're going to extend this list with topics. Extend topics. And then of course here we append to the events. We append to process that. So now we can see we can just print uh both of them. No, it's I guess too much. Uh let's just uh look at first five and I'm going to actually print them. So now yeah we see that uh these are our events and these are our topics but as I said uh we actually need to be able to link back the topic to the event and for that we need to have the ID we need to use the ID as the reference. So it means that we need to do a bit of extra processing for topics which I'll do inside here in the inside this function topics. So have a variable called processed topic which will be a dictionary. First we'll have event ID which is the ID and then topic name which is the topic itself. So now when I do this, yeah, of course I need to also put the results to process topics because we don't say them. Okay, now it looks much better. So now for each of the topic we know what is the corresponding aant ID. So we have uh two lists each of them contains uh dictionaries and these dictionaries is what we are going to put inside a data warehouse. Okay. So we prepared all the data uh this data is in these two lists um process events and process topics. Now they contain the dictionaries with the cleaned event data cleaned topics and now we're ready to load this data inside a into a data warehouse. So let us do that. Uh as a data warehouse we would use DDB which is a simple inmemory database that we can use [Music] and dis Okay. So, first we create a connection to a DDB database. This will create a file github events db. Next, next we will create the GitHub events table. And if we look at our event. So if I look u at process processed events we have um these fields we have ID type public created at actor ID login actor login right and we have all these fields here. So there's one to one mapping between this dictionary and table and yeah here are the types too. So now we want to insert all these processed events in that in this table. So we have all these fans. uh first we need to turn the dictionaries into a list of tpples and then we have this insert statement and we apply it to all the data. So now the data is there and we can verify that by executing a select star request and then um yeah let's see what are the first five events there. So we see that the data is there. We successfully loaded the data to our data warehouse to the GitHub events table and we have these six uh columns there ID type public created at actor ID and actor login. Cool. So we were able to do this and um before I forget let us also close the connection. So now the data is there the and we can now start building uh analytics. On top of that, we can start building dashboards and we can do whatever we want with this data. And even though it works, if we look at this code, there are actually many problems. So first of all, when we when we specify the when we create a table, we hardcode all the values of the fields. So we say that okay type uh is text public is boolean we have to manually do this and this is problematic uh if uh we have um when we want to do more then we have to um maintain a lot of code. So this is not fun to do this. There are other problems like uh we don't have automatic retries. Um we have some code that loss data to our data warehouse. But what if something happens and um let's say in a case we don't use duct DB but use BigQuery. Uh BigQuery is a remote data storage. It's a data warehouse that uh does not live on our computer and if there's there are network issues and something happens we cannot insert data. We cannot load data. We don't have any automatic retire mechanisms. Then also there is no incremental loading. So we take all the data we have and we load the entire data set into our uh data warehouse. But what if we want to run this code tomorrow, right? Tomorrow we will have new events of course but also we will have a lot of old events. We don't want to reprocess all the old events. We want to all only process the new ones. So this is we would solve this problem with incremental loading. And there's a lot of code to maintain like if you look at this code like it's not uh pleasant and we need to maintain it. We need to every time we add something new we need to also keep track of that. So this is not fun. This is a lot of work and um yeah we will need a team of engineers to actually maintain this code over time. So these issues are not fun. Let's say we want to address some of these issues. And the use case we'll have is we want to add a new column here. So I'll take the code that we already have for processing events. And let's say we want to uh add a new field. We want to add a new column repo ID because we want to run this analytics job on multiple uh repositories not just data engineers on camp. We also want to do it for our other courses and we want to know which event belongs to which repository. So for that we need to know the repo ID and we right now will add this. So add result ID which is events event repo id. So this way we will copy the repo id. We will also nest it and we get the ID of the repo. So the modification here in the process event function is relatively straightforward. Let us also execute it for all the events. So now if I look at uh process events, we see that there is a new field repo id. So now we have the new field in our um dictionaries in the processed events. Now we need to load them to load this particular column to back to our data warehouse and we want to be smarter this time. So instead of u hard coding the type uh the type information the scheme information um we want to automatically infer it. I will start with again uh opening a connection to our db. Then the next thing I want to do is I want to understand what kind of columns we already have in uh this table in GitHub events. So these are the columns we have and what we want to do now is iterate over all processed events and if we come across a new column that is not present in this list of columns uh in this set of columns then we create an alter table statement and then we added this to the set. So this is how we do it. We iterate over all the events we have and we check the fields the keys in these dictionaries and if we see that there's something new that something we haven't seen before. We try to understand what kind of type of this thing is in our case it will be an integer. So the column type will be big int and we will issue we will execute the alter table statement on the GitHub and events table. and we will add a new column. So let us execute this code and we see that u yeah we successfully added a new column. Now we need to load the data again and uh right now we don't have any new data but imagine that uh maybe we do it a few days later there are some new columns uh there are some new events that appeared there are some events that uh were before. So we cannot simply just redo the whole thing. So we need to be a bit smart about that. So this is the code we can use for that. Um so again uh first we turn our uh dictionaries into a list of uh toubles and then we construct a SQL query for that. And in the query when we insert into GitHub events we do this on conflict when there is already a record with this ID we actually update it. So we uh see okay the record is already there um but the field is missing uh so let's insert this field. So this is what this query is doing and now we can execute this query. Okay, this query has executed and let's actually take a look at what's inside our table now. So now inside the table we have the same information as we had previously but there's a new column repo ID and the repo ID contains the ID of the repository from which the event is generated. Okay cool. Um now the code is a little bit better even though it's more difficult to understand now but it's more generic. So every time we add a new column, we don't need to hardcode values. So this code will be able to automatically infer the type of the new column and add this to uh the table. Right? So this becomes easier. But as you can imagine with time this code will grow and it will become more complex. it will become more difficult to maintain this code and this is not something that um the team will be looking forward to working with because it will be uh a big spaghetti a big file with lost spaghetti code that is hard to maintain and even if you will take a take a lot of time to make this code really good it's still you probably don't want to spend time on this library you want to spend time on innovating on solving business problems on creating data pipelines or uh maybe new dashboards. Uh so you'd rather put your time there than into maintaining this code. And this is why you want to actually delegate this job to a special library that can do this for you. And this library is DOT and DOT can do automatic schema management. So it handles schema changes for you. Like when you add a new column, it will just automatically add this when you need it. Then it also can handle incremental loading. So it ingests only new or updated records. It can work with any destination that you can think of. So here we use duct db. But when we switch to DT, you can ingest data. You can load data into BigQuery, into Snowflake, into posgress, into anything you can think of. Then uh it also works better because there is uh you can ingest data faster with uh built-in paralization and u something we also don't have in our code is we don't have transactions we don't have uh atomicity the tries dot handles of all that so when you switch from your homegrown uh code for loading data to something like dot life becomes a lot more easier and this is what we are going to cover in the second part of this course. In the second part of this course, Adrian, my co-instructor, will introduce DT which make your life a lot easier. Woo. So, welcome to data engineering for senior professionals, the second part of the course, the advanced topics. My name is Adrian and I'm a data engineer. I got into the data field some 10 years ago. Um I did 5 years of startups, 5 years of freelancing building data warehouses and data teams and um then I started working on DT or data load tool. Uh it's a dev tool for data engineering that we will learn how to use um which I hope you will enjoy today. So why did we create the LT data load tool? Um it's basically coming out of my own need as a data engineer. uh one of the needs that I had was that it's fast and easy to create pipelines that then self-maintain and that also the rest of the team can use them or create them or you know so I saw for example that data scientists can all do pandas data frame to SQL I wanted it to be that easy for everyone to load data in engineering complete best practice kind of way so not like pandas to SQL um so this is DT the tool that will take your weekly typed JSON data or stuff like that and automatically um clean it up and load it strongly typed into a uh data warehouse or data lake. Um it basically automates the unpleasant part of the work. So like turning weakly typed data into strongly typed data. It enables the team and it solves most data engineering problems once and for all. So what I mean is for example you might want to parallelize a pipeline at some point. So at DLT, you know, once we added the ability to parallelize, you can parallelize all your DT pipelines. So it's the boilerplate code that I wish I had. This has already happened. So we created this library. We put it out there and our early 3K uh community already built over 20,000 sources. So it's a different dev tool. I would say we mostly care about enabling you and your ability to build with it. Uh you might ask how because it's a very common question. How do you thrive as an open- source project? And the answer to that is that we're an open core company. So we offer DT OSS for free, no limits, um and DT plus commercially. Um what this also means to be an open core company, it means that we aim to make DT an ingestion standard. Similar companies that do something like this are Confluent and they create Kafka or Elastic who creates elastic search. And in this course basically we will learn about how to do data engineering uh using DT because it already has the best printed and um you know how to have let's say a happier experience when you're working uh with your data stacks. So let's look at uh what we will learn in this session. So first we will uh learn about extraction. uh we will look at into different ways to request data from REST APIs. Uh then we will look at what a DT resource is which is basically an extractor and look how we can configure that. Uh then we will look at normalization. In the case of DT uh we built DT to automate this normalization because this is hard and tedious. So you can adjust how that happens. Uh or you can set a data contract or you can alert schema changes. Uh so we will look into how to do that. Um after normalizing we will move on to loading. Uh so we'll look into how to configure right disposition in a resource. Understanding how uh replace right disposition works a step-by-step example. Then we'll go to incremental loading uh which is uh significantly more complex. So we will uh first look into the theory of how that works and then uh we will go with an example and then we'll look over SQL to SQL copy specifics. So specifically uh because the data in SQL is already typed strongly typed uh we can skip the type inference uh so we can use arrow or connector X uh under the hood. uh I'll show you how to do back filling uh because this is a common question people have and how to historiize data with slowly changing dimension. Then we'll move to performance tuning which uh will teach you how to do memory management with DT so your pipelines never crash running out of memory uh how to do asynchronous requesting so you can uh essentially parallelize extraction and how to do parallelism on the rest of the pipeline. So for normalization and loading then we'll look into loading uh to data lakes and lakehouses because there are some specifics um around them. So like loading to park files um loading to uh iceberg and Athena uh talking a little bit about uh iceberg partitions and uh cataloges uh then we will uh learn about the specifics of loading to data warehouses or MPs. So understanding uh how staging works. Staging is essentially like a data lake for data warehouses and then we'll move to deployment and orchestration with some examples of how to deploy to chron job airflow or dexter. Um looking forward to guiding you through all of this content. So let's talk about data extraction. Uh we have this notebook here. Um I'm pip installing DT with TCD DB and um previously Alex was showing you how to do Python imperative API calls. Now I will show you how to do hybrid uh type of calls. So something between imperative and declarative using the rest client uh from DT. So the rest client from DT was actually built uh for this uh declarative uh API client uh that you can also use. Um and it offers you some components to make requesting data a little bit easier because it automatically detects some things or you can tell it what type of pageionator to use for example. Uh so it's very useful for mixing let's say uh Pythonic code with customizations uh with just a simple uh automated REST API client. You can read more about the docs here from this notebook. Um and you can also self-learn the REST API documentation. You can also find on YouTube some uh tutorials uh how to do it. They're quite comprehensive. So I will not teach that here because it would be quite long. Um so let's uh uh do the step-by-step example. So we had the uh previous example requesting data. So now we will go to documentation and copy uh snippet of the rest client and then uh work with the other elements and build our request. So just quickly going to documentation. Um so I'm just going to copy out this snippet from the top. And now let's implement it. First I will paste it here. Um we don't need most of these things. So we don't need a session. We don't need a data selector. We need a pageionator. But it's not this one. We need authentication. We don't need headers. And we need the base URL. So let's put in the base URL. I can copy it from here. Actually I have it this and let's use also the credential that we were using for the GitHub uh API. So the access token uh we have it in our variable. So let's uh read it there. So it's a collab uh user data can see it here or we need to give uh collab access to it in order to have access to it and we need to uh print the data from this pageator. I'm going to quickly look at the documentation [Music] again. Where was it [Music] here? So, how do I print it? Okay. Yeah, this is how we print it. And for us the endpoint is events. I think it's plural. And we don't have okay we need to put in our base URLs and for authentication we pass our API token and for the pagionator we actually need to use something called uh header pagenator. there. So let's look for header link pagionator it's called. So just header link pageionator. Let's copy this here. Let's import this one. And here our path was next. And uh in the documentation for the header link pagionator, we also have the variable name here. Unfortunately, it's different. Uh we should probably fix that. But yeah, this is how it looks. Let's see if it works. should get pages of data. User data is not defined. Oh, we need to import the collab user data. Okay. Now we basically see pages of data coming. So uh our example is working. Now let's convert it into a DT resource. So what is a DT resource? It's a function that yields data. So we just need to wrap our code in a function and make it yield each page. And then on top of this function we put a decorator. And what this does is it will allow us to customize our function. So basically customize how it will behave in normalization or loading or all kinds of things. So let's uh implement that. So let's define a method here. Um we were calling it pagionated getter. So let's do that. And we want to declare that this is a DT resource. And as you can see by auto suggests is already suggesting me what to do. Um now DT when you yield something it uh works with records but if you pages of records it's also fine with that. It will actually work faster. Um but DLT will break that page into records anyway. So let'