Transcription
Hi everyone, welcome to this tutorial. Today, I'm really excited to introduce you to Prefect.
Prefect is a tool in Python which we can use to automate and manage the workflow and data pipeline in Python. If you come across writing complex programs which involve multiple tasks to execute in a sequence, and if you want to handle all the errors and you want to schedule the execution, Prefect is the tool for you.
Prefect makes it really easy for you to define the tasks, schedule them, and monitor the workflow. Whether you are processing data, or executing machine learning algorithms, or if you are automating something, Prefect can be really useful for you.
In this video, we are going to start with the basics of Prefect. Then we will see how you can write Prefect tasks and combine these multiple tasks into a flow. Then we'll see how Prefect executes the tasks. Then we will see how Prefect handles multiple task dependency, error handling, and shadowing in the tasks. Then we will see how you can monitor the tasks using the Prefect dashboard. The Prefect dashboard gives you real-time insight about all the tasks that executed, whether those tasks were executed successfully, failed, or if there is any error.
By the end of this video tutorial, you will be having a solid understanding about Prefect. You will also be able to write Prefect tasks and execute workflows.
[Music]
Here are the topics that we are going to cover in this video tutorial. We'll start with an introduction of Prefect. Then we will see how you can define and run the tasks. Then we will see creating and managing the flow. How you can handle the errors. How you can schedule the multiple task flow. Then we'll see how you can monitor and manage the workflow using the Prefect dashboard. So, whether you are new to Prefect or you want to sharpen your Prefect skills, this video tutorial is for you. Let's get diving and start writing the Prefect.
Don't forget to like, subscribe, and hit the notification bell so that you don't miss any of our future video tutorials. Let's start coding now.
We are going to install Prefect on your system. Before installing Prefect on your system, I would recommend you to create a virtual environment. To create a virtual environment, you can just type `python -m venv` and then `venv`. So, `venv` is nothing but the virtual environment Python module that I'm using. And then you can give any name to your virtual environment. Here, I'm giving `env-test`. So, once you execute this command, you will be able to create the virtual environment. Then you have to activate the virtual environment. To activate the virtual environment, just type `source env-test/bin/activate`. If you execute this command, you will be inside the virtual environment. You can see that I am inside the virtual environment `env-test`.
Now, install Prefect on your system. To install Prefect on your system, type `pip install prefect`. You can see that Prefect is already installed on my system. Now I'm getting inside my test project directory where I'm writing the Prefect tasks. Now, open the virtual environment. And here I'm just going to write the Prefect task in `hello_prefect.py` file.
Now we are going to import `task` and `flow` from Prefect. For that, you are just typing `from prefect import task, flow`. Now we are just going to write a simple function, say `def say_hello():` and inside that, we are just going to print the message "Hello world". So, this is a simple function. If you want to convert this function to a Prefect task, you just have to add the decorator `@task`.
Now we are going to create another function, `def my_flow():`. Inside that `my_flow`, we are just going to call the function `say_hello()`. To convert this `my_flow` function to a Prefect flow, we just have to write the decorator `@flow`.
Now we're going to call the `my_flow()`. For that, we're just writing `if __name__ == "__main__": my_flow()`.
Now let's execute the program. So, the first of all, like I have to activate the virtual environment. So, for that, type `source env-test/bin/activate`. You can see that I'm inside the virtual environment `env-test`. Now let's execute the program: `python hello_prefect.py`. You can see that you are able to print the "Hello world" message. Along with the "Hello world" message, you can also see the info logs. This info logs is coming from the Prefect server. So, this is how you can write a simple task and the flow.
We can also add multiple tasks into the flow. Let's create another task. I'm just copying the same task and instead of `say_hello`, I will say `say_prefect`. And instead of this one, I'm just changing the message to "Welcome to Prefect".
Now we are going to add the task into the flow by calling the `say_prefect()` function. Let's execute the program. You can see the message "Hello world". You can also see the message "Welcome to Prefect". Both the tasks are executed as part of the flow. Here you can see the message that "All states are completed". So, this means that we are able to execute the Prefect flow successfully.
Now, this is a very simple program that we have written to execute the Prefect flow. This is how you can use Prefect in Python to manage and automate the workflow execution and data pipeline. This is all about the simple program that we have written to execute the Prefect task and the flow.
Now we're going to see a real-time example where we can use the Prefect flow and tasks. Let's consider you are using an ETL program where we have to extract the data from the website, then you have to transform that data, and store the data into the database. So, basically, it involves three tasks. The first task is like to extract the data from the website, then transform the data into the required format, then store or then load the data into the database.
Here is the program that we are going to execute to automate the ETL program using Prefect. So, first of all, we are going to import `flow` and `task` from Prefect. Then we are importing `requests`. We are using this `requests` module to extract the data from the website. We have created three tasks here. So, we are going to write three functions for the three tasks. First function or the task is `extract_data`. Another task is `transform_data`, and the third task is `load_data`.
Inside the `extract_data` task, we are going to get the data from this URL. Now let's see what kind of data is present to this URL. I'm just copying and opening this URL in the browser. You can see that this is the sample data it is present to this URL. Now we are going to read this data into our ETL. So, this is the program where we are using the `requests` module to read the data from this URL and then we are returning this data. Then we are converting this JSON data to a Python dictionary and we are transferring this data.
Now, there is another task in `transform_data`. Now we are going to transform the actual data to get the desired information. And then there is `load_data`. So, in `load_data`, we are supposed to save this data into the database. But instead of storing this data into the database, we are just going to print this data. These are the three tasks we have created as part of ETL.
Now we are going to write the ETL flow. Again, we are writing a simple Python function. Inside that, we are calling three tasks: `extract_data`, `transform_data`, and `load_data`.
Now we are going to execute the flow. To execute the flow, again, we are just calling the `etl_flow()`. Now let's execute the program. Just to make it more meaningful, I'm just renaming this Python file to `etl_prefect_flow.py`. Now let's execute this file. For that, `python etl_prefect_flow.py`.
So, you can see that the program is executed successfully. You can see that the process is completed. "All states are completed" here. Inside the load, we are just printing the message "Loading title". Here is the data that we received as part of ETL. This is the simple ETL flow that we have created using Prefect.
So, we can also add additional features here that are provided by Prefect. Suppose, like if you're extracting data from an external URL, when you extract the data from an external website, it is not always sure that you will get the data. So, that like if you don't get data, or if there is any error occurs while extracting the data, you want to retry that function execution again. So, for that, you can just add `retries=2`. What does it mean is like, if the function fails, right, we will retry it again. Inside the ETL flow, we can provide any name, just like `ETL Flow`.
Now we are going to see how you can start the Prefect server and open the Prefect dashboard. The command is very simple. You just have to type `prefect server start`. If you execute this command, you can see that the checkout dashboard at this URL. So, I'm just copying this URL. This is a localhost URL. You can open that in your browser. Suppose this is the URL that I open. As of now, there is no flow registered to this dashboard.
Now we are going to execute our ETL flow again. We will keep this Prefect dashboard running. Now, in another terminal, we will execute the ETL flow again using the command `python etl_prefect_flow.py`. If you execute this command, you can see that "All states are completed" and we also got the desired output here.
Now, if you go to the Prefect dashboard, get inside the flow, you can see that the flow is created at this time. This is the exact time when we ran the ETL flow. If you get inside the ETL flow, you you can see that there are three tasks, and all three tasks are completed successfully. If you get inside this detail, you can see that this task `extract_data`, then `transform_data`, and `load_data`. And we can also see the time of execution. So, suppose like to execute the `extract_data` task, it takes 1 second. `transform_data` again, it tooks 1 second. And again, you can see the one, one seconds. So, you can see that all these tasks are executed sequentially, one after another.
This is how you can start the Prefect server and create the Prefect dashboard where you can monitor all the tasks.
Now let's understand what's the use of Prefect, or what Prefect actually can do. Prefect is nothing but a task management system where we can break the workflow into smaller tasks, and each task does a specific part of the job, like downloading data, or processing the data, storing data into a database. So, another use of the Prefect server is to create the flow where we can combine multiple tasks to create one single flow. And along with that, it also allows us to add retries and error handling mechanisms to our tasks.
Another feature is scheduling. With scheduling, you can run automatically all the flows at a specific time or after intervals. The another feature is monitoring. Using the Prefect dashboard, you can monitor all the tasks in the flow, just to make sure all the tasks are executed successfully.
By now, you should have a solid understanding about the Prefect tool, how to use the Prefect tool to create tasks and flows, so that you can automate the workflow in Python.
Here are some more ideas to explore Prefect further. So, you can experiment with more complex workflows. You can explore even more advanced features provided by Prefect, just like mapping and result handling. You can also check out the Prefect documentation to get more insight about the Prefect tool.
Thank you so much for watching this video. I hope you find it very useful and you are really excited to use Prefect in your project. If you have any questions or need any further clarification, you can reach out to me in the comments, or you can also connect me. I'd love to hear how you are going to use Prefect in your project.
Don't forget to like this video, subscribe to our YouTube channel, and hit on the notification bell so that you don't miss any update from me. Your support really helps and motivates me to create more such valuable content. Thanks again for the watching and happy coding with Prefect. Bye-bye.