Transcription
Greetings to all future data engineers. In fact, Airflow is currently the market standard, but beginners are encountering more and more problems with it. And therefore, I want to show today how to deploy it locally in Docker, how to work with it, and where to look in the interface. We will write our first DAG today, and I will go through the code line by line, explaining what each line means, and also show the result of what we get. And if you want more info about DE, where there are current questions and help with employment from scratch, then subscribe to my Telegram channel. There, in the pinned message, are the top interview questions. For example. The example will be quite simple. We have an API, we have a database, in this case PostgreSQL. We want to do all this through, where we download data from the API and push this data into the database. For example, every day at 10:00 AM we need to fetch data from the API and push it into the database. And then this data from the database can be used to create, for example, data marts and then go to visualization, go for review by analysts. To do all this, we will deploy everything locally. I will show how it is done now. In fact, we will use Docker. And Docker [music] is a tool that has containers within itself, for example, for PostgreSQL and Airflow. And the whole point is that we take the entire application with all its dependencies. And when we transfer these applications to another machine, to another, for example, computer, it will work exactly the same, absolutely identically. [music] That is, there can be no such problems as this Docker Compose working on one machine and not on another. And Docker Compose is a command that allows you to deploy containers on your machine. Docker has been downloaded. Let's open it. So, Docker has opened. In fact, we don't change anything here. Click OK. Let's wait for it to finish. Docker has been installed. The only thing is that it will ask to restart the computer later, don't be afraid, everything is fine. Accept, click. Most likely, you will see something like this. You can try to enter the cmd. It's better to enter PowerShell. Please copy. You just take it in PowerShell and paste this command, and it will install the subsystem for you. Why is all this happening? Because Docker doesn't know how to work with Windows very well. It was generally made for Unix systems like Linux or Mac. And therefore, in general, it doesn't work well or even stably on Windows. The system is complete. Now, it will think for a bit. It will spin for about two to three minutes. After that, it will show you something like this: Welcome to the Windows Subsystem for Linux. You can read here, explore to get started with Linux. All this, if you really want to. Really want to. And it will look like this. Here we will have containers, images, i.e., snapshots, volumes. You can throw Kubernetes in here, there are its own builds and so on. But now we are more interested in containers. I will show you how to do it now. So that you don't manually write code yourself, to deploy Airflow and PostgreSQL on your own, there are these wonderful people who have already done all this for you in advance. First, you need to clone the repository, but first you need to install an IDE, i.e., a development environment. I installed VS Code for myself, you can take anything you want. There's IntelliJ, there's PyCharm, and so on. I just like VS Code more. I'm just more used to it. So we take all this, copy it. Next, we need to install [music] Git for all this stuff. Let's go to the Windows store. Ah, download X64. There will be a lot here, just click through. We can, for example, go here, type Git version, so that it tells us what version we have. Cool. Let's restart. Here we took it. Let's go here. Let's clone this thing again. This is cloning the repository. Here, click clone for repository. Paste it. And it will clone it for us now. The repository has been cloned. Ah, click open. Trust. And here is the repository that we will use to deploy everything on Docker. And we need the docker-compose.yaml file. The configuration is described here. What do we have here? Executable cores. Here we have services. By the way, there is Superset. Superset is a data visualization service. For example, we transferred something to PostgreSQL. Everything is there in PostgreSQL. Superset pulls it and creates nice dashboards. Here is Superset again. It has workers, builds, PostgreSQL. This is our PostgreSQL, which will be pgAdmin. Well, I will probably use it through DBeaver, but there is actually no significant difference. PostgreSQL for Airflow is a backend database that PostgreSQL uses internally for its needs. Redis is also used by PostgreSQL. And classically, Airflow has a web server, scheduler, and worker. By the way, there is also Triggerer. Cool. And initialization has also been added. You don't need to read the code too deeply here. You just need to roughly understand what components are in this YAML. You don't need to read into it deeply, try to understand. Look at the boilerplate code, start figuring it out. It won't bring you anything. Next, we go, we can go to the README, here we can look here. In fact, for starters, we need to be in the bash here, I already did it with Git Bash, I can do it again to get the user identifier. Mine is like this. Then, in the root here, create a .env file here, write airflow and what you got in bash. Like this. And only after that, run docker-compose. And how to run docker-compose? Go to the terminal, create a new one, type not like this, type docker-compose. Here it takes images, i.e., snapshots. Here we have Airflow, Superset, Redis, PostgreSQL, pgAdmin, and PostgreSQL itself [music]. There are two of them, if anything. One PostgreSQL for Airflow, the other PostgreSQL is the one we will use now. It pulls them, i.e., it gets them from the server, where it downloads them from. And just literally wait for it to finish [music] its magic. Then, when it has pulled everything, you will have this beautiful scroll of logs. Don't be scared, everything is okay. Next, we will probably connect the databases first. For databases, I personally recommend downloading DBeaver. Literally, it's downloaded from here by the link. I have already downloaded it. Then we just go and add new connections. So, here we choose PostgreSQL. Then we go to the README. Everything is indicated here. So, we have user, password, PostgreSQL. They are actually PostgreSQL everywhere. [music] And here we specify, so, some localhost. So, in this case, localhost will be here, but it is. Localhost. Here we have 5432 for PostgreSQL, test connection. I have already downloaded the driver, and it might ask you that something is not installed and there is a download button. Just download it, check with test connection that it is all connected. Done. And it appears here. There are no tables yet. And let's go. Let's do the second one too. It seems to be 5052, right? And here is the second PostgreSQL. It is probably also empty for now. Yes, it is also empty for now. We have configured the databases. Now we need to go to Airflow. Here is its URL indicated, so you literally take it from here, copy it, paste it here, everything is up. I have already logged in and checked, username airflow, password also airflow. It's all default, and you log in here. There are different ones. They are here from zero literally. How do we add new ones here? Here in the repository, we have a folder Airflow. DAGs. Here it is. Well, for example, create a folder, or rather a file du_test_2.py. And, for example, take code from here, copy it, purely for example. And put it here, so2. Ah, you will have to wait about half a minute, maybe a minute, and after that it will appear. Yes, in data engineering, nothing is fast. Sometimes, when you are testing one DAG, you can spend the whole day just adding some new logic to the DAG and waiting a minute for it to apply and appear in the interface. And then you look at the logs, everything is wrong again, and you work in such an iteration. I'll even time it. Oh, it appeared. 2 minutes passed. B3, well, two. Here is our DAG literally. Well, here is the code. This is the code that we copied here. Why is all this happening? It's quite simple. We go to Docker Compose, go to DAGs, for example, and look for volumes. And in our volumes, it is indicated that it takes from this DAGs folder and mounts it inside the container. So they listen to each other. And when you add a file here, after some time the container itself updates and the code appears there. And so you take and in the future do all your testing for projects this way. That is, you add some new DAG here, wait, well, apparently two minutes, for it to update, and it will appear there. Now I will redo my code. For our example, I did it in my infrastructure. It was quite fast there. Where is it? Where is it and blah blah blah? Show yourself. Ah, here is this code. Now I will transfer it, and show how it works. By the way, I forgot to mention, before we connect to Airflow, we must configure the connection. So, I have configured it now. How does it work? So that you don't struggle for half an hour like me later. The whole point is what? The PostgreSQL client in this case is the host. That is, we connect via localhost to the host, but Airflow is inside the container, it needs the internal name of the database. Our client will be PostgreSQL. The schema will be the database, i.e., PostgreSQL. I'll show you now. This PostgreSQL. Login and password, as here, PostgreSQL. PostgreSQL. We specify not the port we forwarded and by which we connect here, but the internal port, i.e., 5432. This can be seen. This can even be seen here, yes? Here the external port is 5052, by which you connect via DBeaver, and the internal port is 5432. You configure all this by it. Here you specify the type as PostgreSQL, connection ID, specify whatever you want. So, I specified this one. Then you save all this. I have transferred the DAG that I wrote on my infrastructure, but I will talk about that a little later. And you just go in and look. I have already executed it, checked that everything works. It turns out like this. Let me delete it, and we will look at how it is executed again. Here it has been executed. This column consists of four tasks. We will now break them down, how to work with them at all. That is, you can go inside, go to logs, and here you will have logs of whether it was executed or not. For example, it has been executed now. So, a table was created in this task, but, for example, as it was here, it is red, meaning it was not executed for some reason. And you can go to the logs and look at the bottom. traceback is everything that it indicates as a Python error. What is wrong with it? Here it says that 54 failed, meaning database public does not exist. Why did this happen? Because in the connection here, I specified the wrong schema. That is, I specified public, and therefore it complained and could not connect to PostgreSQL properly, and therefore it failed. So, this is a case of why it doesn't work for you and how to debug it all so that it works for you. So, first of all, in most cases, we look at the logs and from them we deduce what needs to be corrected in the code for it to work. It has worked again. Let's go here, refresh. The users table has appeared. Data has appeared. Where did the data come from? It came from JSONPlaceholder. This is a special service, where we have fake data and we can fetch it via API. I will probably leave the URL in the comments or description. And here we have the first ID. For example, here is Lina Gram. Here she is. Username Brad, and email. So, this is how it all works. So, now let's see what is happening here. This is the DAG code itself. At the top, we have imports. That is, we import some Python libraries that we will use here later. That is, in the context of Python, Airflow is essentially just a library where we import the necessary operators, hooks, and so on. Next, we have the base URL, that is, this JSONPlaceholder, which we, which I have already shown. Let me show it again. Next, we have two scripts: creating a table and adding some data to it. Ideally, they should be moved to separate files, but for clarity, I left them here. Next, we have a wrapper function. That is, through this function, we hit the URL, where we add a specific endpoint. In this case, it's users, to access the necessary data. I get some data from there, check if there are any errors. If there are, an error will be displayed in the logs. And if everything is okay, it returns JSON. Next are the default arguments, i.e., who wrote this DAG, will this DAG depend on previous runs. That is, these old columns that we have, these, so if there is a dependency on previous runs of this DAG, from what date it starts downloading data, i.e., not when we launched it, but from what date, from the perspective of historical data, it will load this data. And in case of failure, how many times should it try to repeat, the algorithm itself, if it had some error the first time and something didn't work out. For example, the API had a timeout, i.e., waiting time, and it failed for some reason, and we retry it again, to avoid this error, if it is specifically a timeout problem. Next is the description of the DAG itself. We have tags that you specify. They are visible here. Here they are. Next, catchup. That is, if a day was missed in the DAG, will we load this day or just move on? Schedule interval is responsible for when we will automatically run the DAG, for example, every day, say, at 10:00 AM. Next are the default arguments, which are here, and the name of the DAG itself, dag_id. Next, we have the logger. This is a logger. It is needed so that it is convenient for us to view our logs in Airflow and track some errors. I showed them just now. These logs. It is much more convenient to see and display them here than if we just write ordinary prints using Python code. And then our tasks begin. The first task is to get users and log them, that we got them. All this, by the way, goes into XCom, which will be passed to us here. That is, we get them from XCom here. That is, in one task we send this data, in another task we receive this data to do something with it later. In this case, we need this data to load, insert it into the table. This task essentially just creates a table using PostgresHook. In PostgresHook, we simply specify the connection ID that we described here. That is, here con_id, we use it inside this method as a parameter. And then we write run, where we specify create_table, which we have here. So, it essentially creates a table for us. And we log this again to understand that the table has been created. Next, as I said, here we get users via XCom, check them, that they exist, and then add them, in a loop. That is, we take a loop and add each individual user, so that it's not all in one line, but so that there are 10 of them. Next, we connect to PostgreSQL again, create connections, cursors. We insert the data we need there, commit, and close all this. And, of course, at the end, we log again. As you understood, we log any of our actions to understand if something fails at some process, at some task, we have a general idea of where it is and in which part of the code we should dig to find the problem. So, the functions themselves are created here, and then in the tasks themselves, where we specify the name of the task, and in Python callable, i.e., callable Python code, we pass the name of this function. For example, create_table, it is passed here. And we specify the DAG to which all this relates. And in the same way, with all three functions, we convert them into tasks. Below, we specify how these tasks interact with each other, i.e., what is the sequence of these tasks. That is, we can execute, first the table creation is executed, then the user retrieval is executed, then the user loading is executed. But, for example, we can do something like this. That is, we can specify it like this. This will mean that the first task is executed, and all subsequent tasks are executed in parallel. So, for example, in this case, what will happen? We might get an error that it will load users who have not yet arrived. That is, we have not yet received these users, but we are trying to load them. Because of this, there will be an error. And then, in this context, in this case, it will be easier to create the table and retrieve users in parallel. That is, we create the table in parallel, retrieve users in parallel, and only after that we add them to the table. That is, it will look like this. Now we have analyzed a DAG example on local Airflow, but in real work, it looks something like this. You don't do anything manually in terms of deploying services. You click a couple of buttons, work with Airflow, and write code. But, unfortunately, there isn't much of this available in the public domain. This is very much lacking to understand how it is actually done in real work. And therefore, I decided together with DevOps to create and deploy an infrastructure exactly as it is in real work, as close as possible to what can be. Therefore, let's just take our DAG example that we have already done locally, and I will show how to do it on the infrastructure. So, we go to Git, we have it and blah blah blah. We open it in VS Code, it has already been cloned to us. I have written a new DAG, already committed it to Git. Then we go to PostgreSQL, connect it here. Everything, done according to the pictures. Here we choose. So. Everything is here, a table with how we do all this and what we add. Host, port, database, login and password. So, here our login and password are like this. Ah, done. This is mydb. This user has a schema and blah blah blah. In fact, here it is. Here are the users, where I checked that everything works. Let's delete it. Nice. Everything is gone, no tables. Now we go to Airflow, look and blah blah blah. Here we have our test DAG. We run it. So, everything has worked. Cool. Here is the data that we get. Let's go here, refresh, and the users table is already there. That is, you don't need to install Docker yourself, figure out how it works. You just go to the necessary services, connect everything, and start working with it from scratch. And you can get access to all this for only 2k and only for the first fifteen people. There I added additional tasks for Airflow, added additional tasks for SQL, added theoretical blocks for both Airflow and SQL. Later, I will add for DBT, for Spark, and in general for all technologies. And the price will increase. Those who join now, the price is only 2k and access will be forever. Write to me in private messages on TG.