WEBVTT
00:00:00.239 --> 00:00:07.519
You're listening to the Data Flowcast, the podcast about Apache Airflow and the world of data and AI around it.
00:00:07.759 --> 00:00:11.039
We are excited to invite you to orchestrate everything.
00:00:11.279 --> 00:00:14.640
Our free online conference on September 60th.
00:00:15.039 --> 00:00:21.920
Engineering leaders from Lyft, Ram, and Wix will share how they are shipping production AI workflows on Airflow.
00:00:22.239 --> 00:00:28.320
Plus a crash course for the new Astronomer Certified AI Orchestration Fundamentals exam.
00:00:28.559 --> 00:00:31.679
Attendees get a discount code for a free certification.
00:00:32.000 --> 00:00:34.159
A$150 value.
00:00:34.399 --> 00:00:39.439
Go to astronomer.link slash data minus flowcast minus OE.
00:00:39.759 --> 00:00:41.119
Link in the show notes.
00:00:41.359 --> 00:00:43.200
Now let's get into the show.
00:00:43.920 --> 00:00:44.799
Hello everyone.
00:00:44.960 --> 00:00:46.960
Welcome, welcome to the Data Flowcast.
00:00:47.039 --> 00:00:55.280
I'm your host, Mark Lomerti, and today I'm super excited to be joined by Ivana Izajovich, big data engineer at Cinecron.
00:00:55.600 --> 00:00:57.920
First thing first, how are you doing today?
00:00:58.640 --> 00:01:00.640
Hi Mark, thank you so much for having me.
00:01:00.799 --> 00:01:03.280
I'm really happy to be here and I'm doing great.
00:01:03.520 --> 00:01:04.239
Thank you.
00:01:04.560 --> 00:01:12.640
I'm especially excited about today's topic because uh airflow is something I worked uh with very close every day.
00:01:13.120 --> 00:01:14.719
Yeah, I'm very excited about this.
00:01:14.799 --> 00:01:22.159
Uh because actually, you know, about today's podcast, uh, we will deep dive into orchestrating, um, strictly say controlled ETL pipeline.
00:01:22.239 --> 00:01:27.680
I think you have like a way of doing things that I think is truly interesting.
00:01:27.920 --> 00:01:33.040
Um also how you're gonna how did you survive about the subdag deprecation?
00:01:33.200 --> 00:01:41.840
Um, because subdags have been very popular for a while, but as you uh will tell us, you actually uh did the migration, the migration.
00:01:42.000 --> 00:01:48.079
Uh and then also we will uh talk about adapting custom retrail logic all the way through uh Airflow 3.
00:01:48.159 --> 00:01:49.680
So a lot of things to cover.
00:01:49.760 --> 00:01:53.519
Uh so without further ado, let's dive into this amazing episode.
00:01:53.760 --> 00:02:00.480
So, first thing first, um can you introduce yourself and tell us a bit more about what you do at your company?
00:02:01.200 --> 00:02:08.159
Uh okay, as I said, uh as you said, uh my name is Ivana and I work as a senior big data engineer.
00:02:08.319 --> 00:02:15.039
And in my daily work, I mostly focused on big data ETL processes, data processing, uh, and orchestration.
00:02:15.199 --> 00:02:25.439
So basically, I work on making sure the data from different sources is important, transformed, validated, and then made available for reports or some kind of analytics.
00:02:25.840 --> 00:02:41.280
And a big part of my role is orchestration, as I mentioned, uh connecting different steps in a pipeline, making sure that they run in the right order, handling retries, monitoring alerts, and overall pipeline stability.
00:02:41.599 --> 00:02:46.319
And as you know, uh airflow plays a very important role in that.
00:02:46.639 --> 00:02:54.560
Because uh, for us, it's not just a scheduler, it's the central place where we control and monitoring the whole data pipeline.
00:02:55.360 --> 00:02:59.840
Well, I think what you described uh will resonate a lot uh with many data engineers.
00:02:59.919 --> 00:03:05.199
I think like they will all uh they will all uh feel that they are doing exactly what you are doing.
00:03:05.439 --> 00:03:12.560
Um so can you actually uh just out of curiosity, uh, because you just said that Airflow is uh is uh crucial.
00:03:12.639 --> 00:03:17.759
Um but uh when you joined your company, Airflow was already there?
00:03:18.639 --> 00:03:20.800
Uh yeah, uh I think so.
00:03:21.120 --> 00:03:21.919
Oh okay, okay.
00:03:22.000 --> 00:03:23.680
So actually, I'm just gonna keep that.
00:03:23.759 --> 00:03:25.039
I I have a question for you.
00:03:25.120 --> 00:03:32.080
But uh first thing first, uh can you tell us a little bit more about the data stack that you have today and uh where does Airflow fit in?
00:03:32.560 --> 00:03:32.879
Okay.
00:03:33.199 --> 00:03:37.039
On the current project, uh our stack has several main components.
00:03:37.199 --> 00:03:40.080
Uh we use Snowflake as a data source.
00:03:40.159 --> 00:03:48.479
Uh then we load and organize data in MongoDB uh using medallion architecture, brose, silver, and gold layers.
00:03:48.639 --> 00:04:06.879
Uh we use Spark for transformations and processing, uh, Airflow for orchestration, uh for monitoring and error handling also, and uh Elasticsearch for storing uh reports that we are used for application later.
00:04:07.120 --> 00:04:11.759
And uh Airflow is basically uh, as I said, control uh center of the system.
00:04:11.919 --> 00:04:24.319
It decides when each ETL process starts, in which order task runs, um, uh what happens if uh something fails, and when notifications or alerts should be sent.
00:04:24.560 --> 00:04:28.240
And without Airflow, the pipeline would be very hard to control.
00:04:28.560 --> 00:04:32.399
And because we had many, many dependencies between processes.
00:04:32.560 --> 00:04:35.839
And it's not just uh run a spark job.
00:04:35.920 --> 00:04:39.199
Uh the whole flow has to be happened in the correct order.
00:04:40.079 --> 00:04:52.319
Okay, well, actually that that that creates a bridge to the my next my next question, um, which is you just mentioned that that you are using Airfo for orchestrating your ETL um jobs, maybe more than just ETLs.
00:04:52.399 --> 00:04:59.199
Do you have like any like maybe uh other, you know, um like maybe machine learning use cases, things like that also?
00:04:59.279 --> 00:05:03.519
Or is it just like uh no this no, just ETL in this project, yeah.
00:05:04.160 --> 00:05:04.319
Okay.
00:05:04.480 --> 00:05:13.360
And you you just mentioned that dependencies, I mean, uh you definitely have to execute things in the right order, um, which is definitely what Airflow is made for.
00:05:13.680 --> 00:05:20.480
But uh can you tell us a little bit more about you know the ETL uh processes that you have today?
00:05:20.639 --> 00:05:27.600
Um and you know why do you actually require this sequence order that it is you know so important for you?
00:05:27.759 --> 00:05:32.959
Because I think you also have a way to do uh you know uh things that are interesting.
00:05:33.360 --> 00:05:47.040
Uh yeah, this was actually one of the points where things become really interesting and honestly a bit uh painful at the beginning, uh, because at first we tried to uh solve it using the standard airflow options.
00:05:47.120 --> 00:05:56.959
Uh we looked at configurations, uh dog run limits, uh we uh tried to use pools, different uh dependency setups, and so on.
00:05:57.199 --> 00:06:05.360
Basically everything that seems uh like it could help, but uh every solution uh solved just one part of the problem that we had.
00:06:05.519 --> 00:06:19.279
And at the same point, in some point we were stuck and uh not completely desperate, but uh close because we really realized that the problem was bigger than just how we uh stopped two DUG from running at the same time.
00:06:19.519 --> 00:06:30.959
And the real question was here uh how do we control the full execution flow across multiple DUGs, especially when some of them can be triggered at any uh moment from the application side?
00:06:31.360 --> 00:06:51.759
And uh we had to take a step back, sit down, and literally draw the whole process uh end-to-end, uh, which DAGs uh uh which DAGs uh exist, uh, what they um touch, which collection uh they use, which layers they write to, and what could go wrong if two processes uh overlap.
00:06:52.240 --> 00:06:58.399
And the main reason for strict sequential execution was that uh for uh was data consistency.
00:06:58.639 --> 00:07:04.879
Uh all of these DUGs were used to the same MongoDB collections at the same layers.
00:07:05.040 --> 00:07:10.160
And in the bronze layer, uh data could be overwritten by a new snapshot.
00:07:10.319 --> 00:07:13.279
In silver, we had cleaned and prepared data.
00:07:13.519 --> 00:07:19.360
In gold, we had aggregated data used by the applications, uh application reports.
00:07:19.600 --> 00:07:33.600
And so if the two DUs run in the same time and change the same collection or layer in layers, we could uh easily end up with collision uh in uh inconsistency data or partially processed uh results.
00:07:33.759 --> 00:07:41.680
And that's why uh the sequential order uh execution was not just nice to have, it was absolutely necessary for our case.
00:07:43.040 --> 00:07:43.439
Wow.
00:07:43.600 --> 00:07:47.040
So now I I need to learn more about like how did you solve that?
00:07:47.120 --> 00:07:53.199
Um because I'm I'm sure that it's a very, very common issue uh you know in the in the like for many engineers.
00:07:53.279 --> 00:07:55.680
Like how did you what did you try?
00:07:56.319 --> 00:08:02.639
Oh uh before building a custom solution, uh we first tried to use uh waterflow already have.
00:08:02.959 --> 00:08:10.079
Uh we started with the Dug level uh configuration, of course, like limiting the number of uh active DAG run.
00:08:10.480 --> 00:08:16.399
Uh that helped a little because the same DUG could not run multiple times in parallel.
00:08:16.560 --> 00:08:20.879
That was great, but it didn't uh solve the biggest problem here.
00:08:21.040 --> 00:08:23.439
Different DAGs could still run at the same time.
00:08:23.680 --> 00:08:26.399
And then we uh looked at pools also.
00:08:26.720 --> 00:08:31.920
Pools are useful when you have to limit some many tasks, but in our case uh it wasn't a problem.
00:08:32.320 --> 00:08:43.279
And uh uh our problem was uh on a higher level, we needed to control the execution order uh of entire DAGs, not just individually task or individually DAG.
00:08:43.679 --> 00:08:53.279
And the biggest challenge, yeah, can you No, I was I was about to say so, yeah, it's it's really like you you actually want to have multiple DAGs that need to run one after the other.
00:08:53.440 --> 00:09:02.559
It's not because, like you know, for people that that are uh wondering about pools, uh pools allow you to um you know control the the parallelism of your tasks.
00:09:02.720 --> 00:09:08.399
So basically, if you want to execute one task after another, you can do that with pool and then you set one slot to that pool.
00:09:08.799 --> 00:09:21.919
But you will have to attach uh all the tasks to that pool, but still it doesn't solve truly the the issue that uh now you have Ivana uh with the the DAGs that should not run, multiple DAGs that should not run at the same time.
00:09:22.879 --> 00:09:23.679
Yeah, sorry.
00:09:24.559 --> 00:09:26.559
Yeah, it was the problem here.
00:09:26.720 --> 00:09:34.639
And the biggest challenge, as I said, uh was that the some DAGs are scheduled and uh the other DUG was even uh event-driven.
00:09:34.799 --> 00:09:39.600
And uh that event-driven DUG uh could be triggered at any moment from the application.
00:09:39.679 --> 00:09:53.600
And that's why we need to have uh control Q uh and uh uh and we need to have uh uh the right order when uh one DUG uh runs and so on.
00:09:53.840 --> 00:09:56.960
So standard airflow configuration was not enough for us.
00:09:57.120 --> 00:10:03.519
We needed something uh like control Q, as I said, that could decide which process was allowed to run next.
00:10:03.679 --> 00:10:07.759
And that's why we built a custom orchestration layer for that.
00:10:08.639 --> 00:10:15.679
Okay, so tell me tell me more about this uh this um this uh layer that you that you built.
00:10:16.320 --> 00:10:20.399
Yeah, in the end we created a three-layer DAG architecture.
00:10:20.559 --> 00:10:24.320
Uh the first layer is the scheduler or trigger DAGs.
00:10:24.399 --> 00:10:27.440
Uh they are the entry point into the system.
00:10:27.600 --> 00:10:38.240
Uh some are scheduled, for example, for daily imports, uh, and some one uh some are event-driven, triggered by uh user applications uh from application side.
00:10:38.480 --> 00:10:43.519
And their job is not just to run the full uh ETL process directly.
00:10:43.679 --> 00:10:48.399
Their job is to register a request and say this process needs to be executed.
00:10:48.639 --> 00:10:52.559
And the next one, the second layer, is the proxy DAG.
00:10:52.720 --> 00:10:55.919
And this uh here we uh get the magic.
00:10:56.240 --> 00:11:04.080
This is the most important part of the solution because the proxy DUG controls the queue and decides which process can run next.
00:11:04.240 --> 00:11:15.600
Uh, it checks if there is already an active uh DUG, uh, what is waiting uh in the queue, what has priority, or whether the next process is allowed to start.
00:11:15.840 --> 00:11:23.759
And so instead of every DUG decide independently when to run, everything needs to go through the proxy layer.
00:11:23.919 --> 00:11:28.320
And the proxy layer is the main uh brain here.
00:11:28.720 --> 00:11:31.759
Uh the third layer is the main ETL DUGs.
00:11:31.919 --> 00:11:37.840
Um, these DUGs contain the actual logic, data import transformation, validation, and so on.
00:11:38.000 --> 00:11:53.039
And these design help us uh uh uh separate uh responsibilities, uh trigger DAG, as I said, initiate uh requests, the proxy controls uh execution order, and main DUGs go to the actual processing.
00:11:53.919 --> 00:12:00.159
And basically we solve uh our problem with this with this architecture.
00:12:00.799 --> 00:12:02.240
This is this is super interesting.
00:12:02.320 --> 00:12:03.759
Um I have a few questions.
00:12:03.919 --> 00:12:07.759
Um so the the um what exactly is this queue?
00:12:08.000 --> 00:12:10.159
Like uh technically speaking, what is this queue?
00:12:10.480 --> 00:12:12.240
Is it like uh Kafka queue?
00:12:12.399 --> 00:12:13.600
Is it like like what is it?
00:12:14.240 --> 00:12:15.679
It's uh DAG queues.
00:12:16.320 --> 00:12:17.759
Okay, okay, okay, I see.
00:12:17.919 --> 00:12:24.000
Um and uh uh so those those layers are they like how they are built?
00:12:24.159 --> 00:12:27.279
Are they like all like like what is exactly a layer?
00:12:27.360 --> 00:12:28.399
Is it like a DAG?
00:12:29.360 --> 00:12:31.360
Yeah, yeah, everything yeah, it's DAG.
00:12:33.200 --> 00:12:33.840
Yes.
00:12:34.639 --> 00:12:44.240
We have uh scheduler, we have several scheduler DAGs, and we have just one proxy DAG, and we have several uh uh main DAGs.
00:12:44.799 --> 00:12:45.679
Okay, I see, I see.
00:12:45.759 --> 00:12:55.440
Because like I'm I'm just wondering because you know some some people may think that it's it might be like uh you know a plugin that you build or uh you know a Python scripts uh Python script that you run.
00:12:55.759 --> 00:13:12.960
I think like you know, I mean we we also uh we also talked a little bit more about I mean previously uh before this uh this episode uh we discussed a little bit more about how you you know how your um your DAGs are uh um uh written and you know how they are generated and so on.
00:13:13.039 --> 00:13:16.320
Uh I think you have some DAGs that have a lot of tasks.
00:13:16.960 --> 00:13:21.440
Um and I was just wondering like how do you handle that?
00:13:21.600 --> 00:13:24.799
Like how do you handle the creation of your DAGs, I would say?
00:13:24.960 --> 00:13:29.519
Uh do you use like you know, uh a Python script to generate some of your DAGs?
00:13:29.679 --> 00:13:31.039
Do you have like some templates?
00:13:31.279 --> 00:13:33.759
Do you have you know dynamic loops, whatever?
00:13:33.840 --> 00:13:37.039
Like can you tell us a little bit more of how you author those DAGs?
00:13:37.200 --> 00:13:46.480
Because I imagine that you have all of those layers in place, I mean those three layers in place, but I imagine that your DAGs are really standardized as well, so that they can actually run.
00:13:46.960 --> 00:13:47.679
Yeah, of course.
00:13:47.840 --> 00:13:57.360
Uh we have uh DAGs with uh approximately 200 tasks, and of course, we definitely did not write uh 200 tasks manually.
00:13:57.519 --> 00:13:59.679
Uh that will be impossible to maintain.
00:13:59.840 --> 00:14:05.279
And the idea was to make the DAG structure usable and configuration driven.
00:14:05.519 --> 00:14:09.840
And each main DAG has several logical uh sections here.
00:14:10.080 --> 00:14:22.000
On the top level, it looks like uh it looks like uh much simpler, maybe around 30 or 40 main steps, but inside those steps we use task groups.
00:14:22.320 --> 00:14:27.519
And it's in uh inside of task groups, we have more task groups.
00:14:27.679 --> 00:14:31.840
And on that way, we create dynamically generated tasks here.
00:14:32.000 --> 00:14:37.200
Uh uh when everything is expanded, we get around 200 tasks per DAG.
00:14:37.279 --> 00:14:37.840
Yeah.
00:14:38.159 --> 00:14:44.399
And also we use configuration file, define what uh changed between different DAG versions.
00:14:44.559 --> 00:14:52.879
For example, which collections are used, which steps are active, uh, which parameters are needed, uh, how monitoring works, and so on.
00:14:53.039 --> 00:14:56.399
Uh, also versioning uh was very important for us.
00:14:56.559 --> 00:15:00.799
And after every release or hotfix, we increase the DAG version.
00:15:00.879 --> 00:15:14.320
And DAG version, that number is related with the release number, and uh that helps us uh keep a clean and controlled pipeline without new logic uh affecting already running processes.
00:15:14.720 --> 00:15:17.120
And we solve that on that way.
00:15:17.279 --> 00:15:21.279
And with large DUGs like this, uh monitoring is crucial.
00:15:21.519 --> 00:15:26.799
It's not enough to know uh what some uh uh that something uh failed.
00:15:26.960 --> 00:15:34.480
You need to know uh where it failed and why it failed and uh whether you can safely restart only that part.
00:15:34.799 --> 00:15:37.279
That was the logic behind all this.
00:15:38.080 --> 00:15:46.720
This episode is brought to you by Astronomer, the team behind Astro, a managed airflow platform built for data engineering teams.
00:15:46.879 --> 00:15:50.879
From batch pipelines to training models and wiring up agents.
00:15:51.039 --> 00:15:57.120
Astro handles the infrastructure, upgrades, and scaling so you can focus on building.
00:15:57.279 --> 00:15:59.519
Check it out at astronomer.io.
00:16:01.200 --> 00:16:06.240
Okay, um, and so this this uh those configuration files are they like YAML files?
00:16:06.559 --> 00:16:09.360
Uh yeah, I use YAML file for the configuration file.
00:16:09.840 --> 00:16:10.000
Great.
00:16:10.080 --> 00:16:14.559
So you actually mentioned task groups, um, which I think is uh definitely an interesting feature.
00:16:14.639 --> 00:16:36.960
It's a simple one, but it's uh actually a very useful one because to your point, when you have like hundreds of tasks, uh sometimes thousands of tasks, uh grouping your tasks together might be a good idea, um especially if you don't want like to actually almost crash the UI, uh or at least it's more it you won't be able to navigate through your DAG because there there will be way too many tasks.
00:16:37.440 --> 00:16:42.720
Um but before task groups you were not using them, obviously.
00:16:42.799 --> 00:16:47.200
You were some using something different, which I think uh was the sub DAGs.
00:16:47.279 --> 00:16:49.120
Uh and subdags were very popular.
00:16:49.919 --> 00:16:55.919
Can you tell us a little bit more of how how did you migrate from the sub DAGs to the task groups?
00:16:56.080 --> 00:17:06.799
Um, because task groups are easier to use than sub DAGs, but you know, I think like many people actually had to migrate, and you are the first person that finally will tell me how they did that migration.
00:17:07.279 --> 00:17:08.559
So yeah, yeah.
00:17:08.799 --> 00:17:18.799
Uh subdags were useful for us because they allowed us to split one large ETL process, one large uh DAG into logical parts, as I said before.
00:17:18.960 --> 00:17:24.559
For example, one part could be import data, another transformations, another validation, and so on.
00:17:24.799 --> 00:17:30.720
If only one part failed, we could retry just part instead of rerunning the whole DUG.
00:17:31.039 --> 00:17:32.960
That was the point here.
00:17:33.279 --> 00:17:37.039
And when subdugs became deprecated, we had two main options.
00:17:37.200 --> 00:17:41.599
The first one was to replace each subdugs with a separate DAG.
00:17:41.759 --> 00:17:44.880
Uh, from retry perspective, uh, that made sense.
00:17:45.119 --> 00:17:50.079
But uh each logical part could be triggered or retried independently.
00:17:50.240 --> 00:18:04.400
And however, for our case, we would lose the visibility of the full pipeline in one place and introduce more complexity around uh dependencies, trigger triggering logic, uh status tracking, error handling, and so on.
00:18:04.640 --> 00:18:14.480
Since we already had strict functional execution uh requirements, splitting everything into many separate DAGs would make the system harder to monitor and control.
00:18:14.960 --> 00:18:19.839
And that was the main reason why we moved to task groups.
00:18:20.079 --> 00:18:30.799
And task groups uh are better for organizing uh tasks, uh visualization inside one DAG, but they do not have behavior exactly like sub DAGs.
00:18:30.880 --> 00:18:32.720
And that was the first challenge here.
00:18:33.039 --> 00:18:40.240
So the main thing we lost was that availability to retry a whole group as one logical unit.
00:18:40.559 --> 00:18:45.119
So we built custom retryal logic on the top of task groups.
00:18:45.359 --> 00:18:53.279
In Airflow 2.9, uh, we solved this through the metadata uh data DB uh based on the task group identifier.
00:18:53.359 --> 00:18:57.599
Uh we knew which task instance belongs to that group.
00:18:57.839 --> 00:19:08.799
So we needed to retry the group, uh, our logic uh restart the state of the those um task instances so Airflow could run that part again.
00:19:09.039 --> 00:19:16.559
Uh the important thing uh uh is that uh this uh this was not a random manually database update.
00:19:16.720 --> 00:19:20.000
It was control logic inside our orchestration layer.
00:19:20.640 --> 00:19:23.920
So that gave us the same practical benefit as before.
00:19:24.079 --> 00:19:30.160
Now we had uh now we have uh task groups, but with a retry mechanism as subdag.
00:19:30.960 --> 00:19:32.000
Which is amazing, right?
00:19:32.079 --> 00:19:48.960
I mean I think like many many people actually had this issue when task groups um you know um uh came to uh I don't remember the the Airflow uh version, but when they actually replaced the sub DAGs, like people were like, okay, this is great, but also like now I cannot retry my task group.
00:19:49.039 --> 00:19:51.039
I have to retry like each task one by one.
00:19:51.119 --> 00:19:52.160
I I don't want to do that.
00:19:52.240 --> 00:19:56.480
And so I think like many people actually did the exact same solution that you just described.
00:19:57.279 --> 00:20:01.200
But in Airflow 3, I believe this is not possible anymore, right?
00:20:02.240 --> 00:20:02.880
Yeah, of course.
00:20:03.759 --> 00:20:08.160
How did you solve again this uh this you know with with Airflow 3?
00:20:08.559 --> 00:20:10.480
Again, unexpected behavior.
00:20:10.720 --> 00:20:23.200
Uh with the Airflow 3, we had to change this approach because relying directly on the metadata dB was no longer the right direction because of the new architecture, Airflow architecture.
00:20:23.359 --> 00:20:27.680
So we moved to uh the retry logic to the Airflow REST API.
00:20:27.920 --> 00:20:40.319
Now instead of directly changing the metadata dbs, db tables, uh we communicate with the airflow through the officially uh official uh APIs.
00:20:40.559 --> 00:20:42.960
Uh now we need to control a dog run.
00:20:43.359 --> 00:20:52.640
When we need to control DAG run uh task instance or anything else, uh or retry behavior, uh we do that through uh supported endpoints.
00:20:52.799 --> 00:20:55.359
And this is better for several reasons.
00:20:55.599 --> 00:21:10.880
The first one is more uh stable across Airflow versions, because uh if uh you rely on internal database table, something can change after an upgrade, as we had um experienced in previous uh versions, the Airflow versions.
00:21:11.119 --> 00:21:13.839
And the second, uh, it's more secure.
00:21:14.000 --> 00:21:29.599
Uh also I think that uh the architecture is clear uh clearer now, and Airflow remains responsible for its own uh internal uh state, and we interact with with uh with it through the official uh API layer.
00:21:29.759 --> 00:21:37.039
Uh so finally it is easier to maintain and uh maintain and explain uh to the new team member uh also.
00:21:37.440 --> 00:21:45.599
So in Airflow 2.9, our metadata db solution worked, but with Airflow 3, we modernize it basically.
00:21:46.720 --> 00:21:54.799
Yeah, and I think this is um, you know, like for people that are wondering, so in Airflow 3 now you cannot access the Airflow metadatabase directly anymore.
00:21:55.039 --> 00:21:57.519
Um for a good reason, like for security reason.
00:21:57.599 --> 00:22:02.960
Like honestly, uh I mean it was good, it was good for us, but it was also a little bit hacky, right?
00:22:03.039 --> 00:22:09.920
You you should not uh you should not expect that your DAGs could access the Airflow meta database, so that's why this has been changed.
00:22:10.160 --> 00:22:19.119
Um but it's to your point, it's way more elegant uh to use the the the Airflow Rest API, uh especially as in Airflow 3, I think you know.
00:22:19.279 --> 00:22:26.079
I mean basically they keep adding new endpoints so you can do more and more things with the Airflow SPI, it's much better than it used to be.
00:22:26.640 --> 00:22:30.559
Um so I'm I'm glad that you overcame this issue again.
00:22:30.799 --> 00:22:35.759
I hope that in Airflow 4 you won't have to come up with uh something else.
00:22:36.000 --> 00:22:38.400
Um awesome, thank you so much, Ivana.
00:22:38.720 --> 00:22:39.759
Uh time flies.
00:22:39.920 --> 00:22:46.960
We are already uh I think it's already uh already uh three 30 minutes.
00:22:48.079 --> 00:22:54.079
I have to ask you first, um what is the best way for folks to get in touch with you?
00:22:54.240 --> 00:23:01.599
You know, if they have any questions regarding your solution and again like all of the common issues that you uh actually um solved.
00:23:02.480 --> 00:23:03.119
Okay, thank you.
00:23:03.279 --> 00:23:04.960
First, uh thank you for having me.
00:23:05.119 --> 00:23:07.279
It was really great talking about this.
00:23:07.519 --> 00:23:11.519
And the best way to get in touch with me is through LinkedIn, of course.
00:23:11.680 --> 00:23:16.720
And I'm I'm always happy to connect with the people uh with the same topics and so on.
00:23:17.039 --> 00:23:19.039
So I want to add something.
00:23:19.200 --> 00:23:29.440
Uh I think these kind of challenges are very interesting uh because they show the orchestration is not just only for scheduling jobs, and uh, Airflow is not just for that.
00:23:29.599 --> 00:23:38.160
Uh there are always unexpected situations where you have to learn something new and you know you need to adapt and find a practical solution.
00:23:38.319 --> 00:23:43.119
So for me, uh that part of the work uh was really interesting.
00:23:43.359 --> 00:23:47.519
And yeah, uh everyone can feel free to reach out on LinkedIn.
00:23:47.680 --> 00:23:49.599
I'll be happy to connect.
00:23:50.240 --> 00:23:50.880
That's awesome.
00:23:50.960 --> 00:23:55.519
Well, first thing first, uh Ivana, I have to thank you for uh for being here.
00:23:55.680 --> 00:23:58.400
Uh so thank you so much again for joining this episode.
00:23:58.480 --> 00:23:59.440
It was a wonderful one.
00:23:59.519 --> 00:24:02.160
I'm very happy uh about everything you said.
00:24:02.319 --> 00:24:05.680
I think like people will enjoy all the insights that you just gave.
00:24:05.920 --> 00:24:08.720
Um and well, I hope to see you again for another one.
00:24:08.880 --> 00:24:09.519
Who knows?
00:24:09.680 --> 00:24:11.599
But otherwise, I wish you a wonderful day.
00:24:11.759 --> 00:24:13.359
Take care and soon.
00:24:14.400 --> 00:24:17.359
Thank you for listening to the data flowcast.
00:24:17.599 --> 00:24:21.039
Check the show notes for links to everything mentioned today.
00:24:21.200 --> 00:24:25.680
If you're enjoying the podcast, subscribe so you never miss an episode.
00:24:26.000 --> 00:24:28.160
And leave us a five-star review.
00:00:00.239 --> 00:00:07.519
You're listening to the Data Flowcast, the podcast about Apache Airflow and the world of data and AI around it.
00:00:07.759 --> 00:00:11.039
We are excited to invite you to orchestrate everything.
00:00:11.279 --> 00:00:14.640
Our free online conference on September 60th.
00:00:15.039 --> 00:00:21.920
Engineering leaders from Lyft, Ram, and Wix will share how they are shipping production AI workflows on Airflow.
00:00:22.239 --> 00:00:28.320
Plus a crash course for the new Astronomer Certified AI Orchestration Fundamentals exam.
00:00:28.559 --> 00:00:31.679
Attendees get a discount code for a free certification.
00:00:32.000 --> 00:00:34.159
A$150 value.
00:00:34.399 --> 00:00:39.439
Go to astronomer.link slash data minus flowcast minus OE.
00:00:39.759 --> 00:00:41.119
Link in the show notes.
00:00:41.359 --> 00:00:43.200
Now let's get into the show.
00:00:43.920 --> 00:00:44.799
Hello everyone.
00:00:44.960 --> 00:00:46.960
Welcome, welcome to the Data Flowcast.
00:00:47.039 --> 00:00:55.280
I'm your host, Mark Lomerti, and today I'm super excited to be joined by Ivana Izajovich, big data engineer at Cinecron.
00:00:55.600 --> 00:00:57.920
First thing first, how are you doing today?
00:00:58.640 --> 00:01:00.640
Hi Mark, thank you so much for having me.
00:01:00.799 --> 00:01:03.280
I'm really happy to be here and I'm doing great.
00:01:03.520 --> 00:01:04.239
Thank you.
00:01:04.560 --> 00:01:12.640
I'm especially excited about today's topic because uh airflow is something I worked uh with very close every day.
00:01:13.120 --> 00:01:14.719
Yeah, I'm very excited about this.
00:01:14.799 --> 00:01:22.159
Uh because actually, you know, about today's podcast, uh, we will deep dive into orchestrating, um, strictly say controlled ETL pipeline.
00:01:22.239 --> 00:01:27.680
I think you have like a way of doing things that I think is truly interesting.
00:01:27.920 --> 00:01:33.040
Um also how you're gonna how did you survive about the subdag deprecation?
00:01:33.200 --> 00:01:41.840
Um, because subdags have been very popular for a while, but as you uh will tell us, you actually uh did the migration, the migration.
00:01:42.000 --> 00:01:48.079
Uh and then also we will uh talk about adapting custom retrail logic all the way through uh Airflow 3.
00:01:48.159 --> 00:01:49.680
So a lot of things to cover.
00:01:49.760 --> 00:01:53.519
Uh so without further ado, let's dive into this amazing episode.
00:01:53.760 --> 00:02:00.480
So, first thing first, um can you introduce yourself and tell us a bit more about what you do at your company?
00:02:01.200 --> 00:02:08.159
Uh okay, as I said, uh as you said, uh my name is Ivana and I work as a senior big data engineer.
00:02:08.319 --> 00:02:15.039
And in my daily work, I mostly focused on big data ETL processes, data processing, uh, and orchestration.
00:02:15.199 --> 00:02:25.439
So basically, I work on making sure the data from different sources is important, transformed, validated, and then made available for reports or some kind of analytics.
00:02:25.840 --> 00:02:41.280
And a big part of my role is orchestration, as I mentioned, uh connecting different steps in a pipeline, making sure that they run in the right order, handling retries, monitoring alerts, and overall pipeline stability.
00:02:41.599 --> 00:02:46.319
And as you know, uh airflow plays a very important role in that.
00:02:46.639 --> 00:02:54.560
Because uh, for us, it's not just a scheduler, it's the central place where we control and monitoring the whole data pipeline.
00:02:55.360 --> 00:02:59.840
Well, I think what you described uh will resonate a lot uh with many data engineers.
00:02:59.919 --> 00:03:05.199
I think like they will all uh they will all uh feel that they are doing exactly what you are doing.
00:03:05.439 --> 00:03:12.560
Um so can you actually uh just out of curiosity, uh, because you just said that Airflow is uh is uh crucial.
00:03:12.639 --> 00:03:17.759
Um but uh when you joined your company, Airflow was already there?
00:03:18.639 --> 00:03:20.800
Uh yeah, uh I think so.
00:03:21.120 --> 00:03:21.919
Oh okay, okay.
00:03:22.000 --> 00:03:23.680
So actually, I'm just gonna keep that.
00:03:23.759 --> 00:03:25.039
I I have a question for you.
00:03:25.120 --> 00:03:32.080
But uh first thing first, uh can you tell us a little bit more about the data stack that you have today and uh where does Airflow fit in?
00:03:32.560 --> 00:03:32.879
Okay.
00:03:33.199 --> 00:03:37.039
On the current project, uh our stack has several main components.
00:03:37.199 --> 00:03:40.080
Uh we use Snowflake as a data source.
00:03:40.159 --> 00:03:48.479
Uh then we load and organize data in MongoDB uh using medallion architecture, brose, silver, and gold layers.
00:03:48.639 --> 00:04:06.879
Uh we use Spark for transformations and processing, uh, Airflow for orchestration, uh for monitoring and error handling also, and uh Elasticsearch for storing uh reports that we are used for application later.
00:04:07.120 --> 00:04:11.759
And uh Airflow is basically uh, as I said, control uh center of the system.
00:04:11.919 --> 00:04:24.319
It decides when each ETL process starts, in which order task runs, um, uh what happens if uh something fails, and when notifications or alerts should be sent.
00:04:24.560 --> 00:04:28.240
And without Airflow, the pipeline would be very hard to control.
00:04:28.560 --> 00:04:32.399
And because we had many, many dependencies between processes.
00:04:32.560 --> 00:04:35.839
And it's not just uh run a spark job.
00:04:35.920 --> 00:04:39.199
Uh the whole flow has to be happened in the correct order.
00:04:40.079 --> 00:04:52.319
Okay, well, actually that that that creates a bridge to the my next my next question, um, which is you just mentioned that that you are using Airfo for orchestrating your ETL um jobs, maybe more than just ETLs.
00:04:52.399 --> 00:04:59.199
Do you have like any like maybe uh other, you know, um like maybe machine learning use cases, things like that also?
00:04:59.279 --> 00:05:03.519
Or is it just like uh no this no, just ETL in this project, yeah.
00:05:04.160 --> 00:05:04.319
Okay.
00:05:04.480 --> 00:05:13.360
And you you just mentioned that dependencies, I mean, uh you definitely have to execute things in the right order, um, which is definitely what Airflow is made for.
00:05:13.680 --> 00:05:20.480
But uh can you tell us a little bit more about you know the ETL uh processes that you have today?
00:05:20.639 --> 00:05:27.600
Um and you know why do you actually require this sequence order that it is you know so important for you?
00:05:27.759 --> 00:05:32.959
Because I think you also have a way to do uh you know uh things that are interesting.
00:05:33.360 --> 00:05:47.040
Uh yeah, this was actually one of the points where things become really interesting and honestly a bit uh painful at the beginning, uh, because at first we tried to uh solve it using the standard airflow options.
00:05:47.120 --> 00:05:56.959
Uh we looked at configurations, uh dog run limits, uh we uh tried to use pools, different uh dependency setups, and so on.
00:05:57.199 --> 00:06:05.360
Basically everything that seems uh like it could help, but uh every solution uh solved just one part of the problem that we had.
00:06:05.519 --> 00:06:19.279
And at the same point, in some point we were stuck and uh not completely desperate, but uh close because we really realized that the problem was bigger than just how we uh stopped two DUG from running at the same time.
00:06:19.519 --> 00:06:30.959
And the real question was here uh how do we control the full execution flow across multiple DUGs, especially when some of them can be triggered at any uh moment from the application side?
00:06:31.360 --> 00:06:51.759
And uh we had to take a step back, sit down, and literally draw the whole process uh end-to-end, uh, which DAGs uh uh which DAGs uh exist, uh, what they um touch, which collection uh they use, which layers they write to, and what could go wrong if two processes uh overlap.
00:06:52.240 --> 00:06:58.399
And the main reason for strict sequential execution was that uh for uh was data consistency.
00:06:58.639 --> 00:07:04.879
Uh all of these DUGs were used to the same MongoDB collections at the same layers.
00:07:05.040 --> 00:07:10.160
And in the bronze layer, uh data could be overwritten by a new snapshot.
00:07:10.319 --> 00:07:13.279
In silver, we had cleaned and prepared data.
00:07:13.519 --> 00:07:19.360
In gold, we had aggregated data used by the applications, uh application reports.
00:07:19.600 --> 00:07:33.600
And so if the two DUs run in the same time and change the same collection or layer in layers, we could uh easily end up with collision uh in uh inconsistency data or partially processed uh results.
00:07:33.759 --> 00:07:41.680
And that's why uh the sequential order uh execution was not just nice to have, it was absolutely necessary for our case.
00:07:43.040 --> 00:07:43.439
Wow.
00:07:43.600 --> 00:07:47.040
So now I I need to learn more about like how did you solve that?
00:07:47.120 --> 00:07:53.199
Um because I'm I'm sure that it's a very, very common issue uh you know in the in the like for many engineers.
00:07:53.279 --> 00:07:55.680
Like how did you what did you try?
00:07:56.319 --> 00:08:02.639
Oh uh before building a custom solution, uh we first tried to use uh waterflow already have.
00:08:02.959 --> 00:08:10.079
Uh we started with the Dug level uh configuration, of course, like limiting the number of uh active DAG run.
00:08:10.480 --> 00:08:16.399
Uh that helped a little because the same DUG could not run multiple times in parallel.
00:08:16.560 --> 00:08:20.879
That was great, but it didn't uh solve the biggest problem here.
00:08:21.040 --> 00:08:23.439
Different DAGs could still run at the same time.
00:08:23.680 --> 00:08:26.399
And then we uh looked at pools also.
00:08:26.720 --> 00:08:31.920
Pools are useful when you have to limit some many tasks, but in our case uh it wasn't a problem.
00:08:32.320 --> 00:08:43.279
And uh uh our problem was uh on a higher level, we needed to control the execution order uh of entire DAGs, not just individually task or individually DAG.
00:08:43.679 --> 00:08:53.279
And the biggest challenge, yeah, can you No, I was I was about to say so, yeah, it's it's really like you you actually want to have multiple DAGs that need to run one after the other.
00:08:53.440 --> 00:09:02.559
It's not because, like you know, for people that that are uh wondering about pools, uh pools allow you to um you know control the the parallelism of your tasks.
00:09:02.720 --> 00:09:08.399
So basically, if you want to execute one task after another, you can do that with pool and then you set one slot to that pool.
00:09:08.799 --> 00:09:21.919
But you will have to attach uh all the tasks to that pool, but still it doesn't solve truly the the issue that uh now you have Ivana uh with the the DAGs that should not run, multiple DAGs that should not run at the same time.
00:09:22.879 --> 00:09:23.679
Yeah, sorry.
00:09:24.559 --> 00:09:26.559
Yeah, it was the problem here.
00:09:26.720 --> 00:09:34.639
And the biggest challenge, as I said, uh was that the some DAGs are scheduled and uh the other DUG was even uh event-driven.
00:09:34.799 --> 00:09:39.600
And uh that event-driven DUG uh could be triggered at any moment from the application.
00:09:39.679 --> 00:09:53.600
And that's why we need to have uh control Q uh and uh uh and we need to have uh uh the right order when uh one DUG uh runs and so on.
00:09:53.840 --> 00:09:56.960
So standard airflow configuration was not enough for us.
00:09:57.120 --> 00:10:03.519
We needed something uh like control Q, as I said, that could decide which process was allowed to run next.
00:10:03.679 --> 00:10:07.759
And that's why we built a custom orchestration layer for that.
00:10:08.639 --> 00:10:15.679
Okay, so tell me tell me more about this uh this um this uh layer that you that you built.
00:10:16.320 --> 00:10:20.399
Yeah, in the end we created a three-layer DAG architecture.
00:10:20.559 --> 00:10:24.320
Uh the first layer is the scheduler or trigger DAGs.
00:10:24.399 --> 00:10:27.440
Uh they are the entry point into the system.
00:10:27.600 --> 00:10:38.240
Uh some are scheduled, for example, for daily imports, uh, and some one uh some are event-driven, triggered by uh user applications uh from application side.
00:10:38.480 --> 00:10:43.519
And their job is not just to run the full uh ETL process directly.
00:10:43.679 --> 00:10:48.399
Their job is to register a request and say this process needs to be executed.
00:10:48.639 --> 00:10:52.559
And the next one, the second layer, is the proxy DAG.
00:10:52.720 --> 00:10:55.919
And this uh here we uh get the magic.
00:10:56.240 --> 00:11:04.080
This is the most important part of the solution because the proxy DUG controls the queue and decides which process can run next.
00:11:04.240 --> 00:11:15.600
Uh, it checks if there is already an active uh DUG, uh, what is waiting uh in the queue, what has priority, or whether the next process is allowed to start.
00:11:15.840 --> 00:11:23.759
And so instead of every DUG decide independently when to run, everything needs to go through the proxy layer.
00:11:23.919 --> 00:11:28.320
And the proxy layer is the main uh brain here.
00:11:28.720 --> 00:11:31.759
Uh the third layer is the main ETL DUGs.
00:11:31.919 --> 00:11:37.840
Um, these DUGs contain the actual logic, data import transformation, validation, and so on.
00:11:38.000 --> 00:11:53.039
And these design help us uh uh uh separate uh responsibilities, uh trigger DAG, as I said, initiate uh requests, the proxy controls uh execution order, and main DUGs go to the actual processing.
00:11:53.919 --> 00:12:00.159
And basically we solve uh our problem with this with this architecture.
00:12:00.799 --> 00:12:02.240
This is this is super interesting.
00:12:02.320 --> 00:12:03.759
Um I have a few questions.
00:12:03.919 --> 00:12:07.759
Um so the the um what exactly is this queue?
00:12:08.000 --> 00:12:10.159
Like uh technically speaking, what is this queue?
00:12:10.480 --> 00:12:12.240
Is it like uh Kafka queue?
00:12:12.399 --> 00:12:13.600
Is it like like what is it?
00:12:14.240 --> 00:12:15.679
It's uh DAG queues.
00:12:16.320 --> 00:12:17.759
Okay, okay, okay, I see.
00:12:17.919 --> 00:12:24.000
Um and uh uh so those those layers are they like how they are built?
00:12:24.159 --> 00:12:27.279
Are they like all like like what is exactly a layer?
00:12:27.360 --> 00:12:28.399
Is it like a DAG?
00:12:29.360 --> 00:12:31.360
Yeah, yeah, everything yeah, it's DAG.
00:12:33.200 --> 00:12:33.840
Yes.
00:12:34.639 --> 00:12:44.240
We have uh scheduler, we have several scheduler DAGs, and we have just one proxy DAG, and we have several uh uh main DAGs.
00:12:44.799 --> 00:12:45.679
Okay, I see, I see.
00:12:45.759 --> 00:12:55.440
Because like I'm I'm just wondering because you know some some people may think that it's it might be like uh you know a plugin that you build or uh you know a Python scripts uh Python script that you run.
00:12:55.759 --> 00:13:12.960
I think like you know, I mean we we also uh we also talked a little bit more about I mean previously uh before this uh this episode uh we discussed a little bit more about how you you know how your um your DAGs are uh um uh written and you know how they are generated and so on.
00:13:13.039 --> 00:13:16.320
Uh I think you have some DAGs that have a lot of tasks.
00:13:16.960 --> 00:13:21.440
Um and I was just wondering like how do you handle that?
00:13:21.600 --> 00:13:24.799
Like how do you handle the creation of your DAGs, I would say?
00:13:24.960 --> 00:13:29.519
Uh do you use like you know, uh a Python script to generate some of your DAGs?
00:13:29.679 --> 00:13:31.039
Do you have like some templates?
00:13:31.279 --> 00:13:33.759
Do you have you know dynamic loops, whatever?
00:13:33.840 --> 00:13:37.039
Like can you tell us a little bit more of how you author those DAGs?
00:13:37.200 --> 00:13:46.480
Because I imagine that you have all of those layers in place, I mean those three layers in place, but I imagine that your DAGs are really standardized as well, so that they can actually run.
00:13:46.960 --> 00:13:47.679
Yeah, of course.
00:13:47.840 --> 00:13:57.360
Uh we have uh DAGs with uh approximately 200 tasks, and of course, we definitely did not write uh 200 tasks manually.
00:13:57.519 --> 00:13:59.679
Uh that will be impossible to maintain.
00:13:59.840 --> 00:14:05.279
And the idea was to make the DAG structure usable and configuration driven.
00:14:05.519 --> 00:14:09.840
And each main DAG has several logical uh sections here.
00:14:10.080 --> 00:14:22.000
On the top level, it looks like uh it looks like uh much simpler, maybe around 30 or 40 main steps, but inside those steps we use task groups.
00:14:22.320 --> 00:14:27.519
And it's in uh inside of task groups, we have more task groups.
00:14:27.679 --> 00:14:31.840
And on that way, we create dynamically generated tasks here.
00:14:32.000 --> 00:14:37.200
Uh uh when everything is expanded, we get around 200 tasks per DAG.
00:14:37.279 --> 00:14:37.840
Yeah.
00:14:38.159 --> 00:14:44.399
And also we use configuration file, define what uh changed between different DAG versions.
00:14:44.559 --> 00:14:52.879
For example, which collections are used, which steps are active, uh, which parameters are needed, uh, how monitoring works, and so on.
00:14:53.039 --> 00:14:56.399
Uh, also versioning uh was very important for us.
00:14:56.559 --> 00:15:00.799
And after every release or hotfix, we increase the DAG version.
00:15:00.879 --> 00:15:14.320
And DAG version, that number is related with the release number, and uh that helps us uh keep a clean and controlled pipeline without new logic uh affecting already running processes.
00:15:14.720 --> 00:15:17.120
And we solve that on that way.
00:15:17.279 --> 00:15:21.279
And with large DUGs like this, uh monitoring is crucial.
00:15:21.519 --> 00:15:26.799
It's not enough to know uh what some uh uh that something uh failed.
00:15:26.960 --> 00:15:34.480
You need to know uh where it failed and why it failed and uh whether you can safely restart only that part.
00:15:34.799 --> 00:15:37.279
That was the logic behind all this.
00:15:38.080 --> 00:15:46.720
This episode is brought to you by Astronomer, the team behind Astro, a managed airflow platform built for data engineering teams.
00:15:46.879 --> 00:15:50.879
From batch pipelines to training models and wiring up agents.
00:15:51.039 --> 00:15:57.120
Astro handles the infrastructure, upgrades, and scaling so you can focus on building.
00:15:57.279 --> 00:15:59.519
Check it out at astronomer.io.
00:16:01.200 --> 00:16:06.240
Okay, um, and so this this uh those configuration files are they like YAML files?
00:16:06.559 --> 00:16:09.360
Uh yeah, I use YAML file for the configuration file.
00:16:09.840 --> 00:16:10.000
Great.
00:16:10.080 --> 00:16:14.559
So you actually mentioned task groups, um, which I think is uh definitely an interesting feature.
00:16:14.639 --> 00:16:36.960
It's a simple one, but it's uh actually a very useful one because to your point, when you have like hundreds of tasks, uh sometimes thousands of tasks, uh grouping your tasks together might be a good idea, um especially if you don't want like to actually almost crash the UI, uh or at least it's more it you won't be able to navigate through your DAG because there there will be way too many tasks.
00:16:37.440 --> 00:16:42.720
Um but before task groups you were not using them, obviously.
00:16:42.799 --> 00:16:47.200
You were some using something different, which I think uh was the sub DAGs.
00:16:47.279 --> 00:16:49.120
Uh and subdags were very popular.
00:16:49.919 --> 00:16:55.919
Can you tell us a little bit more of how how did you migrate from the sub DAGs to the task groups?
00:16:56.080 --> 00:17:06.799
Um, because task groups are easier to use than sub DAGs, but you know, I think like many people actually had to migrate, and you are the first person that finally will tell me how they did that migration.
00:17:07.279 --> 00:17:08.559
So yeah, yeah.
00:17:08.799 --> 00:17:18.799
Uh subdags were useful for us because they allowed us to split one large ETL process, one large uh DAG into logical parts, as I said before.
00:17:18.960 --> 00:17:24.559
For example, one part could be import data, another transformations, another validation, and so on.
00:17:24.799 --> 00:17:30.720
If only one part failed, we could retry just part instead of rerunning the whole DUG.
00:17:31.039 --> 00:17:32.960
That was the point here.
00:17:33.279 --> 00:17:37.039
And when subdugs became deprecated, we had two main options.
00:17:37.200 --> 00:17:41.599
The first one was to replace each subdugs with a separate DAG.
00:17:41.759 --> 00:17:44.880
Uh, from retry perspective, uh, that made sense.
00:17:45.119 --> 00:17:50.079
But uh each logical part could be triggered or retried independently.
00:17:50.240 --> 00:18:04.400
And however, for our case, we would lose the visibility of the full pipeline in one place and introduce more complexity around uh dependencies, trigger triggering logic, uh status tracking, error handling, and so on.
00:18:04.640 --> 00:18:14.480
Since we already had strict functional execution uh requirements, splitting everything into many separate DAGs would make the system harder to monitor and control.
00:18:14.960 --> 00:18:19.839
And that was the main reason why we moved to task groups.
00:18:20.079 --> 00:18:30.799
And task groups uh are better for organizing uh tasks, uh visualization inside one DAG, but they do not have behavior exactly like sub DAGs.
00:18:30.880 --> 00:18:32.720
And that was the first challenge here.
00:18:33.039 --> 00:18:40.240
So the main thing we lost was that availability to retry a whole group as one logical unit.
00:18:40.559 --> 00:18:45.119
So we built custom retryal logic on the top of task groups.
00:18:45.359 --> 00:18:53.279
In Airflow 2.9, uh, we solved this through the metadata uh data DB uh based on the task group identifier.
00:18:53.359 --> 00:18:57.599
Uh we knew which task instance belongs to that group.
00:18:57.839 --> 00:19:08.799
So we needed to retry the group, uh, our logic uh restart the state of the those um task instances so Airflow could run that part again.
00:19:09.039 --> 00:19:16.559
Uh the important thing uh uh is that uh this uh this was not a random manually database update.
00:19:16.720 --> 00:19:20.000
It was control logic inside our orchestration layer.
00:19:20.640 --> 00:19:23.920
So that gave us the same practical benefit as before.
00:19:24.079 --> 00:19:30.160
Now we had uh now we have uh task groups, but with a retry mechanism as subdag.
00:19:30.960 --> 00:19:32.000
Which is amazing, right?
00:19:32.079 --> 00:19:48.960
I mean I think like many many people actually had this issue when task groups um you know um uh came to uh I don't remember the the Airflow uh version, but when they actually replaced the sub DAGs, like people were like, okay, this is great, but also like now I cannot retry my task group.
00:19:49.039 --> 00:19:51.039
I have to retry like each task one by one.
00:19:51.119 --> 00:19:52.160
I I don't want to do that.
00:19:52.240 --> 00:19:56.480
And so I think like many people actually did the exact same solution that you just described.
00:19:57.279 --> 00:20:01.200
But in Airflow 3, I believe this is not possible anymore, right?
00:20:02.240 --> 00:20:02.880
Yeah, of course.
00:20:03.759 --> 00:20:08.160
How did you solve again this uh this you know with with Airflow 3?
00:20:08.559 --> 00:20:10.480
Again, unexpected behavior.
00:20:10.720 --> 00:20:23.200
Uh with the Airflow 3, we had to change this approach because relying directly on the metadata dB was no longer the right direction because of the new architecture, Airflow architecture.
00:20:23.359 --> 00:20:27.680
So we moved to uh the retry logic to the Airflow REST API.
00:20:27.920 --> 00:20:40.319
Now instead of directly changing the metadata dbs, db tables, uh we communicate with the airflow through the officially uh official uh APIs.
00:20:40.559 --> 00:20:42.960
Uh now we need to control a dog run.
00:20:43.359 --> 00:20:52.640
When we need to control DAG run uh task instance or anything else, uh or retry behavior, uh we do that through uh supported endpoints.
00:20:52.799 --> 00:20:55.359
And this is better for several reasons.
00:20:55.599 --> 00:21:10.880
The first one is more uh stable across Airflow versions, because uh if uh you rely on internal database table, something can change after an upgrade, as we had um experienced in previous uh versions, the Airflow versions.
00:21:11.119 --> 00:21:13.839
And the second, uh, it's more secure.
00:21:14.000 --> 00:21:29.599
Uh also I think that uh the architecture is clear uh clearer now, and Airflow remains responsible for its own uh internal uh state, and we interact with with uh with it through the official uh API layer.
00:21:29.759 --> 00:21:37.039
Uh so finally it is easier to maintain and uh maintain and explain uh to the new team member uh also.
00:21:37.440 --> 00:21:45.599
So in Airflow 2.9, our metadata db solution worked, but with Airflow 3, we modernize it basically.
00:21:46.720 --> 00:21:54.799
Yeah, and I think this is um, you know, like for people that are wondering, so in Airflow 3 now you cannot access the Airflow metadatabase directly anymore.
00:21:55.039 --> 00:21:57.519
Um for a good reason, like for security reason.
00:21:57.599 --> 00:22:02.960
Like honestly, uh I mean it was good, it was good for us, but it was also a little bit hacky, right?
00:22:03.039 --> 00:22:09.920
You you should not uh you should not expect that your DAGs could access the Airflow meta database, so that's why this has been changed.
00:22:10.160 --> 00:22:19.119
Um but it's to your point, it's way more elegant uh to use the the the Airflow Rest API, uh especially as in Airflow 3, I think you know.
00:22:19.279 --> 00:22:26.079
I mean basically they keep adding new endpoints so you can do more and more things with the Airflow SPI, it's much better than it used to be.
00:22:26.640 --> 00:22:30.559
Um so I'm I'm glad that you overcame this issue again.
00:22:30.799 --> 00:22:35.759
I hope that in Airflow 4 you won't have to come up with uh something else.
00:22:36.000 --> 00:22:38.400
Um awesome, thank you so much, Ivana.
00:22:38.720 --> 00:22:39.759
Uh time flies.
00:22:39.920 --> 00:22:46.960
We are already uh I think it's already uh already uh three 30 minutes.
00:22:48.079 --> 00:22:54.079
I have to ask you first, um what is the best way for folks to get in touch with you?
00:22:54.240 --> 00:23:01.599
You know, if they have any questions regarding your solution and again like all of the common issues that you uh actually um solved.
00:23:02.480 --> 00:23:03.119
Okay, thank you.
00:23:03.279 --> 00:23:04.960
First, uh thank you for having me.
00:23:05.119 --> 00:23:07.279
It was really great talking about this.
00:23:07.519 --> 00:23:11.519
And the best way to get in touch with me is through LinkedIn, of course.
00:23:11.680 --> 00:23:16.720
And I'm I'm always happy to connect with the people uh with the same topics and so on.
00:23:17.039 --> 00:23:19.039
So I want to add something.
00:23:19.200 --> 00:23:29.440
Uh I think these kind of challenges are very interesting uh because they show the orchestration is not just only for scheduling jobs, and uh, Airflow is not just for that.
00:23:29.599 --> 00:23:38.160
Uh there are always unexpected situations where you have to learn something new and you know you need to adapt and find a practical solution.
00:23:38.319 --> 00:23:43.119
So for me, uh that part of the work uh was really interesting.
00:23:43.359 --> 00:23:47.519
And yeah, uh everyone can feel free to reach out on LinkedIn.
00:23:47.680 --> 00:23:49.599
I'll be happy to connect.
00:23:50.240 --> 00:23:50.880
That's awesome.
00:23:50.960 --> 00:23:55.519
Well, first thing first, uh Ivana, I have to thank you for uh for being here.
00:23:55.680 --> 00:23:58.400
Uh so thank you so much again for joining this episode.
00:23:58.480 --> 00:23:59.440
It was a wonderful one.
00:23:59.519 --> 00:24:02.160
I'm very happy uh about everything you said.
00:24:02.319 --> 00:24:05.680
I think like people will enjoy all the insights that you just gave.
00:24:05.920 --> 00:24:08.720
Um and well, I hope to see you again for another one.
00:24:08.880 --> 00:24:09.519
Who knows?
00:24:09.680 --> 00:24:11.599
But otherwise, I wish you a wonderful day.
00:24:11.759 --> 00:24:13.359
Take care and soon.
00:24:14.400 --> 00:24:17.359
Thank you for listening to the data flowcast.
00:24:17.599 --> 00:24:21.039
Check the show notes for links to everything mentioned today.
00:24:21.200 --> 00:24:25.680
If you're enjoying the podcast, subscribe so you never miss an episode.
00:24:26.000 --> 00:24:28.160
And leave us a five-star review.
