What's New in Apache Airflow® 3.3

Speaker A: You're listening to the Data ah, Flowcast, the podcast about Apache Airflow and the world of data and AI around it. Let's get into it.

Speaker B: Okay. Huh? Hi, everyone. Welcome to the Dataflow cast. I'm your host, Kenton Danis. And today we have another special episode where I am m joined by our other host, Mark Lamberti, who is our education content lead at Astronomer. We're here today to talk about the new airflow 3.3 release. Mark. Mark, how are you doing?

Speaker C: Well, very good. Thank you so much, Ketan, for having me in this episode. I'm just sweating a lot right now because it's super hot in France. But, uh, I will try my best

Speaker B: for this episode in the middle of the European heat wave, which I hope by the time 3.3 comes out and we're airing this episode, that that's over and you've gotten some relief and all of our friends in Europe are a little bit cooler.

Speaker C: But I hope so, I hope so. Otherwise I might be dead before it happens.

Speaker B: Well, we hope not. There's so much cool stuff in the Airflow project right now. You wouldn't want to miss that.

Speaker C: That's true.

Speaker B: Yeah. All right, well, we'll go ahead and dive in. Uh, so like I said, for today's podcast, we're going to talk about the new Airflow 3.3 release that has just come out. Uh, we'll talk about high level, kind of some of the stuff that's in it, and then provide some additional resources for people to learn more and digging right in. As with most Minor airflow releases, 3.3 has a ton of great stuff in it, has some new features, uh, improvements, bug fixes, all that good stuff that everyone will be interested in. I want to start by talking about some of the biggest new features that have come in this release. Uh, one of the big updates was the addition of this new Task State Store, also the Asset State store. This was AIP 103, in case anybody is following the AIP. Uh, and this high level allows tasks to persist some amount of state information. So I guess first, what do you think about this feature?

Speaker C: Yeah, uh, I think actually this is a really, really great feature. I'm very excited about 3.3, mostly because of that feature, honestly. I mean, there are other features that are great, but this one is amazing. Um, so basically the idea is, imagine that, for example, you are running a long running task. It might be like a spark job. This is actually pretty common, but it might be something else. Just think about a long running task and then, uh, for some reason your worker dies bad. Right? Um, but then because of that you might still have your, um, having your um, job, your spark job that keeps running, but just airflow is not aware of it anymore because the worker died. Uh, and so if you retry the task, you will end up with two spark jobs, uh, and you don't want that, obviously. And so the idea of the task store, one of the ideas is for you to be able to actually reattach this job to your airflow worker so that you can resume where you left. So this is pretty amazing. Um, and so the idea is, as you said, you have this task store that by default will be the airflow metadatabase behind the scene. And so you are just going to persist data that will stay even when you retry the task. So basically, uh, it is coped to your task instance. And so if you retry the task, the data will still be there, but as soon as you run a new task instance of your task, then the data is not, you know, it's something different. But that's the idea of the task store is that you can persist data across retries, uh, so that you can, for example, reattach jobs that were running and for some reason you got an issue with your workers. This is for the task store, um, for the asset store. It's a little bit different, but also pretty amazing. Now what you're going to do is instead of having your data, um, scoped to the task instance, it will be scoped to the asset. To an asset. Uh, and why this is interesting is because, well, now you can do watermarking. So for example, let's say you read some files in a S3 bucket, uh, uh, until a certain date. Well, when you want to read, when you run the same task again and you want to read the rest of the files in that bucket, or just new files that just arrived in that bucket, you don't want to read all the files again. You just want to read from the latest date at which you read the previous files. That's exactly what you can do. Now what you're going to do, for example, is that you're going to persist that date in the asset store. So as an asset, so that you can just come back where you left. And again, the idea here is that you won't have to, uh, read everything again. You will just read where you left. And obviously that applies to databases and so on. So that's pretty amazing. So that's what I would say. I mean, that's why I love uh, this new feature like the Task Store and the Asset store is because to me they are just amazing. They solve real issues, uh, that you have in the data engineering world. And also like for AI agents, if you have to, um, instead of um, I would say retrying all the actions that your agent took, now you can just come back where your agent uh, left. So pretty amazing.

Speaker B: Yeah, this is so cool. Very exciting new feature and that's a great description of it. I love the two examples you gave because one of the early questions that I've encountered when talking about this feature as it's been in development has been know we've always talked on our team about a best practice within Airflow being to design your DAG so that your tasks are atomic so that they each do kind of one unit of work. And one of the reasons that we say that is so that if something has to retry, you're not, you know, redoing a bunch of work that has already happened. Uh, so I have gotten the question like, is that still a best practice? Can you just cram all of your logic into one task now? And I think you actually gave the examples where it does make sense to use this feature in this way. So I don't think that you should just cram all of your tasks logic into a single task anymore. But it does make a lot of sense in cases where Airflow is communicating with some external service where as you said, the spark job might still be running and you don't want to submit a duplicate. Or the agentic workflows is another great example where you might have an agent that does multiple steps worth of work that you can't reasonably split into different tasks, but you still want to maintain some sort of caching so that you're not redoing the whole thing if something gets interrupted. So uh, yeah, I think in those use cases this is very, very useful.

Speaker C: Yeah, it is, it is. And also, you know, like some people may think, well why not using XCOMs? But uh, keep in mind that your XCOMS get reinitialized, you know, like destroyed after every task we try. And so that's why for example, this didn't work. Um, and that's why also we have the Task store now. Um, so now I'm pretty excited about this. It solves a couple of issues, uh, for sure. And it's very easy to actually use. Uh, it comes out of the box because again it uses the Airflow metadatabase and then to access the task store or the asset store. You just have to import, um, the task store from the context of your task and then you can access it and you can get the value that you want, you can save the value that you want. It's very easy. Pretty much like XCOMs. Um, so it's, it's really cool.

Speaker B: Yeah, we will have a guide published with some um, examples of how you can use this in your code. And yeah, like you said, it's, it's very straightforward. I think the asset store as well is really interesting and I guess a question for you because I haven't tried this yet. I assume that means that you can also use this, you know, when you're triggering other dags downstream based on an asset or an asset partition, maybe that if you had only, you know, process some of those that they would, the relevant downstream dags would be triggered, but the ones that got interrupted would not. If I'm describing this well enough for you to.

Speaker C: I. No, I mean, look, I didn't try that. But yeah, I guess, I guess it will work that way. But I didn't try that yet with um, with asset, uh, partitioning. But uh, yeah, I mean again like, very excited about it. I think the, the guide will be very um, interesting for sure. I highly recommend people to look at it because you will have a ton of examples of how this feature is great.

Speaker B: Yeah, definitely. Yeah. Well, and I've been kind of talking high level about the 3.3 release in this context of, you know, a big theme with the new features has been making failure management easier and that if we kind of think about it, like a lot of data engineering or maybe any engineering work actually is really just about managing failures. Because if it wasn't, if nothing ever failed, like you could write all of the messy code that you wanted and just ship it off, it would always be fine. Uh, so, but that's obviously not what happens in production, so this is a big piece of that. Another new feature that is part of this sort of making failure management easier is pluggable retries. Uh, which I'm really excited about. Have you had a chance to try this out yet?

Speaker C: Yeah, yeah, actually I did. And uh, also this, this is one of the features that I'm very excited about for airflow 3.3. Um, imagine that you have a task that for example, fails for um, a network issue. Well, in that case it makes sense to retry the task because maybe, uh, at some point the network issue will go away and then it will work. Um, but now imagine that you are getting uh, an issue because of a permanent, uh, like a forbidden, like a page that is forbidden, you know, 403er or something like that. Well, you can just keep retrying the task. That won't change anything, right? I mean, you will keep eating the issue and you will keep wasting resources and time and so on just because of that. So it's kind of, uh, a little bit dumb in some way. I mean, not dumb, but could be smarter. Um, and that's exactly the idea of the retry policy. I mean, that solves, uh, other things. But the idea of a retry policy is that now you can attach a retry policy to your task. So what does that mean? Well, it means that, um, basically you're gonna define a rule, like how you want to retry your task based on the exception that you get. Okay, so for example, I'm getting a HTTP error. Classic. So I'm gonna put that as an exception in my retry rule object, which is a new object that you can import port, uh, in airflow. And then once I've defined the exception, I'm going to define the action that I want to make. So either I want to retry the task or I just want to stop the task. Again, if I'm eating like a forbidden error, then I might want to stop the task right away. Okay, I don't want to retry the task. And the default is retry. Okay, so you have the option, like you have two options here. Either you retry or you stop the task. And then you can also change the retry delay if you want to. For example, if you are getting a network issue, maybe you want to wait a little bit more. Maybe you want to wait, uh, I don't know, five minutes instead of three. And again, you can define that in the retry rule. So according to the exception you will get, you can define how you want to retry the task, if you want to retry it at all, how long you want to wait before each retry, and also you can attach the reason of the retry. So you have a lot of flexibility now in how you want to retry the tasks than you used, um, to have. Now one thing that I have to say about it is if that's not enough for you, you can even subclass, um, I don't remember exactly the class name, to be honest. I think it's a retry rule or retry policy. But you can subclass this and then you can implement the logic that you want. If you want a more complex logic like, uh, I don't know. Um, I don't know. I want to add some cleaning or. The retry mechanism should be more complex than just retrying the task. Whatever you want, you can do that. Okay. You have all the freedom that you want for retrying your task. And I know that this has been a very common issue. People were saying, hey, uh, I've just retried the task, but now I want to change the retry number, for example, to retry the task more than three times. Now I want four for some reason, and I want to do that dynamically. People used to change the metadatabase, like, you know, change the data in the metadatabase. That obviously doesn't work anymore in airflow 3, and was already hacky in airflow 2. But now with the retry policy, you have all of that, uh, freedom. Uh, so all of that to say I'm very, very excited by, uh, this feature as well. I think after the task store and the asset store, this one is definitely, um, in my top list.

Speaker B: Yeah, I think this will be so impactful. And as you said, it makes retries within Airflow a lot smarter. Or less sort of brute force was the term that I was using before of like, you can just sort of keep retrying this thing until you say not to, but it's not going to do any other logic than that. So so much additional flexibility here. I think this will be really helpful for, for our team. Like, we have at least one pipeline I know of that just periodically has like a transient failure that there's nothing to do anything about other than it just happens from time to time and we can't really fix it. It's not actually a problem. But retrying in that case does make sense because it will go away if we just retry it. But we also probably don't want to retry like a minute later. We might want to wait like half an hour or something like that. Um, we also have other pipelines where I know of, like the pipelines that help us produce this podcast. Like, if there's a failure, there is nothing that retrying will help with. Like, that is something where I have to step in and fix something. And so there's just no point if it's, you know, an error from the LLM that's coming back or something like that. So, uh, so yeah, I think this will be really helpful for us and probably for a lot of teams as well.

Speaker C: Yeah, no, no, this is really cool. Uh, this is really cool. To your point, there are so many use Cases in which retry policies will be useful. Um, so I can't wait to see how people uh, uh, what policies uh, they will define. I think it's going to be an interesting one for sure.

Speaker A: This episode is brought to you by Astronomer, the team behind Astro, a managed airflow platform built for data engineering teams. From batch pipelines to training models and wiring up agents. Astro handles the infrastructure upgrades and scaling so you can focus on building. Check it out at Astronomer IO.

Speaker B: Yeah, I think, I think it's great. Hopefully we'll alleviate some of the frustration with. I don't know, do you get sort of attached to like random errors, you know, that you see come up a lot where I'm like, oh, it's that error again, like an emotional response to like one particular exception. And so feels at least useful to be able to say like, okay, for this exception we're going to do something different.

Speaker C: Yeah, yeah, yeah, no, absolutely, absolutely.

Speaker B: Yeah. Awesome. Well, I think it is also worth discussing kind of on a bit of a different topic from these last two features that we just covered. Uh, there have been some updates to the asset partitions feature. So this of course was first released in airflow 3.2, you know, a couple months ago. Um, and it has gotten, like I said, some updates for this release. So what's new with asset partitions that you think users should be aware of?

Speaker C: Yeah, that's a good question. Um, so about asset partitioning, I think like the new, um, the new update that has been, uh, that's going to be shipped with um, airflow 3.3 is what we call segment based, uh, partitioning. Um, so concretely imagine that uh, you are partitioning your table by regions, for example. Uh, so you have like eu, apac, us. Um, and so what you can do now is you can actually run a DAG for every partition and then you can have a downstream dagger that will run as soon as all of those partitions have been processed. This is you take all of your partitions and then they will be all, uh, processed under a new asset that could be all regions in that downstream dag. So this is something that now you can do uh, with segment based uh, partitioning. Um, so that's one thing and I think the other, um, the other update I think that has been made, but this has to be verified honestly, uh, is that now you can also have for every partition, again like us, eu, uh, region, you can actually trigger one DAG per asset. Um, which I think was not possible. Uh, uh, in 3.2 you have this Fan out. Uh, um, um, way of uh, of uh, of uh, processing your, your assets, your asset partitions. So I think like those two, those two uh, updates are definitely interesting and worth the time to, to dig in.

Speaker B: Yeah, definitely. I think the new mappers, I think that's what it's called, mappers, the fan out mapper, um, as you said, which allows you to then sort of take an asset and split it into a bunch of different pieces. And then as you said you can use those to schedule downstream Dags. So if you have a downstream DAG that says, okay, this one is only dealing with APAC data, but this one is dealing with EU data, you can schedule that way. I think you can also go the other way. So there's. Yes, I'm going to get the names wrong. So anybody listening definitely check our guide before you like write code based on what I'm saying here. But it's like uh, a fan in. I don't think it's actually called fan in. I think I'm getting that wrong.

Speaker C: It's the roll up, Roll up mapper.

Speaker B: Oh, roll up. Thank you. Yes, that makes more sense than fan.

Speaker C: It's fun. It's fan out and roll up mapper.

Speaker B: Roll up. Okay. Yeah. Um, yeah, so that uh, as I said, allows you to go the other way. So if you need to have a DAG that's scheduled when all of the partitions are done or something like that, um, you can do that. And of course there's other pieces involved so that you can map them to cleanly. You know, especially if this is like date time based, um, cleanly to like an hour or a day or something like that, so that it's not attached to you like the full timestamp of the DAG and stuff like that.

Speaker C: Yeah, and as you said. Right. I mean uh, uh, before it was um, it was only time based partitions and now you have like um, segment based partitioning. So that, that's the Update in uh, 3.3. So I would say if you want to use asset partitioning, you have to be aware of those mappers because that's how you can basically use asset, ah, partitioning. Um, uh, you will enjoy that.

Speaker B: Yeah, definitely. We have uh, like I said in our documentation, some examples of how you can combine those mappers that we've talked about in the code, um, to make sense of them. I think it can look a little, I don't know, intimidating maybe uh, when you first working with this feature, because there are just a lot of concepts and it can Take some. It's one of those things where I feel like I have to, like, draw it out on a piece of paper, like, yeah, okay, how are we breaking this up? Um, but once you have like a good understanding of all of those, it is actually pretty straightforward. And I think using some of the code examples will help folks implement that.

Speaker C: Yeah, yeah, no, absolutely. I think you definitely need some code examples to, uh, actually grasp how asset positioning works, because it can be simple, but it can be also quite complex. Um, so either way, you have to find, uh. And the guys are for that.

Speaker B: Yeah, definitely. Well, I do think the updates to this feature, as with most things in Airflow, they're all having ongoing development. So, uh, it's definitely top of mind, I would say, for lots of the people working on the Airflow project, trying to take user feedback from the initial iteration of a feature and making sure that they're making updates to it, uh, to make it more usable or work for more use cases or whatever it is. And so that's the case here of again, this thing that was released in the last minor version still being developed to, uh, kind of fully bake it so that again, it's useful to the most number of people possible. So very cool to see those updates in the project.

Speaker C: Completely agree, agree.

Speaker B: Yeah. And then so, as I mentioned at the beginning, of course, minor releases always have so much stuff in them. We'll direct everybod to the release notes because all of it is relevant to at least someone. Um, we can't go through all of it in our podcast today, but I'm curious of the remaining, I'll say, quote, smaller features because again, they might not be smaller to whoever's using them, but of the stuff that we're not covering more in depth in this episode, what are you most excited about?

Speaker C: Well, interestingly, I don't think, uh, it's a small feature, but in airflow 3.3, I believe you will be able to run tasks in Java and go, uh, so I think those, you know, for those who I believe mostly like for Java, I think, like, there are quite many people that obviously interact with Java every day. Um, and, uh, the fact that now you will be able to run like, you know, Java using Airflow directly without having like, I remember that someone told me, hey, we are using like custom operator that runs Jar. They were doing like some crazy stuff to run Java code. Well, guess what, now you won't have to do that anymore. I have no idea how mature is the feature, to be honest. Uh, you can expect that there will be some issues. Uh, but I think it's quite exciting to see that airflow is going to be like uh, a multi language, um, Orchestrator at some point. I mean everything will be still in Python but you will be able to run tasks in different languages, which I think is quite amazing.

Speaker B: Yeah, yeah, it's really cool as you said, not actually a smaller feature. And I think it will be super impactful for teams who are moving from another orchestrator where things are more written in another language. I know Java, that's really common, especially with legacy Orchestrators. If you're exporting jobs that were built with like a drag and drop or something, oftentimes that ends up in Java. Uh, and you can now just run that uh, without having to convert everything to Python. So yeah, I do think I'm also going to get this wrong. So check the documentation. But of the Java and Go ones I think one of them is still experimental at least maybe both of them, I can't remember. But keep, keep that in mind. The, the ui, um, or the user interface might change a little bit. But yeah, uh, yeah, we're excited to see what people are able to build with this.

Speaker C: Same. No, not very exciting.

Speaker B: Yeah, that's a great answer. When I was just looking through the release notes, I think one of my favorites uh, that stood out to me was you can now with DAG versioning, uh, choose in the ui, if you rerun um, a uh, past DAG run, you can choose whether you want to use the most recent DAG version or an older DAG version. I think before this if you were using a local DAG bundle you would use whatever the version was at that time, um, that it ran even if you were on a newer version. And with the um, uh, no, sorry, I said that wrong. With the local DAG bundle you'll use the most recent version of the code. With a version DAG bundle you'd use the version at the time that you ran it. Uh, you could do either or, but now you just have the option, um, I'm pretty sure this is only if you're using a version DAG bundle that that's possible but uh, just a small piece of addition choice on the part of the user. So I can see where this will be useful in the case where like you've shipped new code and that has caused a failure and so you want to just quickly rerun with the older version to get things back on track and then go fix whatever has gone wrong.

Speaker C: Yeah, no, yeah, it's going to Be great. And also like, for people that don't, don't know about, um, DAG bundles, I highly recommend you to look at that feature. Yeah, I won't say, I won't say anything else. Just look at them.

Speaker B: Yeah, definitely. And then I guess very quickly, you know, for anybody who, who might not already be upgraded to airflow 3, because we know there are still users in the community who are on Airflow 2 and they're interested in the features in 3.3, uh, or any of the other minor versions in the 3. X series. Uh, what is your advice to them for going through that upgrade process?

Speaker C: Yeah, so I think we have a. I don't think. I'm sure we have a great ebook for that, uh, to upgrade from airflow 2 to airflow 3. I highly recommend you to uh, take a look at it. And also we have auto, uh, which if you want to go with the AI way of upgrading your dags from airflow 2 to airflow 3 and follow best practices and so on, I think it definitely worth the shot. Um, because uh, it's going to save you a lot of time.

Speaker B: Yeah, absolutely. It can take over a lot of the kind, uh, of manual changes, especially minor changes in your code that you have to make. Uh, yeah, extremely helpful.

Speaker C: Nobody wants to do my migrations, so

Speaker B: no, talk about like undifferentiated heavy lifting. Especially if you have a lot of dags, that's not a fun process to go through. So uh, it's great to offload that to some of your AI coworkers, those that have been trained specifically in Airflow.

Speaker C: That's right, yeah.

Speaker B: Cool. Well, we will throw links to all of those things in the show notes. I know this has been just a very high level overview of some of the new stuff. In 3.3 we will be having a webinar, uh, that we will also link in the show notes if folks want to register where we'll go through all of this in a little bit more detail and show some code examples and a demo and stuff like that. So a little more tangible than an audio version of a podcast. So that's a great follow up resource. But yeah, we're really excited about the new release. There's always so much stuff going on in the Airflow project and the momentum is still there. So it's great to be able to talk about another one.

Speaker C: Yeah, can't wait for it. Thank you so much, Gaetan for having me. Was a pleasure.

Speaker B: Yeah, thanks Mark, for going through all this.

Speaker C: Thank you so much.

Speaker A: See you thank you for listening to the data. Uh, Flowcast. Check the show notes for links to everything mentioned today. If you're enjoying the podcast, subscribe so you never miss an episode and ah, leave us a five star, uh, review.