Name: Observing Airflow DAGs in Prefect
Uploaded: 2025-03-05T20:16:04.143Z
Duration: 31 min 20 s
Description: Observing Airflow DAGs in Prefect

Transcript for "Observing Airflow DAGs in Prefect": Hello, everyone. Good morning, good afternoon, or good evening, wherever you happen to be in the world. Welcome to today's Prefect webinar, observing airflow DAGs in Prefect. I'm Brendan O'Leary, and I'm thrilled to have you all here as we explore how Prefect can help you gain better visibility, control, and flexibility over your existing airflow setup even as you might be considering a full migration path to a modern orchestration platform. Before we dive in, let's quickly outline how today's session is gonna flow. First, we're gonna discuss why observability is critical in Airflow and with with workflow orchestration and the challenges that a lot of teams face. Next, we'll introduce how Prefect observability tools seamlessly integrate with your airflow DAGs, giving you real time insights, alerts, and error handling. And then my colleague Steven's gonna join us to walk you through the technical implementation with a deep dive into the actual code. By the end of this session, you'll have clear, actual steps to start improving your airflow workflows all without disrupting your current setup. So let's get started. I'm coming today from a pretty stormy Annapolis, Maryland. It doesn't look like this right now. It's actually a little little windy outside. Hopefully, my Internet doesn't go out, on the East Coast Of The United States. And so we'd also love to hear where you're dialing in from. If you're out there, drop in the chat and we'd love to see who might be like the closest to and the furthest away from my desk right here in Maryland. If you can win one or the other of those, I'll give you a couple internet, couple hundred Internet points, let's say. And then hopefully today, everyone will get some actual next steps when it comes to your journey with data orchestration. If you've been struggling with limited observability and airflow or the the desire to, like, bolt on another tool just just for observability, wrestling with the rigid structure of airflow, or wondering how you can maybe gradually transition your team to a more modern orchestration framework, you're definitely in the right place. We recognize that a lot of teams have legacy airflow DAGs that are, you know, critical to their operations, and not all of them can easily be converted overnight. Some workflows could be highly complex, deeply integrated into your existing infrastructure, or simply too risky to modify all at once. That's why we really believe in meeting teams where they are, providing them with the tools to enhance their current workflows and existing workflows without requiring a, like, rip and replace migration all at once. Our goal is to help you modernize your orchestration incrementally so that you can enjoy the benefits of Prefect while keeping your mission critical pipelines running smoothly. Okay. But I've used that word orchestration a lot. And honestly, it's probably one of the most overloaded terms in data engineering and data science. Or maybe it's a close second to like all the various phases of water like lakes and snow and ice and all that. And historically, orchestration has almost been synonymous with, you know, scheduling batch jobs. So much so that, like, other orchestration providers offer add on products for things like observability. But we believe very strongly that that's the wrong approach. At Prefect, we see that, like, observability, error handling, a clear separation of concerns, those are table stakes for what modern orchestration and data workflows require. Engineers should be able to focus on writing their meaningful business logic that drives real world impact, not a lot of boilerplate code that's just to make the orchestration framework work. That's why Prefect provides built in observability and built in flexible orchestration tools that work the way engineers need them to right out of the box. And so let's talk about why we're here. If you're using Apache Airflow today, you already know it's a very powerful tool for orchestrating workflows. But as your data pipelines grow, so do the challenges. So there's a few common challenges we hear from teams that are running Airflow today. Limited visibility across all of their DAGs. Right? Airflow's UI shows you an individual DAG run and and a list of all the DAG runs, but lacks a real clear centralized way to track the state of your entire data platform. And debugging errors often means, like, going somewhere else to dive into the logs manually. And And then when an error handling that does exist is very rigid. If a task fails, the retry logic is often not enough. You you really need a more flexible way to handle dynamic failures, either with source or or destination data. And speaking of moving data around, X COMS is just a limited way of passing between tasks. It's very cumbersome and many teams rely on external storage solutions, you know, serializing and deserializing data multiple times throughout their pipelines for no reason, just adding a whole bunch of complexity to the workflows. And then lastly, scaling issues. Right? Workflows become more complex. Teams, you know, more and more teams are relying on it. They hit this performance bottleneck with, you know, Airflow scheduler or salary executors, those things that try to scale out Airflow. And if any of this sounds familiar to you, the good news is you don't need to rewrite everything overnight. Instead, you can enhance Airflow with Prefect's observability tools to give you better logging, alerting, and debugging capabilities right from day one without touching any of your existing workflows. So what makes Prefect the right choice for teams that are using Airflow today? Well, Prefect isn't just a tool for monitoring Airflow. It's it's a modern orchestration solution and provides observability as a natural part of that workflow management process. Rather than treating observability as an add on, it's a core feature for Prefect, allowing teams to see their existing DAGs and planning migrations towards more flexible workflows. With Prefect, you can get the best of both worlds. Visibility into your existing Airflow DAGs while laying a foundation for a more modern approach. Prefect provides a centralized view of your workflows, helping teams to track failures, retries, performance, all in real time. And because you can run it alongside Airflow, you don't have to rearchitect it all at once. And observability really then is just the first step to transitioning to a more dynamic, Pythonic orchestration system that better supports the ever evolving needs of your team. Unlike traditional, you know, monitoring solutions that you might bolt on and treat workflow execution as a black box, Prefect gives you this real time task level insights into what's happening with your pipelines. Okay. That's a lot of like information and me like talking at you. But let's look at some code. So next, Steven, who's on our developer relations team, has put together a fantastic demo of some of these concepts that I've been talking about, and really showing how you could integrate this observability features of these observability features of Prefect into your Airflow today. And by the end of this demo you'll have the knowledge and we'll send you the code that you need to add observability to your airflow workflows right away. So with that, I'd like to hand it over to Steven to show us the details. Take it away, Steven. Hello. Hello. This is Steven from Prefect. In this video, I'm gonna show you how to add Prefect observability to your Airflow runs. So why would you wanna do that? Well, I'm glad you asked. If you're running Airflow but struggling with limited observability, switching entirely to a new tool might sound daunting. But what if you could integrate prefix monitoring without changing how your DAGs run? Well, to quote Hannah Montana, you really can't have It's the best of both worlds. Excuse me. I I don't know what got into me there. Let's take a quick look at the airflow DAG that we're gonna monitor inside of Prefect. Again, I work at Prefect. I'm not an airflow expert. So this is this DAG is not doing anything spectacular. I'm assuming that your airflow DAGs are doing some are are doing more spectacular things, but this process is gonna work whether it's the simplest the airflow DAG or the most complicated one. So I just built something simple here. So I just wanna run through it really quickly so we know that we're we're what we're testing here. So we have our imports up here on the top, our default arguments, everything that you would expect to see there. We have three functions that we're running through this DANG. So we have the first one that's task success that, guess what, its job is to succeed. The second one is task random fail. So on this it's so 75% of the time it's gonna fail. 25% of the time it's just gonna succeed and go to step three. We built it out that way so that, we can get more failures than success. You'll see why here in just a few minutes. Our third task, if it gets there, is just a final task where it returns done, so it's just gonna succeed as well. So now we're gonna create our DAG. So we're gonna name it test workflow, with our default arguments in there, throw a description, a schedule, and we're just gonna put false and catch up there, and then define our DAG here for the task one, task two, task three. So perfect there. We're gonna set our dependencies for task one, task two, task three, and then we're just gonna down run the test down there on the bottom. Our goal here is simple. We want Airflow to keep executing your DAGs, but we'll send the status as the Prefect. This way you can monitor your old Airflow DAGs running as they are now and build new processes inside of Prefect. This will allow you to have observability into your new and old workflows all in the same place. Now Now that we know what we're aiming for, let's jump into the code and see how it works. Looking at our code here, so starting off at the top, we have all of our import statements like all good Python scripts do. I'll skip over that. So jumping down into this first class that we defined, so we we have an airflow monitor error exception. So this is a custom exception, just for the airflow monitoring to differentiate between, like, airflow specific errors and then other exceptions. I ran into a few, like, deprecation error things that were going on with airflow. So I built that out to kinda get past that, and so we know what's going on there. So jumping down into our first task that we have, which is the check the airflow setup. So this is just to verify the airflow is properly configured and running on your computer. So this is just making sure that things are happening. And so, basically, we're gonna check, we're gonna get the, the airflow home. We're gonna print that out. So we'll we'll see that inside of Prefect to make sure that we know where it's running, which is, I've learned while using airflow this week, pretty important. And then, of course, we're gonna run our subprocess of airflow drops check, to you know, and return that, return that health check if we get something back there. And if not, if there are exception with the health check failed with that string. So nothing too complicated there. We just wanna make sure that airflow is running because guess what? It's not running. We probably can't check our statuses here. So that's our first task. Our second task is to get DAG statuses. So So that's gonna get the detailed status of a specific DAG and its recent runs. And so we can see our we have our DAG ID as an argument, and then it's gonna return a dictionary that has the current state, recent runs, the last run time, and run ID. And then it raises this airflow monitor error if there's any issue getting, the DAG the DAG statuses. So that's kind of explaining what's going on there kind of in in, you know, just just words. So now let's look at the code and see how we're gonna do this. So we have our logger set up, obviously. So we're gonna send our command, which is the airflow DAGs list runs. So that's gonna grab those it's gonna grab those DAGs and return them in a JSON format. We're gonna throw those in the logger as well. So we're going to run that commands. And, if we don't find any, so if there's no DAGs that are returned, which would not be great, we're gonna we're just gonna put in our logger warning that there's no DAG runs found, and then return that current state of no run, so on and so forth there. But if there is DAGs, we wanna parse the JSON output so we can take a look and see what's going on there. So we're gonna load the JSONs in there. We're gonna we're gonna split it so we only get the we only get the last five. Of course, this can be changed based on whatever you want it to be, last 10, last 20. If you don't want to split anything at all, that's fine as well. So that's what's gonna happen there. So we're gonna say we found the the last five recent runs. And so if we get recent runs, we're gonna look at them, grab the current state, recent runs, last run time, the run ID. If not, we're gonna say there's no runs. Then we're gonna have our exceptions down here. So, we have our failed to execute airflow command, failed to parse the airflow outputs, and then we also just have our, oh, something broke, exception there as well. So, that is what we're gonna do to, get our DAC status. So now we're going to analyze the DAC health. So once we get the status, we gotta analyze to make sure it's doing what we want it to do there. So for analyze DAC health, so we're gonna analyze the DAC health by checking the recent run history, and we're gonna raise an error if three or more of the last five runs failed. So that is just an arbitrary number that I chose. Maybe you want one of the one or more, two or more, four or more, or maybe you just wanted it for all five. You can obviously change that in the code. We'll talk about that in just a moment. We have our arguments, which is the status info, which is the status information from the last function. It's gonna return a dictionary that has the issues, so the list of problems, the severity, it depend showing how bad it is, and then recent failures. So the count of recent failures there, and we're gonna raise our raise our error if the threshold is exceeded. So set up our logger again. Again, we're gonna go through. If nothing's there, we're gonna say that. We're gonna ignore that for right now because we definitely have stuff there. And then we're gonna find our recent runs, loop through those that failed. We're gonna count the ones that failed, and then we're gonna count the total. So that is the five and then the one, two, three, four, or, I guess, zero one, two, three, four, or five. And then we're gonna, you know, kinda print out some re some readability stuff in the in the code. So we're gonna analyze the recent runs. So if it's failed, we're gonna throw an x. If it's if it passed, we're gonna throw a check mark, because who doesn't love emojis? And then the next the next little bit is the really important part here. So we're gonna check to see if we we exceeded that failure threshold of three or more failures. So if the count is more than three, we're gonna throw a critical alert of the dag failure threshold exceeded. We're gonna say that, you know, the three, four, or five of the last five have failed. Then we're gonna show those failed runs here. And so it's gonna show those there. So the run ID with the start and end date so you can know exactly which runs failed. And then we throw that error of this, you know, the you know, if if that if that's not the error, we're gonna throw that error of the fail count out of the last runs have failed, then raise that exception. So, you know, we're gonna the error the Prefect the Prefect flow is going to fail if three or more has failed. So I wanted to make sure that if you're looking inside a Prefect that this monitoring if that three or more failed, it's actually gonna fail the Prefect run. So you're gonna know for a fact looking at, oh, that's red. Something bad's happened there. Then you could take a look and see there. So we're gonna return the issues, the severity, and the recent failures there. So that's the code that's gonna analyze that DAG count. So now we just gotta throw it all into a flow and make this thing run. So our our flow is here, with the name of airflow monitor. We're defining that, defining that we're defining the monitor airflow DAG, and telling us what it's gonna do here. So, basically, all the stuff that we set it out to do earlier. So running through here, set up our logger, verify that it's running, so that's using check airflow setup. Of course, raising that exception if it's not. We're gonna get the current DAG status using get DAG status, and then we're gonna analyze the health and check the failure threshold using analyze DAG health. We're gonna return the status and the health analysis. That's our full flow. That is our full workflow that we can use to monitor, airflow your airflow DAGs inside a prefect. Now we just gotta call that thing to run. And we're gonna take a look and see what it looks like inside the terminal here whenever we run it, and then we're gonna look and see what it looks like inside the prefect UI. So let's check that out. Alright. So running our script here, it's gonna it's gonna grab those last five, airflow DAGs and and get some results for us. I'm gonna run it. You're gonna see a bunch of stuff pop up, and see that the flow was started, finished the state completed. And then you can see that we got the airflow monitor error. The four out of the last five diagrams have have failed. So it's important to note, but we also wanna check and make sure is that actually true. So let's let's pull up our airflow UI here, and so you can see here, it looks like something else has ran since then. So let me refresh this real quick. Yes. There we go. So you can see that four of our last five, so the the the fifth one the fifth one passed, and the last four failed afterwards. So that's perfect. That's exactly what we should get there for our for our monitoring. And then you can see here going back into air or go back in the Prefect, excuse me. If I refresh this, you can see Vivid Cheetah is the one that just ran, and you can see that we even have an error popping up for you failed due to an airflow monitor error indicating that four of the last five diagrams have failed. Further investigations need to address the root cause. So that's giving you the monitoring that you need, but we can even click into it. We can get even more. So when I click into it, and if I open everything up here, you're gonna see that here's some of the things that we printed out throughout the code. I just wanna show this off inside the UI. So you can see there's our current airflow home. So there's the actual that's where that's where the working directory is. In fact, I forget how computers works somehow. But, yeah, that's where the working directory is. If I open up DAG status, you can see the the command that we're executing. We found the five recent runs, so it's showing off that information there. And then our DAG health, so you can see, like, there's our last five runs. So we got a fail, fail, fail, pass, with the times there as well. So you can see, what that's 191910, 19 15, 20. So it's running every five minutes. That makes sense. There's our critical error, that the DAG failure threshold exceeded, and then it shows us our failed run. So it gives us the run ID, gives us a start and end time there. So giving us all that information that we told it to give us, which I know, surprise. The code's working like it's supposed to, which is great. So we're getting all that visibility. So now what I wanna show off is what happens if we actually get less like, if we get, like, only two failures or one failure. So let's go take a look at that real quick. So there's a magic of technology. We cut back in here to airflow. So I've reran this. We've actually you can see now that we have four four past runs or four successful runs, with one failure here in the last five. So now we should be able to rerun our prefect code, and and it should tell us that. So rerun that. Let's see what it gives us here. Perfect running through. So you can see, ran through all of this, finished and state completed. There's our last four runs that are that are successful, the one failure the one failure out of the last out of the last five. We get all the information there. We can view that at that URL, but I already have it pulled up here. So you'd go like to prefect cloud, go back to our monitor, go to our chinchilla flow around here. Again, same information. There's our airflow home. There is our executing or there's our command we executed and that we found the last five runs. And then for our health, there's an information that was displayed for us in the terminal, with the flow with with the flow steady completed here. So this has this is went through went the airflow and grabbed those last five DAGs. This is what it looks like if it's successful with a complete you know, when it gives us that green check mark. So when we're looking at the logs here, you can see, like, hey, like, uh-oh. This failed. Which then at this point inside of Prefect, it gives us the opportunity that if this thing failed and then whenever we have it set up, whether it's last through, you know, through the last five, four of the last five, two of the last five, however we have it set up. This allows us to, like, then take the next step and say, like, hey. If we get this failure, send me an email. Send me a Slack message. Like, send me a Teams message. You know, it gives me the opportunity to, like, really take action on this thing and make something happen, versus, you know, there's nothing worse than waiting until, you know, a dashboard is broken or you give a presentation and there's data wrong. Something worse than waiting until that point to see a failure. We're we're letting you see that DAG failure for airflow inside of Prefect here, so that whenever you're if you're making that transition over into Prefect, you don't have to transition all of your airflow stuff at the beginning. We wanna take that little roadblock away from you, and that's what this does here. The script is in a GitHub repository, and we'd love for you to contribute. If you have ideas, open a PR. We want this to be a great resource for future airflow defectors. Awesome, Steven. Thanks so much. And thanks to all of you for attending today. I didn't see any questions in the chat, but if you have any other questions, feel free to join us on, the Prefect community Slack, which has over 30,000 data engineers on it talking about, you know, how or they're building their orchestration platforms, how they're building their data platforms. We'd love to have you there. You can go to Prefect.io/Slack to join. And, yeah, thanks for joining us today. This webinar will be available on demand if you wanna share it with any of your colleagues, and have a great rest of your day. Thanks, everybody.