For the next session, we're coming up to the end of the day now. I'm delighted to have
Ben Talis and Balza Ribero, fantastic name, Shaps. Who are you talking about enhancing
air, global analytics, data engineering, ML and Canadian. So take it away guys. Thank you.
Before we start, I just want to make sure that you can hear me at the back and on the stream.
Okay, can you hear me at the back if I speak this loudly?
Okay, so we'll speak loud.
Okay, thanks very much. So yeah, just to explain myself, I'm Ben, the spouters are we're both SREs
at the Wikimedia Foundation and we're in the data platform engineering team.
And yeah, so this presentation is an implementation story about air flow really.
And it should hopefully be interesting because of the fact that it's the Wikimedia Foundation,
it means that all of the code that's there, all the tags, all the puppet manifest,
everything else, all the tickets around implementation, they're all open. So it should be an interesting sort of reference case
and see where we're up to at the moment.
Yeah, so seven to see, Wikimedia Foundation, I don't know how familiar you all are with it,
but there are some interesting aspects of working at Wikimedia Foundation.
So firstly, we're all on prem, we're all on premists in physical data centers.
So no cloud services, no AWS, no commercial stuff, we're all fully open source.
And we don't allow for policy reasons any running of container images that are made by third party sources.
So everything has to be on dead end, everything has to be built on a gold standard image that we build ourselves.
So that just means that when you're implementing software like this or systems like this,
there are additional steps because we have to take all the sources from all the code from sources and build them ourselves.
Okay, this is the sort of raise on debtor of the data platform.
So there's sort of a mission statement there.
Effectively, we provide a data platform which has various storage and transformation and serving layers.
And we have various teams that we support, so from data engineering and analytics.
Things to do with the experimentation platform, we're building at the moment, the search platform.
So that's everything to do with searching on MediaWiki projects and also query services.
So all of the data query services, this kind of thing that they're all falling under the data platform.
Okay, this is the best sort of reference we have at the moment for what the whole data platform looks like.
I'm not going to go through it in any detail, but we've got like the event producer and things.
So everything like the web request logs that go into it.
Then we have various sort of storage streaming data lake kind of areas here and then various serving systems serving layers out here.
And you find a nice,
super set dashboard here.
Please do that's linked up to Presto and do it and various things.
But we're talking at the moment here.
So we're talking about batch job scheduling for workloads that are generally on the Hadoop cluster.
Let's move on.
So we've been using the Airflow for quite a long time since it was an incubator project.
And it's built up, we've built it up from a sort of MVP stage to where it has been beginning of 2024.
And we use it for all sorts of things.
So we've actually moved our Hadoop workloads from UZ into Airflow.
And by, whenever I was in 2023 maybe we've finished moving all of the UZ workloads into Airflow.
And these are our sort of primary use cases for researchers, for such product team analytics and so on.
Okay, and yeah, if we, this is a sort of screenshot of what a typical data pipeline in Airflow looks like.
And most of the time what we're doing is we're using sensors to say, you know,
when data arrives in HGFS or in various locations.
And then when it's there, we then run spark jobs, which you spark on yarn on the Hadoop cluster.
So Airflow workers themselves were never intended to do the bulk lifting at this time for analytics and for most of the other things.
Airflow workers launch yarn jobs on the Hadoop cluster.
So that's where we are. And Airflow is effectively orchestrating these spark jobs on the cluster.
This was the infrastructure as we came to it in sort of beginning of 2024.
So each of these instances for teams or for functions, we had an Airflow instance.
And they were running on either a bare metal host or individual virtual machines with a local web server, local scheduler.
And all of the Airflow jobs running as sub-processes of the Airflow scheduler here.
And they were connecting to a shared Postgres metadata server with a replica but knows sort of automatic high availability,
full-over mechanism.
Wikimedia in general is a MariaDB house.
So most of the effort that we have around high availability and, you know, database administration is in the MariaDB space.
Airflow doesn't work with MariaDB, so we've had to migrate to Postgres.
And we didn't have, you know, the huge amount of Postgres experience.
So we had this model at the moment and it had a number of shortcomings in terms of scalability.
I think that's one we've got to do.
Yeah, I think this is pretty much what we have covered it.
We're saying that it's all Kerbross authenticated, Hadoop, yarn, hive and spark.
Yeah, that's pretty much it. Also we had the local executor and local task logging.
So as the number of tasks in Airflow ramped up, the number of logs that we had on the individual machines was getting to be in the millions.
Okay, so this is then talking about some of the problems that we had with this infrastructure.
And it was really around about around what happened when we had to do maintenance of the metadata database,
rebooting, restarting that service.
We didn't have any high availability in Postgres.
That cause is downtime for the scheduler, which then causes problems for the jobs that are running as sub processes.
Sometimes they were would be okay, but sometimes they just sort of disappear a little bit.
So not great.
And yeah, we had a high number of connections to Postgres as well coming from there.
Actually as we got into dynamic bags that were doing more and more database querying.
And next one piece. Yep.
So yeah, we then took much scaling. How we go about scaling our Airflow instances and our Airflow things effectively.
Because we were linked to we were limited by a single machine on each instance.
It got pretty difficult to scale. We couldn't use the local executor in a way that was easy to scale.
Okay. Also, there were a few things around about the user experience that were pretty poor.
Engineers would have to use SSH tunnels in order to gain access to the Airflow web server.
So that's not very nice as a user experience to tell people you got to have a terminal window within SSH connection.
So you can log into your Airflow.
And also, if there were people who were not necessarily technical, but management wanted to see how things were going.
They would still have to have an SSH user shell posics account with all the various overhead that had to do.
And then because we had that, there was no authorization level.
We just bypassed all of the Airflow authorization of our back levels and this kind of thing.
Because everyone was already authenticated with SSH.
So these are some of the things that were not ideal about our current setup.
So we came up with a plan to make things better.
We've got the ticket links here, get to the things.
It's all open so you can all look at it and how we manage the project.
We wanted to make sure that we could eliminate all the single points of failure.
Make sure that the Airflow services are resilient.
And a key thing really is that we, as the SRE, although we're changing the deployment model,
we didn't want to have to force the DAG authors to change the way that they were doing their DAG.
So we didn't want to have to have a big bang migration for users.
It was still going to be supporting spark jobs on Yarn.
Okay, so in terms of the infrastructure now, the project infrastructure,
we have a Kubernetes cluster for the data platform engineering team.
So we built this over the last couple of years.
It's likely used at the moment so it's good deployment target for the compute.
So we've also got a couple of these machines here, these gates workers.
This one and this one have got GPUs in.
So all the generation and newer generation GPUs.
This is going to be useful for the research and ML team.
So we're going to be onboarding to do model training work.
And you go to the next one, please, yep.
We've also invested in safe storage in the DAG platform team alongside the Hadoop cluster.
So this is something that we really wanted to be able to use for this project to help us achieve this availability targets and things.
So we have in the safe cluster, five nodes.
We have a large tier with backed by disks and cache and then we have a smaller high performance SSD tier.
So this cluster again was up and running ready for use in this airflow project.
Okay, so the architecture that we have come up with is that we wanted to move from the local executor to the Kubernetes executor for pods.
So I don't know if you are familiar with airflow, but every time a task runs with the Kubernetes executor it will run as a Kubernetes pod, run to completion.
So it will no longer be running as sub-processes of the scheduler.
We also wanted to make sure that airflow stores its logs remotely rather than local to the scheduler works running previously.
And key part of it is that we wanted high availability postgres built in or in some way.
So on the postgres side we decided to take the decision to move postgres into Kubernetes, backed by the safe cluster.
And we have a Kubernetes operator for postgres clusters.
That enables automatic leader, failover, sorting out replication and postgres poolers and this kind of thing.
And we wanted to make sure that we have got competitive postgres backups at the same time.
And the safe cluster we have got the blocked device interface, the SFFFS, POSIX file system interface and the S3 interface.
I don't know if everyone knows with SFF as well, but effectively we are using the three main interfaces of the safe cluster in this project.
Okay, yep. This is then the architect he do want to take over from here.
So this is the visual representation of what Ben was just talking about.
The blue boxes here are the air flow components.
We are like we mentioned before. We're using Kerberos, so we need some way to renew our tokens regularly.
So this is the Kerberos subcommend that's running.
We have the webserver, obviously the scheduler that itself manages the pods running on Kubernetes.
And the way that we've designed the PG cluster is that it automatically connects to PG balancer and not to post risk itself.
We actually set up some natural policies preventing anyone from connecting to POSIX girl.
We enforce traffic to go through PG balancer to avoid overwhelming PG itself with connections, especially when we run in the thousands of map tasks.
So we can get pretty here. And like Ben said, all of that is backed by SFF.
We had to expose different interfaces. So S3 is used to store the logs and the walls and the base backups for the POSIX girl.
We also needed a block device, which actually stores the actual PG data.
And we also needed some distributed file system in which we store the DAX themselves that we mount on the scheduler.
And we also store the Kerberos tokens that we regularly renew and we store on every single airflow component that needs to talk to Kerberos.
So the first thing that we kind of set out to do, and I say the first thing we actually worked on this in parallel, but it makes for a better story,
is that we integrated SFF with Kubernetes. So if you're not familiar with Kubernetes, the way that we do it,
that you integrate storage with it, is through what is called the CSI, a container storage interface.
And so this is the man in the middle, so we speak that we'll see that you are requesting some volume to be created in your storage mechanism,
the storage software, and we'll then actually provide the volume for you.
And so for us, we actually had to install two CSIs, one for the block device, the RPD and one for SFFFS.
And we also expose an any cast endpoint for S3, so basically it allows any clients who wish to talk to S3,
to talk to the SFF server that is if available in the same network racks, so that we minimize the number of network hops,
and we maximize bandwidth. And so what you're seeing here in the bottom screen chat is basically just the persistent volume claims,
and so you can see the airflow dags that is stored on SFFFS, the curb loss volume on SFFFS as well,
and then the PG data on our block device, and you can see in the names that all of these data are actually backed by the SSD pool that we talked about,
because these are actually latency critical.
We investigated multiple operators to run in, to run cross-scrisco in Kubernetes.
I have a background of running data stores in Kubernetes, thanks so much, and so I know that this is actually pretty hairy, and you can be pretty dangerous.
And so we have the stick at, if you want to have a look at that, we investigated nothing five of them,
and we settled on one called cloud native PG. It had basically the features that we wanted, it had support for pooling, automatic backup,
and it failed over data imports, and also exposed all of the PG metrics to premieres.
Some of the operators we investigated had proprietary metrics interface, and so that just couldn't fly.
And so the way that we designed the cluster, we have two PG servers and three PG bansers, the right ahead logs,
and the base backup, so we exported to the S3 interface of SFF, and the backups.
If you think about it, it's kind of dangerous, because the PG data is stored in SFF, and then we export it back up in SFF as well.
So we export trade them outside of SFF every night as well, so that we don't store every single X in this Mbaskar.
As you can see on that little screenshot, there we have a, we can use QCTL to inquire about the state of the cluster, because it's a custom resource that it's integrated.
I don't know exactly why they can name it cluster, because it's not super explicit, but for now.
When it comes to security and airbag, what we've done is we wanted to have a public UI.
We didn't want anyone to have to go through a stage tunneling, and we wanted that UI to be authenticated.
Using OIDC, and basically what that meant is that when you logged in, your LDAP membership would be mapped to roles and thus to permissions.
So for example, event on myself, belonging to the SRE group, so we're admin on every single Airflow instance.
But Gabriela here is a member of the search team, and so he is getting some search related permissions, but doesn't have permissions on other influence instance.
So we have some flexibility that wasn't really possible for us before.
And we also wanted to have the CURBeras, the API CURBeras authenticated, which turned out to be a pain in the book.
And the migration plan was simple, and the reality wasn't.
First, we would migrate the web server.
The reason for the web server migration would be that we actually provide value to our customers, so to speak, by having a public UI, a public URL, or back in permissions.
Then we also created an Airflow instance for ourselves, and we would test every single kind of operator we could find in our Airflow DAGs repository, until we were confident that we could accommodate every DAG that we had in our codebase.
So we thought, then we would migrate for one instance, it's database to CURBeras, sorry.
Then we would migrate the scheduler, and then everything would explode, and then we would have to fix that years and so we're all improved.
And so we're telling this story when we applied for a fast end, we hoped that it would be done.
We are mid-biagration, and it's still a bit of an issue right now.
But we'll get there.
Some things did help us back, did hold us back.
We found that out of the box, the CURBeras API authentication was broken, and it took a little bit of fidgetting to fix it.
And the fix has been released in that pull request, and fixed it in that pull request and release in the fat provider, one that five to one.
It turns out the fun part of it is that I found some finger, I basically found some traces from the next building of mine, from a previous company, so I was able to reach out and we kind of talked about that.
But then, by default, even if you had a valid CURBeras token, the API would just respond for one.
I also found out that the automatically generated API client doesn't support CURBeras authentication.
So if you use CURBeras authentication, you're kind of on your own to use the client to talk to the API.
So we had to patch it in an ugly and sweet possible.
And the issue at working at Dibellium F is that my ugly hacks are probably, you can check them out.
And I think it will take a little bit of talking about it upstream, and so we can release that as a proper batch because right now this is shameful.
And like I said, migrating an instance with a lot of dice, was tricky.
Our biggest instance has about 150, and some of them have many, many, many, many tasks.
And so we found out that some of them actually needed to be rewritten just a little bit to account for the specific environments in which they were running.
And the one thing that was the trickiest to get right was a networking.
Because who is familiar with CURBeras networking?
You were lucky, you were lucky, yes.
It's not a great place to be, and it's complicated.
And we found out a lot of discrepancies between the previous environments in which we were running and the new one.
So we also run a service mesh that allows you to talk to local hosts and certain ports.
And we redirected to a service within the internal set of services that we keep media.
And integrating with that in a way that worked, took us some time as well.
So all of these little things were paper cuts.
And I think we're getting to the bottom of them, but you know, well you never know.
What do we get out of it today?
I think we've fixed most of the issues.
If not all of the issues that we mentioned, we have a better reliability story.
And rolling a restart set for we can rolling a restart crew in the cover that is workers.
And the air flow tasks will not be impacted because they run in pods.
And the worst thing that could happen is that they get rescheduled.
We can restart the schedulers without killing the pods.
We can, you know, we can do a lot of things that weren't possible before.
The Postgres QL cluster can be rolling restarted.
We have automatic fell over when it used to be a manual process.
Every air flow instance has its own PG cluster.
So there's no more noisy neighbor effect that we actually experienced mid migration at some point.
And we couldn't migrate because one air flow instance was hammering the PG cluster.
And so that was a little bit of a fun story.
We have an improved security story.
We have permissions in our back and logging.
And not everyone needs to be admin anymore, which is great.
The API isn't free for all anymore.
It's for us authenticated.
The scalability story is actually just easier because of Kubernetes.
We can scale with the number of Kubernetes workers that we have.
We are not constrained by the size of one machine.
So some, that's actually a little bit of a lie in the sense that we a lot of the air flow instances.
We're running on VM.
So we could resize them on the fly.
But for the bare metal instance that we have that was running the most number of jobs.
That was just the machine that we had for five years.
The UX is better.
When you're received an alert email by air flow that will take you to local host.
It was kind of a pain because you needed to run an HH cell to get your air flow.
So now you can click on the link and you get to air flow analytics.
We can be dead or something like this.
And it's just a better UX story.
And I think what I'm the most excited about.
And I think Ben is as well is that we were able to create and provide reusable building blocks.
We have some flink pipelines that Gabrielle is actually working on.
That are storing data in S3.
We have teams that are interested in getting a PG cluster that wasn't particularly easy before
WMF.
So this is getting easier now.
And the adoption, the increased adoption story is interesting.
We actually didn't know how to write DAX before.
And so we ended up creating our own instance.
And we kind of trained at air flow DAG authoring.
And then we ended up writing DAGs to administrate, to handle operations on that instance.
So we have a DAG that would just remove all files from S3 or clean up the database.
So things like this.
And it got us into a position where we are actually more equipped to talk to the data engineering teams.
Because we share more of their vocabulary that we speak.
And the final slide is now that we're there.
So we speak.
How do you have a question, sir?
Not like, yeah.
What we'll invest on in the near future when we're done with the migration.
We have weird mechanisms at the moment, allowing DAG authors to run ad hoc jobs.
And we can unravel, we can remove all of these weird things.
It also relies on a piece of software that is no longer main thing that we'd like to get rid of.
And so instead of bundling that ad hoc environment, we can allow DAG authors to just run pause themselves.
So they would run a task that itself would talk to Kubernetes to create an ad hoc job running in Kubernetes itself using the Kubernetes pod operator.
I don't know if any of you, or any of you, is it familiar with the Wikipedia dumps?
But it's a regularly updated data set that you could download yourself for directly or appear to peer.
And it's basically the entire content of being a Wikipedia or the different wikis.
And we have a nightmareish system.
Let's call it this way.
That does that.
And we are on the hook for it.
And we are trying to migrate it using these Kubernetes pod operator to airflow.
Because the fact that it would run on airflow would give us some more observability and re-try mechanisms to actually make it easier to manage.
And finally, we have, I think I've reworked the slides, but then I might have not saved them.
We don't only have, we have the machine learning team who is interesting in running airflow,
but we also heard from other SNR teams who would like to have quite complex operations.
The more that we, the easier we make it to get a new instance and to run it and to manage it, the more we found that teams were asking us about airflow.
And could I get an instance and how can I get running and things like this?
And so it seems to us that it's becoming part of the lingua franca of Wikimedia.
And I think on that, thank you for your time.
And if you have any questions, I think we have just a little bit of time.
Four minutes.
And there was a question over there.
That's right, so I'll repeat the question.
The question was, for the postgres, is the data actually stored on the network drives,
or is it local to the Kubernetes workers?
At the moment, it is stored on-saf in the RBD using the RBD applications.
So they are block devices from-saf that are provided to the pods so that the pods can,
the rapid Jason, so the very low latency speeds.
We are aware that there could be a right amplification problem.
So if we write a lot of stuff to the postgres database, it's writing it to two database instances,
which themselves have three replicas of the copy.
So, you know, we need to make sure that what we don't do is overuse the network resource
to try to give us better postgres availability.
But we're aware of that situation, so we're watching it.
And if that happens in future, then we may move those postgres databases to the Kubernetes
workers themselves, for example.
So the question was, have we considered migrating Spark to Kubernetes?
Yes.
I actually had it on the slides before it, and I removed it.
But it's a fair point.
A lot of what Airflow is doing is, like we mentioned, is orchestrating Spark jobs.
Right now, we have an enormous yarn cluster and a small, pretty small Kubernetes cluster.
The more we can migrate to Hadoop workers, as Kubernetes workers,
I think the best and the better will be, because we only will have a single orchestration layer.
And this is something that we can do now that we couldn't do before.
We could use the Spark Kubernetes executor, that instead of running the Spark driver on yarn
and the Spark application on yarn, because we could actually just run the Spark driver
and the application on Kubernetes itself using pods.
We haven't really tested it yet, but it's something that is on our minds.
Yeah, just to say on that point, we have at the moment much larger yarn cluster,
and we have Spark running with the yarn shuffler.
So the jobs themselves still run on yarn and make use of the fact that we have the data locality.
If we move Spark itself into Kubernetes, which we're enabled to do,
we have the Spark operator.
We would lose that data locality benefit.
So we're trying to get to the point where it's more flexible that certain Spark jobs
could run in Kubernetes, and certain ones could run on yarn.
And we're trying to make sure that Docker containers can be used throughout for that.
Yes.
Thanks for your time.
We are any insights for us,
regarding upgrading our flow, because it also has migration and stuff, I know.
Okay, so the question was, do we have any insights on how to upgrade our flow,
because there are certain migrations and things?
As it's got a flask core and the SQL alchemy,
the limbic.
A limbic.
The database migrations have not been a big problem for us.
I think we are pretty aggressive about upgrading.
I tend to, because we have a testing sense,
we can see pretty quickly whether it fails or not,
but whenever I see an announcement on the Airflow Slack channel,
I try to upgrade into one of those.
My experience is that it's been pretty successful,
but your mileage may vary on them.
And as I say, we are still in the throws of this big migration.
We had hoped that we have been able to come here and just talk about all the successes,
but we are still halfway through.
So we have one environment where we build our flow in a Debian package,
and we have another environment where we build it in the container.
We sometimes have problems keeping those two environments in sync,
and also we need to make sure that dark authors have an air flow development environment
that they can use and run their unit tests and this kind of thing.
So yeah, they are pitfalls, but in terms of, in terms of the actual production services upgrades,
we have not run into serious problems with any database migration.
I think we are all the time.
Thank you.