Transcription
Modern data platforms are built with a lot of different tools, which is great until you need to coordinate all of them to work together. Because a lot of times you want them to run in a particular order, but the problem is it's hard to do this without a centralized place to coordinate and monitor all these things together.
And so, in today's video, I want to share with you one tool in particular called Prefect, which is an open-source task orchestrator tool that I've actually used quite a bit on projects lately. And it's a great one to get started with because it's based on a common language, which is Python. So, in this video, we'll review one: what is Prefect and what makes it a little bit different? We'll walk through how to set up a very simple workflow. We'll talk about some of the built-in features that make it a little bit easier to organize everything. And then finally, I'll show you how you can run this in production using Prefect Cloud for free and then hosting it on your own servers.
All right, so first of all, what is Prefect and how are teams actually using it? So, as we mentioned, Prefect is what's known as a task orchestrator, which allows you to coordinate all the different components of your data stack from a central location. And like a lot of tools in the modern stack, Prefect is open source, and it's something you can start using effectively for free. They also have a cloud environment, which, as we'll see, is also something you can use for free to monitor and just get started with the basics. It's also built around a really common language, which is Python. So, if you know the basics of Python, you can use Prefect. It's also designed based on what they call their hybrid model, which is a way for them to separate your infrastructure in terms of where you're actually running things and where your code lives. So, your data itself never actually lives on Prefect servers. So, from a security perspective, that's a really nice feature. And then lastly, one thing that I really like is that it has a pretty sleek and modern-looking UI, which makes it really easy to use and I think approachable for a lot of people who might be a little intimidated by this concept otherwise.
And the other thing here is how are teams using this? At least in my experience, I've noticed obviously for coordinating, so building data pipelines and connecting all of your tools together and being able to monitor it all in one place. You can also create custom scripts for data extraction and loading. Maybe you have some sort of process that doesn't fit with the pre-built connectors and some tools, but instead, you want to create some custom Python logic to handle all that. You could do that in Prefect and coordinate it as well. And then lastly here is scheduling and alerting because now that you have everything kind of being coordinated from one place, you can run it on a schedule and you can get alerted if things go well or things fail and address it right away. So, it's nice to have that all from one location.
Okay, so now that we understand what Prefect is and how teams are using it, let's now get into the actual code itself and I'll show you what Prefect looks like and how you can get started using it with a simple workflow.
Now, to get started with Prefect, step number one is go to the website, prefect.io. And what we'll do is deploy open source. I want to run this locally on my machine. And from here, we're going to go right to install Prefect. And as I mentioned, this is built around Python, and right now, at the time of this recording, it requires Python 3.7 or later. So, just make sure that you have that. And like a lot of other Python projects, the recommendation is you install it in a virtual environment. And what we're going to use is venv. And obviously, there's other options you can do here, but that's something that you're going to want to do. And to install it, all we need to do is run a `pip install prefect`. And this is all we need to run to get this on our machine, which is super easy.
I'm going to open up Visual Studio Code, and I have a blank Prefect directory here that I'm going to use for this demo. And I've opened up a new terminal here in the Prefect directory. And so, this would be wherever you want to build your project. Hang on, the way you're going to create your virtual environment will determine this code, but for me, I will run `python3 -m venv` and I'm going to create my virtual directory as `venv`. And again, this is creating a virtual environment within this directory. And we should see it up here. And here it is. To activate a virtual environment, if you're on a Mac, it's `source venv/bin/activate`. Now we can see this is here. Understanding virtual environments is outside the scope of this video, but this is something that you should consider doing for this.
So, now that we have our virtual environment ready, let's go ahead and `pip install prefect` so that we can install Prefect into this virtual environment. And it should take a few seconds here. I've done this in the past, so it might be a little faster. So, this looks good. And just to confirm that we're good, we'll run `prefect version`. And so, here it is. We can see we installed Prefect and everything looks good.
So, at this point, technically Prefect is ready to start using. But just to give you something interesting to look at here, and let me, I'm going to run a command `prefect server start` to show you what you get locally out of the box. But like a lot of other tools, Prefect comes with this built-in UI. And here you can see right away, it spins up a nice UI that we're ready to use and monitor everything that we would build. Now, obviously, we don't have anything here yet, but out of the box, this is designed in here for you to use. So, this is an example of what I mean by that nice UI that's easy to use.
All right, so at this point, we've installed Prefect. We can see we have a server locally that we can start to work with. Let's now create a simple Python script. Nothing crazy, just to see this in action and show you the different components.
Alright, so back in Visual Studio Code here, I've opened up a new terminal, and we can start to build our first flow, as it's called, really a Python script. And what I'm going to do is within my Prefect folder, just create a simple file called `hello.py`. Now, I realize this is going to be a very simple example, but the idea here is don't get caught up in the actual logic that I'm building, more so pay attention to the fact that this is a Python script. We're going to use things to implement Prefect. You can make this as simple as a hello world script to as complicated as running extract and load functions and all that. But with all that said, I'm just going to do a hello world function here, nothing else involved, just to prove that this is a working Python script. So, if I were to come down here now and do `python3 hello.py` just to run this script, we can see it printed it out and it works.
Now, at this point is where we're going to need to implement Prefect stuff, which is those decorators that I mentioned earlier. And if we go back to the documentation here, there are three concepts that I want to just point out that you need to understand. One are flows, which is the most basic Prefect object. Essentially, you're just going to wrap your code in a flow using an `@flow` decorator, which I'll show you in a second. And you can think of them like functions. They take inputs, do work, all this stuff. And then within flows, you can have the idea of a task, which is a function that represents a discrete unit of work within a workflow. You're not required to use these, but you can think of them like breaking down your flows a little bit more. So, if you look at this example here, we have, this is a flow which is calling a task. And the reason this is helpful is because it kind of helps you modularize your code, break it down a little bit further so that you can manage it and monitor it in certain ways.
And the other thing I'll mention here is that you can also have the idea of a subflow, which is really just a child of another flow. You're basically calling flows within flows. And a little bit later, we'll talk about why you would use that versus a task, because they're kind of similar. But high-level, these are the concepts that are important. And what we're going to start with here is the idea of this flow decorator.
Back here in our project, let's now turn this into a flow so it's something we can monitor on the UI. Because right now, it's just, you know, sitting here. There's two things we want to do. One is import from Prefect, the idea of a flow, so that we can use it. So, we're importing `from prefect import flow`. And as I said, the decorator is what we need to add. So, if we do `@flow`, this is all we really need to do. So, now if we run this exact same script, so imagine you had something that existed already and you run this, it should pick up that it's Prefect. And you can see the engine is running, it ran the script. Also, if we go to our server, we should be able to see that it picked up this run. And here it is. So, we see "hello world," that was the name of the script, and it gave it a random name. It just comes up with names of an instance of that script. So, we can see it ran here, and there's some other information here that we'll, we'll talk about later. But here we can see there's a flow, and here you can see there's task runs. We would have it here, subflows, you can have parameters, all sorts of stuff. We can already see it's here as an example of something we ran locally.
So, now let's actually see that task in action. And for sake of time, I'm just going to copy and paste this in here. Here we can see now, instead of just writing the message here, we're going to create a new function called `create_message` which returns a message. And then we're calling it within the flow. So, by adding it within your flow, it understands that it needs to call this. And if we wrap it in a task, it will inherently understand that this is a task below this flow. But we need to also include this here so that it can bring it in. So, let's try this again and see what happens. These are different components that you can build out and nest together to create your workflow.
So, now that we have the task, let's go back here and see what this looks like again. Now we can see there's a new version here, a new instance. And here we can see this is now logging the `create_message` task that it just called it as, because that was the name of the function. And here it is. So, here's the task. There's no inputs, there's nothing here, but you can see now this is what that would look like. But more importantly, within flows, it's still nested under one flow. It just now has a subtask.
Now, lastly, let's check out the subflow idea. Let's see what the difference is. So, again, I'm just going to copy and paste for time's sake here. And what I've done here is I've added another function. So, all these are just defining functions, just like simple Python, but adding a `@flow` decorator instead of a `@task`. And it's just printing "result" again, nothing crazy here. But let's now save and run this and see what it looks like in the UI, because this is going to have a subtle difference here that I think is important to point out. We can see it added it as a separate flow, whereas before, in the task, it was just a task. So, now we have two of these concepts. But if we look into "hello world," we can see not only do we have the task run, but we have the subflow run.
And so, the main reason you would want to pick a subflow over a task is for the UI purposes, for the coordination, and all this stuff. You can create a deployment for a flow, which gives you a lot more flexibility in terms of infrastructure and where the code lives and all this stuff. So, think of flows as kind of bigger concepts that you want to string together, whereas a task is just a smaller unit of work within a particular flow.
All right, so at this point, we have our local script, and we've been able to trigger it from our local machine. And everything that we see in our server is based on just that. But what about when you want to schedule this or have people remotely work with the same code who aren't on your machine? That's where the idea of deployments come into play, which are API representations of flows, which we'll explain what that means in a second, and allow for remote configuration and scheduling.
Here's documentation on deployments. And an important topic here is indicating an entry point flow. So, remember we made those flows and tasks, so it's asking where does it need to start? What's the entry point flow? You can think of a deployment as a configuration for managing your flows. Ultimately, it's a YAML file that's packaging together your requirements for that deployment. Again, we're going to see a little bit more about what this means. And this is what we have mentioned before is that hybrid model that they have. So, first, we have our flow code, which is the code we've been writing in Visual Studio Code. We're going to create a deployment definition, which, as it said before, is a configuration file. This is going to be a YAML file that we'll build here shortly. This file gets stored on the Prefect API server as a deployment. So, you create the definition and then you push that definition to this server. Right now, the server for us is our localhost, but you could also use Prefect Cloud for this same purpose.
Within a deployment, there's a lot of configurations, but two of the most important ones are storage and infrastructure. With storage, you're saying where does this code file exist? Is it, does it just exist on that local machine, or is this in a cloud server? Is it in an S3 bucket, a GitHub repository? Where should it look for the code? And number two is infrastructure, which is how is this going to run? So, for example, again, is it just on your local machine, or are you running this in a Docker container, in a virtual machine? You can indicate this here. And the agent is a small polling service that's going to be constantly checking in with the API to see if there's work for it to do. And just to reiterate here, there's always a boundary between your code, your private infrastructure, and the Prefect backend. So, again, that could either be your local server or Prefect Cloud. You only register the deployment metadata on the backend, allowing for a clean separation of concern. So, this is really important for security purposes. It's just the configuration indicating where and how to run it.
So, what with that said, let's now go back into our project and build a deployment to see how this works.
Alright, so to create your deployment, there's two steps. First is to build, and then the next is apply it. And when you apply it, then it's actually going to be on your server. So, let's go through the steps here. We're going to run this `prefect deployment build` command. And the first thing we'll put in here is that entry point. Remember, it needs to understand where should it go to run this flow. And the entry point for our sake is the `hello.py` file. So, this file, and we want it to be this function, so that flow. So, we do `hello.py:hello_world`. So, again, we're saying this file, this flow, is the entry point. Now, I'm going to do `-n` for name to give this deployment a name. I'll call it `demo-deployment`. Now, there are many other options we could put here. So, for example, infrastructure or storage, but for now, we're just going to keep it simple.
Let's go ahead and create this and see what it says. Alright, so it worked. But there's actually one subtle thing we need to adjust, which is to add this condition right here, which is pretty standard for most Python, but you should remember to add this here. Okay, that works. And we can see here it created this YAML script. And this is what I meant by creating that configuration file. And within here, there's simply metadata information about what this deployment represents. We have a work queue, which is saying when we run this, what queue should the flow go into? And you can create all different types of ones. And this is where that agent concept is going to be looking. So, in this case, anytime we trigger this deployment, it's going to get added to the default work queue. But of course, you could override that. But this is the, the default setting. We don't have any tags, we don't, we don't have a schedule. The infrastructure we have is the default process, which is saying a local process on your machine. But this again could be somewhere on the cloud. You can indicate here where you want that to be. Down here, we see our storage, null is the default, and it's saying it's again localhost. So, whenever you trigger this, it's going to look locally on the same machine as you're running it for this path. And then the entry point, which again is saying, okay, look for this file and this script and run it. So, this is how this all goes together.
But at this point, we also still don't have this on our server. It's just created the deployment file. What we want to do is now move this to the server so that other people can use it and trigger it however they want. And the command to run that is `prefect deployment apply` and the name of the deployment file, which for us is `hello_world_deployment.yaml`. Okay, so now let's refresh this here. And here we can see now a deployment exists. And if we go in here, we can see it has more information for us. So, again, talked about that work queue. It just gives you an anonymous infrastructure for your local, but this would depend on whatever you set it to. So, again, this deployment concept is really important because now anybody can trigger this. You can run it on a schedule. You can version control your deployment YAML files, and you can start to set remote infrastructure and storage to run your flows rather than just triggering it manually here.
And now what we can do is do a quick run. But there's actually something that's going to happen here that we'll have to explain. But let's just see. And you click run, view run. And so, what happened now when we click that is this flow run, and it was given an instance name, and it's sitting in a queue. It's sitting in that default work queue. And we need something to tell our infrastructure that there is something in this work queue waiting to be run. And this is where the idea of an agent comes into play. And we can see it's even mentioning this here as well. So, let's now talk about the agent concept.
So, at this point, we have our flow code. We've created the deployment definition that's pushed to the API server. It knows the storage is local, the infrastructure is our local machine. But our infrastructure will never know when there is a flow ready for us to run. And that's where this agent comes into play, which again, is that polling service that's constantly looking for work to be done. And that's why this is currently in "late" and waiting. So, what we can do here is we can say `prefect agent start` and then the work queue is `default`. And we're saying `default` because that's the name of this here. But if you had a custom one, you would, you would change that. So, let's run this command here. It's just a lightweight service. It's going to be constantly looking at that work queue. And here you can see it picked up right away that there's one there and it's going to start running it. Here we can see this now changed to "pending." And, and that's because this look to the API on our local server saw there was a flow waiting in that queue, picked it up, and ran it locally. And what you're going to need to do is whenever you set up your infrastructure, have an agent there also being the intermediary to understand when there's actually work to be run. And the API is just passing along metadata. And once this triggers, there's nothing else going to the Prefect server. It's running on your preferred infrastructure and can connect with your storage and everything else completely separate. So, that's nice from a security perspective. And we have the ability to view this and monitor it all from the UI.
The other item here to point out is work pools. You might have multiple agents running on a particular infrastructure depending on the scope. But having a pool allows you to manage them together and set limits, maybe on concurrency or other things, and just group things together to make it easier to manage.
Okay, so up until this point, everything we've done has been on our local server. What I want to do at this point is show you how you can shift this backend portion, so essentially the UI and where the backend API is going to be hosted, from your local machine to Prefect Cloud. And there's a few reasons why you might want to choose Prefect Cloud over just hosting this locally. One is maybe you don't want to host your own server. You can see there are a few steps involved, and there's just some general overhead with that. Two, you might want the built-in authentication, which helps with different permissions, working with service accounts, and keeping everything organized from a security perspective on the cloud. And third, cloud options tend to make things a little bit easier in terms of the workflow of things for your team, so making things like workspaces, sharing it throughout your organization, onboarding, using the automations and connectivity, everything that's built into it is a little bit easier with the cloud compared to the CLI. And it's actually really quick and easy to do this.
So, what we'll do is we'll come back to prefect.io. And from here, all you need to do is add your email address, and it's going to send you a link to sign in. And just like that, I'm in Prefect Cloud. What we need to do is create a workspace. It's going to default to your owner name, and we'll give it a name. I'll call this `prefect-demo`. So, we can see it's a similar view as to what we had locally. But what we need to do is make sure that our local terminal connects to this as our API and our backend, as opposed to what we have right here. Because right now, all of our flows and everything is going to run here. We want it to instead point to our cloud service right here. Fortunately for us, Prefect makes it really easy to do this. And all you have to do is run that one command in our editor.
So, back here in Visual Studio Code, I'll type `prefect cloud login`. And immediately, it's going to give us two options. It's going to say, "How would you like to authenticate?" Log in with a web browser or paste an API key. But considering we don't have an API key yet, this is our first time, we can just log in with a web browser. So, I'll select enter. And it immediately brings us to this authentication where we can go and authorize. Let's go ahead and close the window. And here we can see we are authorized using our workspace. And if we go into our settings under API Keys, we can see it automatically added this for us. So, we are ready to go and start using Prefect with the cloud as our backend.
And just to show you that this is working, if I just run this again here, so `python3 hello.py`. Previously, it was showing up on our local server. But if I run this now, this information should show up in our Prefect Cloud account. So, if I refresh here, now we can see these both ran here. We have the flows, it picked all this up, and it's showing in the cloud now because of that connection. But if we go to deployments, we see that our deployment script is in here. And again, this comes back to this concept here of how everything is stored. This deployment definition was on our local backend. We need to put the definition to our Cloud API backend. And then everything else would be the same. So, that's how you can think about these concepts being separate.
So, I'm going to rerun this apply command to apply it. But because of the fact that we are authenticated in the cloud, it should apply that to our Cloud environment. And as we can see here, we can see it in the UI directly. So, I just open this up, it should take us right here. And everything else is the same. So, we have our "hello world," default pool, infrastructure. It's saying, okay, it's on our local one. It's just an anonymous name. And again, it's that same situation where if we try to run it, there's no agent currently available to pull work from that queue. So, let's go ahead and start it here, even though we're locally. Again, this is just to show you those separation of concepts now. So, Prefect is running the agent here. It picked it up right away, and it's going to start running it. Here it is. It ran successfully from the cloud on our local machine, just through all of those connections working together.
The next concept we need to talk about are blocks. And blocks are ways to securely store credentials and configurations. And it's nice because you can manage them all from a centralized UI, which is easy to work with. Let's go ahead and add a block. And as you'll see right away, there are 79, at least at the time of this recording, out-of-the-box blocks that you can use. And there's common things like adding a secret, Slack credentials, GitHub, GitLab, all sorts of stuff. And if there's something that's not here, you can even open up this catalog here, and there's way more other things that you can add into. So, for example, with DBT or Fivetran, you know, there are specific things that you can add to your project and essentially import custom blocks that are built for these things. So, all of Prefect here is designed with really a lot of data tools in mind, and this integration between all of them to make it easy for you to manage your workflow and your credentials, all this stuff from a central location.
So, let me just show you an example of how this might work. We're going to do something really simple here, and I'm just going to add a string just to show you the workflow of adding a block. So, for example, let's say we want to add a block, and I'll call this "Future NBA Champs" because this is my favorite sport, and why not. And the value here, hopefully sometime in my life, will be the 76ers. So, let's assume we create this block. And here is exactly how you would implement it in a script or a workflow file. So, let's see how we would use this. Let's copy this. I'm going to paste it into our file here. And what I'll do is the message will be instead of this here, we'll say `string_block` and I actually need to do `.value` because it's the value of what's being loaded. So, just a recap here, what it's doing is it's bringing in a block component, it's importing the string one specifically, and loading it from your backend, which is in our case now the Prefect Cloud. But otherwise, it could have been your local server, wherever you're kind of hosting that backing component.
So, let's run `python3 hello.py`. Nothing else has changed other than we added this. Let's see what the message says. "76ers 10." So, that means it grabbed it from that UI. Now, the other thing we can do here is we can edit this here. So, let's say we want to edit the value to be "Warriors." Now, we edit it from that central location. And if we run again, nothing changing here. Now we see "Warriors" was added. So, again, this is stored here on our server, in our case the cloud right now, and we can adjust and do whatever we need to. For those of you out there who might use DBT, there's a lot of DBT built-in blocks here. There's also things for Snowflake, for BigQuery, Redshift, all sorts of tools that you can easily manage your credentials. And they're set up to be easy. And I'll point out here that you can not only add these through the UI, but there are ways to do this through code as well. So, if you want, you can write a Python script to also upload blocks and have that in code and version controlled for future folks as well, if that's something you want to do.
Now, to bring everything home here, what I'm going to do is version control our code here. So, I'll host it on GitHub. Hey, we're going to create a block to use GitHub as storage so that our flows will run using the GitHub values as opposed to what we have here. And then we're going to see this all play out in action. Now, for the sake of time, I've already gone through and created a brand new repo and I just pushed up the changes here to it. And the only thing I added was a `.gitignore` so that we don't have these files added to version control.
Alright, so with this here now, what we can do is add this as a block. So, let's do that. Look for GitHub. We'll add this block. And here is where we can give it a name. So, I'll call this `prefect-repo`. Here we need the repository in HTTPS or SSH format. So, you can find that here. And depending on your situation, you might also need an access token. So, I already have one. But what you could do is you go to settings of your whole account, go down to developer settings, personal access tokens. And here is where you would create them. But for me, I will paste in what I have here and create. So, now we have this repo here that's available for us to use for whatever we want.
What we can do is alter our deployment to include that as storage because if we look here now, our storage is null. But we want to set it to that GitHub so it's always using this. How would we do that? Over here, I have the docs open for storage. And in here, we can see when we create a deployment, what we want to do is add a `--storage-block` flag here. And we're going to need to put `--storage-block` and then for us, it's going to be `GitHub` and if we look, this identifier here, we can see the slug is `GitHub`, and we would put `GitHub:prefect-repo` (the name of our block). So, let's put this all together now. I've pasted in here our original deployment command, but now what we're going to do is do `-sb` for storage block, and then our type is `GitHub/prefect-repo` right here. And let's build that. See if it works.
Alright, so here we can see in real time it updated it. The storage is this repository using the token that we have up there that's hidden. And everything else stays the same. So, once we deploy this and move it up to Prefect Cloud, anytime somebody tries to run from Prefect, it's going to look here for the code as opposed to my local machine here. So, let's go ahead and apply this. And now if we look back here to our deployments, and I call this one `test-deployment` as opposed to `demo`, but if we look at `test` here, we can see storage is `prefect-repo`. And if we click it, it'll take us to that block. So, it all works together. And you could take the same concept and apply it to infrastructure. So, rather than having this as a local process, instead, you know, you could create one for any of these run infrastructures, so a task, a Docker container, Cloud Run, any of these things, you could build them and use them as infrastructure and then apply that to a deployment. And then anytime you run a deployment here, it will use whatever storage and whatever infrastructure you have as part of that deployment.
And now, just a few final things to round out here. You do have the option for automations. So, you can do all sorts of things like, depending on the state, let's say something is in, you know, enters running state, you can do all sorts of triggers. You can send a notification. You could integrate it with a block, for example, your Slack block or Teams or something, and get yourself notified here. For concurrency, you can set a limit on the number of tasks running simultaneously with a given tag.
So, hopefully now you have a better understanding of what Prefect's all about and how you can implement it in your own workflow. So, thanks as always for watching, and I'll see you next week.