The Hyperscale Uncertainty Principle: Debugging Tail Latency in a Trillion-Object... Yashraj Kakkad

In this talk, Yashraj Gakkad, a software engineer at Google Photos, discusses the complexities of managing referential integrity at exabyte scale, focusing on the challenges of tail latency within a vast relational metadata database and object storage system. He highlights the intricacies of ensuring data consistency between two databases handling trillions of objects, stressing that even minor issues in read performance can disrupt the entire operation. Gakkad shares insights into the strategies employed to optimize performance, such as partitioning the user namespace and implementing parallel concurrency pipelines. Despite initial optimism regarding their design, the team faced unexpected challenges stemming from high variance in read stages. Ultimately, the talk delves into lessons learned about system architecture, over-sharding, and the balance between effective computational strategies and resource management.

My name is Yashraj Gakkad. I'm a software engineer at Google Photos. I'm going to talk about what happens when an elegant, perfectly balanced design collides with brutal physical reality of underlying systems and bare metal. So, let's talk about tail latency. So, at Google Photos, we operate at an exabyte scale footprint. We have trillions of objects in our systems. And this is massive scale even by our own

standards. So, at this scale, we are not just thinking about storage. We are looking at managing memory at a planetary scale. So, storing the bytes undoubtedly is a challenge. But, there is a lot more. The challenge that I'm going to talk about is that of referential integrity. So, we have a massive relational metadata database and this relational metadata database is perfectly in sync with an object storage

system. Of course, objects which are referenced by the metadata storage have to be And this metadata system has to operate as a source of truth. So, if an object is referenced by this metadata storage system, it has to exist in the object storage system. If it is if it does not exist, we have a problem. And vice versa as well. If an object is not referenced by

this metadata system, it's it's it's an object that we no longer need in our system. And and we we for various reasons need to like clear it off as soon as possible. yeah, I mean, this is like just a foreign key, but it's across two different databases and it's at trillions of objects of there is nothing out of the box which exists for something like this. And

to verify this consistency, um like basically doing both these things, like detecting data loss as well as doing garbage collection, uh we run a very huge concurrency pipeline. And this is kind of like the basis of most of this talk. So, yes, um I mean, we have a pipeline. Uh we need to verify trillions of objects. And for various reasons, we need to read uh the entirety

of the two data sources when we do this verification. So, yes, obviously Okay. Um so, yeah, obviously, we are going to need um massive massive amount of parallelism to do this. Uh so, not surprisingly, even the best batch processing frameworks that you see break at this scale because of like various issues in the underlying systems and things like that. I mean, things just keep breaking. Uh so,

what we did was we we designed what we thought is like the perfect blueprint for this kind of a problem. Um so, we just took our um user name space and we just like logically partitioned it into 16 equally sized partitions. And instead of running one pipeline, we we started to run 16 different parallel concurrency referential integrity pipelines. And yeah, they were just operating on isolated name

spaces because the user space was divided. They did not have to interfere with each other. Uh we ensured that our data model was aligned in a way that something like this was possible. Uh like doing scans on both the sides was not a problem. Uh and yeah, I think hunky-dory life was good. Like I mean, there is nothing magical about 16 here. It's just an empirical number

based on like our experience with like uh a trade-off between like number of partitions, maintaining our SLA, and things like that. So, yeah, I mean, we we thought this is good. This this gives us like a very nice um mathematical guarantee. We have a perfectly balanced work across 16 partitions. We are happy, users are happy, things work. Uh yeah, so this was designed as per our SLA

and things like that. And okay, we we thought things things should be like predictable. That's what we're looking for here. But, unfortunately, we got a reality check. When we deployed this, our system just fell apart. Despite a uniform design, uh runtimes were very unpredictable with very very high worst-case actuals. Um so, yeah, it kind of looked something Um with some tails breaking our 3-day SLA. a problem

like this uh is a storage hit from two fronts. Um one is that of course, like I mean, there is storage overhead, garbage collection is happening at a slower pace, and therefore, we are having objects around longer than we need. We're paying for them. It's not ideal. Secondly, um SLAs of this pipeline are very important for our uh recovery baseline. So, so this this pipeline is super

important. We decide our recovery overheads based on this And if this pipeline is out of SLA, we cannot be more aggressive with our overheads. We we we need to be safer. We we we need to preserve user data. So, just not being aggressive enough means that okay, there are going to be more overheads. And these overheads matter at our scale. when an infrastructure engineer sees pattern, there

is like one obvious thought. Okay, maybe this is just data skew. I mean, we did partitioning at the user space level, right? Like I mean, did you go and look count the number of objects? We we didn't do that. So, that's what we thought first. And we thought that uh these slow partitions probably have more data hiding inside them. Um so, yeah, we we set out to

do our adventure. We did we put some counters. We did some audits. And interestingly, there was less than 1% variation across these 16 partitions. And that was like more accurate than we thought. The data was really uniform. But, as we saw here, uniform data does not equal uniform performance. So, we had initially assumed that only certain partitions were not doing well by just like empirically looking at

the graphs. But, we were not right. What hap- what was happening was um every partition just had a very high variance in the performance. It was not isolated to any partition. There were just no patterns. we dug into the metrics and we found our ghost. one thing I did not emphasize on a lot is that there is a large-scale join which is going to happen across the

two databases, right? Which is going to be massive. And it needs to happen for for referential integrity to work. So, one may think that okay, maybe the bottleneck is somewhere there. Like the join is not performed. But, that was a problem our engineers had already thought of. Like we we knew that such a join is not going to scale. And therefore, there were a lot of optimizations

in place to make that join work. I mean, it it was not like a simple cross-table join which can get get like really expensive. There were there are certain optimizations in place to make it happen. And that was not the bottleneck. The join was going operating perfectly fine. It was neither um any post-processing that we do on the data. Like I mean, there there are a bunch

of validations, there are a bunch of other uh data quality checks that we need to do once we have this like uh picture on both the sides available. None of that. Everything was working surprisingly well. And the problem was the read stages. Like the operation where you read the the metadata storage as well as the object storage. Like the required data on both the sides before you

do the join. That's where the bottleneck was. That's where the variance was. And what we were seeing was our P95 latencies were like over 2 seconds. Our P99 latencies were like over 10 seconds. And there were very high error rates, too. So, I mean, for for those of you who have worked extensively with batch pipelines, MapReduce, things like that, you probably know this brutal truth that you

only need like a tiny fraction of slow request to like govern the whole pipeline's runtime. I mean, it's always always always the the the slow request, like the the tail. So, that was our ghost here. And when you have that, it actually doesn't really matter if your if the rest of your reads are fast. I mean, your pipeline is going to wait for that 1% to complete

uh before you go to the next stage. I how do you fix a slow read? So, a very common cloud-native pattern a lot of you folks would have heard about it. It's called request hedging. Like I mean, if if a request is slow to a server, you just issue a back paddle backup request. Uh so, so the way this is supposed to work is like okay, probably

if your request is slow, there is a very high chance you hit a server which is like loaded and maybe going to another server, maybe the load balancer next time chooses another server which is going to be like hopefully less loaded, less crowded and you'll probably get a faster response. So, this is your hedge against against the latency. And if you get the response, you can just

go ahead and cancel the previous request. And hopefully you will have better run times. Uh probably one thing one thing uh we need to ensure in such cases is that the request has to be idempotent. And that's really the case here, right? I mean it's it's a read. So, it's idempotent. but but yeah, there is one architectural debate here. Hedging is going to work only if the

latency is transient. Was it the case for us? We'll answer that later. But if it is not the case, um turning on hedge hedging would essentially mean a self-inflicted DDoS attack. So, yeah. Um this is what we call the hyper-scale uncertainty principle, which is the title of the talk. One more thing which we had discovered was a fundamental architectural mismatch. So, we have those 16 pipeline partitions

on the left. Uh what we realized was the object storage system also had underlying beneath because of our massive scale, again things like that. And they were not really aligned with our logical storage with our pipeline logic. we had 16 partitions here, 23 there. Fortunately, they were all again aligned by the username space and we we thought that okay, hey um what happens if we align them?

Like why don't we just like go ahead and align Um because maybe what's happening here is that um one of the partitions here is probably issuing like multiple range scans uh on like probably more than one of these object storage systems when it does the read. And and maybe that's like causing like uneven run times, maybe it's causing load across some of the partitions and things like

that. Uh What What happens if we align them? Would it happen Would it help balance the load? Would it help improve the tail latency? So, so we we set out in our venture there and we realized this at a point. while there was a logical misalignment, these these partitions I mean, not going too much into the internals of the system, but the way it worked was uh

they were having the same underlying physical storage. Uh they were sharing the same resources, similar concepts which basically aligned the partitions to the same infrastructure, same bare metal. So, so it didn't really matter much. If you if if you changed them and and it probably made more sense to like let these systems grow independently. So, aligning them probably was not going to change anything for us. So,

yeah, let's let's let's look at some math. Um Coming back to the to the slow reads, right? I mean, there is a term for that. Uh the term is called stragglers, if you've heard of it. So, if you have if you've ever watched a bad job run, you'd you'd know the pain, right? I mean, every time you look at the pipeline, it's going to be stuck at

99%. It's going to happen far far more common more commonly than you'd you'd realize. Um So, so why does this happen? So, say imagine you have 1,000 workers. Generally, what happens is like you'll probably give them like 1,000 shards of work. A shard basically here is just a unit of work that a worker can process yeah, just imagine like even like uh 1% or like 10% of

those shards, they experience something wrong. It can be uh size like it can be uneven size, some of the shards are just like too big. It can be some transient failures isolated to those shards. Um It can be something else, some other So, that's going to decide your entire pipeline's run time and this is going to be a decisive factor by an order of magnitude. Yeah, so

so what's going to happen is like one worker probably is going to work on some heavy shard and most of the other workers are going to be done. Uh and yeah, that's just like they are just going to wait idle idly and probably going to like burn money without being used. yeah, and since the world is not designed for bad jobs, uh parti- like databases are not

necessarily designed for bad jobs as like first-class citizens, uh this this is actually a very common problem. I mean, this happens far more than what we And there is a very simple mathematical fix for this. I mean, it's not always possible that you can change the underlying source. I'll I'll come to some of that later in But um it's it's always possible to change your pipeline. And

the And the math says that you should over-shard. I mean, you should have um more workers available than the number of shards and you need to have that by a factor. Uh someone has done the math for us. Um So, so if you have n workers, the ideal number of shards is order of n natural log n. Um and there is a constant there and generally like

what people have observed is that the ideal constant is three. So, you you need to have like three n log n shards. So, say you have um 100 workers, I think for for that the number comes out to be around 1,300. So, you basically over-shard and the intuition here is that that is going to hide your tail latency. I mean, your workers are going to have enough

work that your stragglers are no longer going to dominate the run time. That that's like the key idea here. So, yeah, I mean, we just set out to look at this for ourselves. And what we realized was that okay, there was a gap. So, so we had an object Uh we checked we were actually requesting for 450k shards uh in one partition, by the way. And what

we were receiving was just like 30k shards. So, so we had parallelism And we were just having those massive unsplittable blocks which were dominating our run time, which were causing this very high variance in the So, the actual solution here um required some platform modernization. So, what was happening there was that the the reader behind this object it was some legacy map reduce bridge. There was nothing

much we could do there uh for for for making changes to those shards until we did some modernization. Uh we introduced some dynamic work rebalancing. Again, there is a term for that. It's called liquid sharding. Um So, yeah, you can look at those red dash lines, right? I mean, they are the ghost, as we called out previously. what liquid sharding can enable you to do is like

it can take some of these big chunks of work. It can help you I mean, the the framework can help split those chunks into smaller chunks at run time. So, your workers can basically consume those that parallelism. And and And this actually um leads to like very very strong improvements in the run time. So, yeah, I mean, when when our worker pool would go idle, the system

would like unleash um basically try to like take some of these heavy heavy work items and break them down. So, that our workers can just like consume those chunks and we we see improvements. So, yeah, I mean, this this can actually take like hours of tail latency down to minutes. I I it's really powerful. But yeah, of course, this is not something you get for free. I

mean, your platform needs to support that. And yeah, I mean, on its own generally a bad job framework will not be able to do it because you it needs uh support from from the platform itself to do to be able to like show a logical breakdown point for for like breaking those shards. So, I mean, this does require some work. So, yeah, I mean, I I would

want to um pause here and and talk a bit about the economics of compute. FinOps dashboards, they they love best effort or spot instances because the hourly cost looks great on paper. Um and generally the SLOs for bad jobs or async workflows are much more relaxed compared to user-facing requests, live requests or things like that. So, we tend to prefer these for bad jobs. But this case

is a little different. Um our pipeline consumes over hundreds of thousands of CPU units, it consumes almost a petabyte of RAM. I mean, if you combine this um legacy shard issue with this kind of a setup, uh preemptions tend to cost us a lot. The heavier shards get preempted quite often. It leads to a lot of lost work because after preemption, generally the pipeline has to start

processing that shard again. was causing a lot of waste. We were seeing like a greater than 70% waste ratio in um in the read stages. So, yeah, I mean, if your workers if your work chunks take longer to process than your preemption window, cheap compute is not really cheap. So, yeah, I mean, this brings me to the three hard engineering truths which we learned when we operated

at the First, you tend you need to architect your systems for the P99 tail. That's going to dominate your end time. That's going to dominate everything. Um your your batch pipeline is only as fast as your stragglers. You must over shard and there is also something known as re-sharding that some framework support where you where you uh let the framework do it after the read. So that

the first subsequent stages can um benefit from the same. So So what generally happens is that um a batch framework will probably club certain stages in in in one big stage for optimization reasons. So what can a read may be followed by some processing on the data, may be followed by some API call and things like that and that can become one stage. So if you introduce

a re-shard after the the rest of the stages will be split and handled separately by the batch framework. And there it will try to maximize the parallelism by splitting the work. So this generally has a trade-off that okay, you're going to have your pipeline do that. You're going spend some uh disk footprint as well because it has it has to rely on disk to do the re-sharding

but it's going to give you parallelism benefit at least for the subsequent stages. And And this also this trade-off often tends to be worth So yeah, I mean you need to architect Uh we need to respect the bare metal. Uh software boundaries are generally not enough. Um we we need we needed to look at infrastructure here a lot. yeah, I mean finally um be aware of the

ROI of preemption. I mean in our case what we was that um I mean if if the retry rate of the failure rate is so high best effort compute is probably a liability. I mean the cost the cost gains are probably outweighed by this. So yeah, I mean these were the insights. I I hope this was useful. Um you can scan the QR code on the left

to share feedback on the talk. Um the one on the right has the slides, the math, and the references in case someone wants to go deeper. Um I'd like to thank Ben Appleton from Google for for the insights on the math. I mean it's it's the the research is all his work. Uh we are just applying it. Yeah, and I'd also like to thank the photo storage

infra team for for the contributions on like making the system work and protecting the users' memories. Yeah, thank you everyone for your time. This was a pleasure. Happy to take questions. We have like 7 minutes. >> [applause] >> I thank you very much for the talk. I was wondering um so you were always talking about like hot partitioned um or hot partitions. So there's likely not a

shared uh a big name space out of which you can cut virtual partitions that you would then um be able to use for smaller jobs. So you said it's it's always about like really changing the the partitioning. So you So the the jobs cannot be artificially small, right? So that's that's what I took from your talk. So you you always have to to go this big on

an individual partition. You can't go smaller. Like usually you would you would hash this down, right? So um you have a a name space like a bit bit a big hash name space out of which you can cut any arbitrary size of partitions and make the jobs much smaller. Okay, so uh we did try considering increasing the number from 16 as as we grew the product. Um

what happens there is like we I mean being smaller there has an advantage your run time tends to be more predictable. So there is some math which says that uh your worst case run time is going to be higher if the number of partitions there are higher. So So instead of say 16, we had like 160 Um intuitively also we can think that uh the worst case

time for a one of the partitions to run is going to be much higher than what it is when you have lesser number. So while that is the case, our problem here with the tail was within the partitions. So when one partition, I mean even if you have a very high number here, every partition is reading like terabytes of data at least. And it is going to

rely on our underlying systems to provide those batch APIs to do the reads. And when that was happening, uh generally this is tied to the underlying data model. Like for for example, the relational metadata storage is going to have some model which is going to determine the splits of the data and that is going to be those are going to be written. And And they are generally

not I mean they are generally I mean they can be very uneven and they often tend to be. So that's where we were seeing the problems. I mean I I hope I understood your question. But yeah, I mean that's the bottleneck. Right. Yeah, I was I was wondering because you you were mentioning about the the the large compute jobs that you had to do for individual partitions.

That's why I'm saying so preemption on all that. I mean I mean I understand the ROI um argument but it seems like you you do lots of like in insanely heavy lifting um for a single partition. So the question is couldn't you just make the individual jobs even smaller to not lose as much data, not have as much of a >> Got it. like a like a

hot spot. And I took from your talk that the 16 like by design. So there's not like a huge hash name space out of which you can cut any arbitrary length of subset to make smaller jobs. That was my idea. Yeah, actually our system I mean it So this number was arrived at empirically. So the system is capable of like just setting any arbitrary number that we

want. We just don't want to go too high for predictability. I mean it's our engineers were going to like look at jobs. I mean So it just increases complexity and also the math that I talked about that it increases the worst case last um end-to-end completion time. That That was also one of the problems. So that's the trade-off. Like I mean you can cut the jobs and

like reduce the amount of work but you pay the price in this sense in terms of like maintainability and predictability. So hence we we probably like chose to be a bit more conservative here. Right. Yeah, I didn't see that there was another question. Yeah, thank you for the talk. I guess I had a pretty similar question. Let me just quickly go through it. Uh you talked about

exabyte scale which yeah, to me still mind-boggling and 16 shards is essentially just one order of magnitude which still leaves petabyte shard size. yeah, I was also wondering like does that even make a difference versus just running the whole thing as one batch job I still can imagine there must be much more parallelization further down for this to make sense at all. So um going a bit

deeper into why our team ended up doing the partitioning, um so we hit a point at some point some years ago where the batch framework which we relied on was hitting the limits. So the framework I mean these frameworks are generally built on top of MapReduce. There are uh shuffle stages. There are temporary files which are relied upon by the hosted in some uh distributed file system.

So those things were hitting their hard limits. I mean the framework had some limits set at that okay, these temporary files can be up to this order of magnitude and our work was was so yeah, I mean things were just not working. So that was one reason where this became like a very hard requirement. And what we were realizing was like apart from these unpredictable run times

which are derived by the underlying sources uh things were quite predictable when we started off. Like this number was empirically helping us hit our SLA. And we cannot just like I mean to make a change on the order of magnitude here, we probably would have to add like a couple of zeros to the number of partitions which which would be a maintenance nightmare. I mean we would

not we we would need like systems on top of it to see our worst case SLO that okay, what's the worst case uh time where we have a full run completed to ensure that okay, we can base our recovery numbers on that. So we didn't realize or we didn't find that to be worth it. Uh it was its own set of problems and we wanted to rather

um drill down on the engineering solution to like make this work. I mean we probably wouldn't have even partitioned if not for the batch jobs limits. I mean we would probably have tried to like scale the pipeline as is but it it kind of comes from there. Like I I I hope that answers your question. I mean if if I got your question right. Yes, it does.

Thank you so much. It sounds like it's basically the system is already doing lots of partitioning by itself and then it hit its limits so you just made multiple of that. So it's like multi-level partitioning in a way. Yes, yes. It is multi-level partitioning. Yeah. Thank you so much. Thank you so