MIFO: A Query-Semantic Aware Resource Allocation Policy
Kalmegh, Prajakta
Babu, Shivnath
Data Analytics Frameworks encourage sharing of clusters for execution of mixed workloads by promising fairness and isolation along with high performance and resource utilization. However, concurrent query executions on such shared clusters result in increased queue and resource waiting times for queries affecting their overall performance. MIFO is a dataflow aware scheduling policy that mitigates the impacts due to queue and resource contentions by reducing the waiting times for queries near completion. We present heuristics that exploit query semantics to proactively trigger MIFO-based allocations in a workload. Our experiments on Apache Spark using TPCDS benchmark show that compared to a FAIR policy, MIFO provides an improved mean response time, reduced makespan of the workload and average speedup between 1.2x-2.7x in highly concurrent setting with only a momentary deviation in fairness.
 So this is actually join work with my student, Prajakta. So she couldn't make it to the conference for some reason. So here I am giving the talk. I'm a professor of computer science now adjunct at Duke University and also doing a startup on some of these interesting kind of challenges that you'll see, right? I want to start off with a few observations just to bring everybody on the same page. We'll be talking about data analytics frameworks like Hadoop and massively parallel databases where you usually have a lot of different tenants who are submitting queries. Queries get converted into plans and with stages in there where each of these stages, it might be a map-reduced job or a spark, right? Or actually an operator and a query plan which will have a whole bunch of tasks that end up running on the system. So the scheduling of these tasks is usually handled by a resource manager where there are hosts, hosts will actually say, look, I have a slot where I can actually run a particular task and then it becomes a problem of how do we take all of these tasks that are ready to run and then figure out an order. I can figure out which ones should actually run on which host, like, you know, of course, tasks will get co-located on host. So that's usually how task scheduling works. If you really drill down and look at an individual task where it's spent time, usually it kind of goes into three phases. There is a initial wait to get a slot to actually get scheduled on the cluster and then it becomes, I'm computing, I'm actually waiting for resources and I'm computing. So basically there are these three phases that often get interleaved, especially the resource wait and the compute time. And the goal is to ensure that these tasks are actually spending the minimal amount of time waiting, be it on the scheduling part or be it on the resource wait part and then the highest fraction is the compute time. That is the ideal goal. So the first observation that we actually saw analyzing a very large number of, like, you know, data analytics platforms and the workloads that actually run is this notion of, which might be very intuitive, right? What we call decreasing resource profile. So at a high level, right? Tasks are basically queries. They will initially have stages which are reading a lot of data, like maybe doing some map and those kind of transformations on the data, right? And then there'll be joins and aggregates where there might be shuffle involved. And then finally, like in the results are getting written to some external store or back to our file system, right? We kind of look at this initial stages, intermediate stages and like, you know, later stages. The initial stages tend to often be IO intensive. The intermediate stages might be CPU intensive sometimes, like, you know, network heavy. And the later stages, right? They tend to be like, you know, very low on resources because by the time most of the data has been cut down significantly. So really, when you look at a graph like that where we are measuring the compute time spent in each of these stages, the initial stages will have a good fraction of like, you know, compute time. So it's progressively decreases and the latest stages will have very little compute time, right? Now the problem, this really kind of leads into the very first kind of problem that we saw which we call online slot hogging. What this means is like, if you really look at the time spent in the later stages and how much time they're actually waiting, right? Yes, the compute time is small, but they're actually end up waiting a lot of time, especially in the scheduling, right? And the intuitive reason is for this disproportionality is that like most of the scheduling is like in the use of a concept called fairness. That's fair scheduling, we've just heard of it, right? Where the key ideas as soon as these slots are basically the opportunities to kind of schedule a task, they free up, right? They're given to the most underserved, right? And if queries are coming continuously, if stages are running continuously, if tasks are actually being ready for scheduling continuously, often these underserved stages become the earliest stages or the intermediate stages, right? Or like, you know, a task in a new query. As opposed to that task, which is part of a query, which has been, which is a later stage, but the query itself has been running for a long time. So sort of fairness sort of plays in a very interesting way where these final like, you know, stages, the latest stages are task on those stages just end up having to wait for a long time, which is something that like most users don't like because you don't want to see your queries 99% complete, but it's stays in that state for a long time, right? The latency becomes very high. So the challenge what we want to solve, especially if you want to ensure that users who are using the system get their results and the queries finish in a very timely fashion and a quick fashion, right? You want to reduce these scheduling weights so that like, you know, as the latest stages of a particular query start to run and the tasks of them run, you don't end up like, you know, forcing them to wait for a long time, right? So the goal is how can we reduce these scheduling weight, especially for the latest stages of queries that have a decreasing resource profile. The next observation again, might be intuitive, but most of these scheduling systems do not account for when they are scheduling tasks, they don't really account for the actual resource usage. The, like, you know, they don't account for the fact that maybe a particular host is actually like in a pretty resource intensive or maybe is having an IO bottleneck or a CP bottleneck, right? We carefully look at the graph I've shown here. The X axis is a different host and the Y axis plots the runtime of tasks, similar tasks on these different hosts. You quickly see that there is a host number 10, which is like way off the charts. So task scheduling being scheduled and that hosts are taking a long time. If this task scheduling is not aware of this, especially for those latest stages, then you can end up with like, you know, high resource weights, unpredictable resource, like weights. So taking these resources and the contention and maybe it's IO, maybe it's CPU, maybe it's memory, maybe some combination into account, again, especially for those latest stages becomes very important. Otherwise latency can really be unpredictable or pretty high. All right, so that brings us to the second challenge. So this is where we invested some time because it's not an easy problem. It's a problem which has been worked on for a long time, scheduling is not a new problem, right? As we started understanding how these workloads run in these different, like, you know, data analytics platforms, we identified a way in which something that seems counterintuitive, right? Like, you must have heard first and first out, but here we are calling it, like, you know, the most in first out is kind of like, you know, give some priority to the latest stages, to task the latest stages, how can we bring that and make it work while, like, you know, keeping all these other things like trooper, and starvation, and resource fairness, everything into account. That's how we came up with this concept called MIFO and the scheduling algorithm. So it breaks down into, there is a, although I'm calling it offline, right? What I mean by offline here is a phase that is not like, you know, running when a query is actually running, as opposed to a phase which kicks in, just like the scheduling, the real-time adaptive stuff that is actually kicking in when a query is running. So there's an offline phase of identifying which queries are, like, you know, show this kind of a pattern, this decreasing resource profile pattern. So these become the MIFO eligible queries and then find out a time. When do they get to that stage? Well, what is the latest stage in such a query, right? Find a trigger age for coming up with a different way of trading these queries and then a third phase, which is really the phase that runs when the query is actually running of query acceleration, right? I'll go through each of these things, like one step at a time, what I'm calling the MIFO techniques. So the first and foremost thing is to identify when a put like query or a pipeline, if you kind of want to think it that way, has a problem where it can suffer from disproportional scheduling weight. So a pictorial looks like this, right? So the graph you see here, the X axis is a fraction of work done in the query. The Y axis is the fraction of scheduling weight time, right? So what you really want is a straight line, like in a 45 degree slope, where as more and more fraction of work gets done, you're actually proportionally spending the scheduling weight. But a lot of queries, if you really look at in a real system, they will have a pattern that looks like this, right? But this sort of skewed to the right or to the bottom right corner, where a lot of the work happens and then most of the scheduling weight sort of happens in the final phase, right? So identifying queries, not all queries have this behavior. Of course you can have queries which are looking much like what is on the left side. But whenever such queries are identified, you also want to identify what we call the MIFO trigger age, right? What does that knee of the curve where you kind of want to start treating and giving special attention to this query so that it's not incurring a disproportional wait time. So overall, this offline phase kind of looks like this. I know it's pretty high level, but more than happy to kind of go through the details of this in the poster session, right? First step is like, you know, the over the entire workload, identify these queries on which you kind of want to apply MIFO algorithm. So first step there is what we call the needle algorithm, that algorithm that keeps analyzing the, like, you know, the wait time to work done curve and finds the point of maximum curvature, right? And once you have identified such a query, and of course you don't want to react to an individual run and look at repeated runs, so identify such queries. And then the next question is, is the disproportionality high enough that you want to apply MIFO? Like, you know, again, a threshold is used for that. And once you have identified such a query and the disproportionality threshold, then like, you know, you have that MIFO trigger age, right? And then it becomes a question of applying the acceleration like a leader. And the high level of this acceleration itself that happens when the query is really running, we call it dynamic prioritization. That's the main thing. Again, the equation looks like this, but intuitively the way to understand that is MIFO is applied on top of a conventional scheduling algorithm like FAIR. So the FAIR algorithm will usually apply a wait per query or per tenant or like, you know, whatever that, like in a unit of generatis, which is what I'm referring to here as the wait defined by the user or the system. Like, you know, that is the fairness wait. And on top of that, based on the time the query has been spending on top of the, like, you know, the MIFO trigger time, giving it a little more priority so that it can actually be accelerated. Now, this equation is pretty easy to state. There are lots of, like, you know, things you have to take care of, you have to avoid starvation, right? So you don't simply want to kind of treat it that way. You need to have, like, you know, some checks and balances or some high and low water marks. So the details, unfortunately, I don't have time to get into them. They are in the paper, but there are also other factors you have to take into account. And one of the things which I would say, like, you know, five, six years back raised a lot of attention is this concept called delay scheduling to increase the chances that a task will be run on a node where the data is local. Now, with the, like, you know, ways in which network speeds have increased, the ways in which the decoupled storage and compute systems, especially in the cloud, has basically kind of grown, this has not become as important. In fact, the delay scheduling tends to hurt these queries which have, like, you know, the disproportionate wait times significantly. So these are sort of things you kind of want to identify and turn off as well. So it's not just in the scheduling algorithm alone, but it goes, like, you know, it goes slightly broader than that. So that was a scheduling wait. Now coming to the resource wait times, right? Constantly the same. Ideally you want the resource wait times spread across the runtime of the query, but often you end up seeing the, like, you know, the graphs I show on the right side where the latest stages kind of run into disproportional wait time. Same idea, identify the trigger age, right? And then apply it. Now, the way in which it applies at runtime, the way it is taken into account is we actually, as these queries are running, the last stage is running, tasks are running on every host, but every kind of resource, keep, like, you know, computing the resource wait time on average that is seen on that host. Now when a host says, look, I have a slot to run a task, right, compare the average resource wait time, for example, with the resource kind of wait time that we are actually seeing for that particular, like, you know, the tasks of a particular stage. And then if it is higher than average, like, you know, just relinquish that. Don't even schedule the task over there, but if it is less than running over there, right? That's a conceptual idea that needs to be incorporated to ensure that, like, even the resource wait times end up not being a significant fraction of the latest stages, right? So that's all I'm going to talk about, the techniques themselves, more than happy to drill down into details, and there are quite a bunch of details in this to kind of really make it work in practice. What I want to spend some time on is actually, like, you know, how does it all work, right? So we have been doing a lot of experiments, so once I will actually share, have been done in Spark 2.3. Spark has, like, you know, this is done in Spark, has its own standalone scheduler. This is not Spark running on a, like, a framework like Yarn, but the concept is the same. We just chose this because it was easier for us to implement, and, like, showcase the value. We have been running mostly, like, you know, SQL kind of queries, as I was going to mention, a whole bunch of them chosen from the TPCDS benchmark, right? So the key things to keep in mind are when you're actually working on a scheduling problem, it is sort of a multi-objective optimization problem, right? From an end user's perspective, the metrics they care about are things like, how much time do I take for my workload to finish, right? The mix span, or what is the mean response time of my queries, right? Or what is the, like, you know, speed up, I'm actually getting with respect to a different technique. So that is the end user's view. End users don't care about throughput, and, like, you know, resource utilization, and all of that stuff. But if you just transition to the operator, right, of the, like, the so-called DBA in this environment, right, the operations person, they're not focusing on individual queries. They want high throughput. They want a CPU decision to be high. Now, the big growth on the cloud, right? Of course, you actually want, you're looking at cost. End users never look at cost, right? They're basically trying to make their queries faster. So sort of becomes a kind of interesting trade-off. So whenever you're developing algorithms, you really have to think about their trade-off and work with it, right? And we have done extensive set of experiments comparing the MIFO algorithm with the sort of the state-of-the-art algorithm that's employed in most places, the FAIR algorithm, as well as, like, you know, things like FIFO or the shortest, like, you know, remaining processing time, first kind of algorithm. And, of course, as you can imagine, when you are implementing a scheduling algorithm, especially one that depends on things like, you know, some kind of prediction of the runtime, you need to think about sensitivity, you need to think about scalability, because it's actually running, not in the context of few tasks, but, like, you know, sometimes hundreds, if not, like, you know, thousands of tasks concurrently, right? Scalability and whatnot. The results I'll show are primarily comparing with the FAIR algorithm, as well as, like, you know, as concurrency in the workload increases, right? A quick, like, you know, overview of the results. So the top, like, you know, the dashboard kind of numbers you see are the numbers across the entire spectrum of experiments that we have done, the speedups and all these different metrics, now, how much we have improved, the FAIR algorithm improves over FAIR, which is the sort of the state-of-the-art algorithm in many of these systems. As you can see, the makespan, which is the completion time of the workload, like, you know, 30% increase, or better than, you know, what FAIR can actually offer or the mean response time of the queries in the workload has increased, or actually in terms of absolute numbers has decreased, but it has improved 48%, speedup is high, right? So those are the end-user visible metrics. Interestingly enough, the moment you can have taken to account, like, you know, the wait time and an attitude of, like, I want to minimize wait times, it also ends up improving the, like, you know, the resource metrics, like CPU utilization, right? A huge fraction of that, which I'll kind of drill down into a little bit more detail, whose fraction of that comes because the moment you've taken to account the resource usage of all of these different, like, you know, hosts when you're scheduling tasks, like, you know, and be intelligent to that, resource utilization can increase, or task throughput can increase, right? Where we are paying some penalty is on this thing called the resource fairness index, right? Yep, resource fairness index is smaller, but how many of you even know about resource fairness index? Right? You probably don't even care, right? The other things that would really make the most difference, but fairness is something which in the grand scheme of things is important to consider, otherwise starvation can happen, right? So on the bottom part, like, when you see some more results, the bottom, like, you know, left, what you're seeing is the overall mixpan of these four algorithms, when it says fair five, what that means, that five means is the concurrency, like, you know, number of queries, like, you know, coming in, right? What you see is mixpan, all the other techniques are better than fair. Now, one thing to keep in mind is the FIFO and the shortest remaining processing time first, they come up with big disadvantages of their own, right? FIFO can run into starvation, FIFO often actually means, like, you know, convoy effect and things like that, but FIFO also has a, like, a behavior that most people don't even think, right? Tasks in a query or tasks in a stage end up having a similar resource profile. So if you're just letting things in a FIFO fashion, you end up having to schedule tasks, you end up having to schedule tasks, which all have a similar resource profile. And then it's pretty hard to, like, you know, balance the resource usage across hosts, right? So pretty interesting things to kind of keep in mind as you really drill down into how these different algorithms behave. Like, you know, the, I'm just going to skip to the, the plot on the bottom right, which is actually showing, it's on the log scale. So it shows how the, across these different scheduling algorithms, overall compute time remains the same, but there are big, like, you know, improvements in the scheduling and resource wait times. So I want to use one more minute to wrap up. Is that, that is not okay. So, great. So that is pretty much it. So overall, I'm going to actually, like, you know, summarize by saying, this is a problem that has received a lot of attention, but there's still a lot of room to improve. And if you're interested in these sort of problems, like, you know, we are actually working on these things, not just in academia, but also in the industry. There are interesting opportunities for great algorithms to, like, you know, be applied to application performance management. Thank you. Thank you. Thank you. Thank you. Thank you. Thank you.
 So this is actually join work with my student, Prajakta.  So she couldn't make it to the conference for some reason.  So here I am giving the talk.  I'm a professor of computer science now adjunct  at Duke University and also doing a startup  on some of these interesting kind of challenges  that you'll see, right?  I want to start off with a few observations  just to bring everybody on the same page.  We'll be talking about data analytics frameworks  like Hadoop and massively parallel databases  where you usually have a lot of different tenants  who are submitting queries.  Queries get converted into plans and with stages in there  where each of these stages,  it might be a map-reduced job or a spark, right?  Or actually an operator and a query plan  which will have a whole bunch of tasks  that end up running on the system.  So the scheduling of these tasks  is usually handled by a resource manager  where there are hosts, hosts will actually say,  look, I have a slot where I can actually run a particular task  and then it becomes a problem of how do we take all  of these tasks that are ready to run  and then figure out an order.  I can figure out which ones should actually run  on which host, like, you know, of course,  tasks will get co-located on host.  So that's usually how task scheduling works.  If you really drill down and look at an individual task  where it's spent time, usually it kind of goes into three phases.  There is a initial wait to get a slot  to actually get scheduled on the cluster  and then it becomes, I'm computing,  I'm actually waiting for resources and I'm computing.  So basically there are these three phases  that often get interleaved,  especially the resource wait and the compute time.  And the goal is to ensure that these tasks  are actually spending the minimal amount of time waiting,  be it on the scheduling part  or be it on the resource wait part  and then the highest fraction is the compute time.  That is the ideal goal.  So the first observation that we actually saw  analyzing a very large number of, like, you know,  data analytics platforms and the workloads that actually run  is this notion of, which might be very intuitive, right?  What we call decreasing resource profile.  So at a high level, right?  Tasks are basically queries.  They will initially have stages  which are reading a lot of data,  like maybe doing some map  and those kind of transformations on the data, right?  And then there'll be joins and aggregates  where there might be shuffle involved.  And then finally, like in the results  are getting written to some external store  or back to our file system, right?  We kind of look at this initial stages, intermediate stages  and like, you know, later stages.  The initial stages tend to often be IO intensive.  The intermediate stages might be CPU intensive sometimes,  like, you know, network heavy.  And the later stages, right?  They tend to be like, you know, very low on resources  because by the time most of the data  has been cut down significantly.  So really, when you look at a graph like that  where we are measuring the compute time spent  in each of these stages,  the initial stages will have a good fraction  of like, you know, compute time.  So it's progressively decreases  and the latest stages will have very little compute time, right?  Now the problem, this really kind of leads  into the very first kind of problem that we saw  which we call online slot hogging.  What this means is like,  if you really look at the time spent in the later stages  and how much time they're actually waiting, right?  Yes, the compute time is small,  but they're actually end up waiting a lot of time,  especially in the scheduling, right?  And the intuitive reason is for this disproportionality  is that like most of the scheduling  is like in the use of a concept called fairness.  That's fair scheduling, we've just heard of it, right?  Where the key ideas as soon as these slots  are basically the opportunities to kind of schedule a task,  they free up, right?  They're given to the most underserved, right?  And if queries are coming continuously,  if stages are running continuously,  if tasks are actually being ready  for scheduling continuously,  often these underserved stages become the earliest stages  or the intermediate stages, right?  Or like, you know, a task in a new query.  As opposed to that task, which is part of a query,  which has been, which is a later stage,  but the query itself has been running for a long time.  So sort of fairness sort of plays  in a very interesting way where these final like,  you know, stages, the latest stages are task  on those stages just end up having to wait for a long time,  which is something that like most users don't like  because you don't want to see your queries 99% complete,  but it's stays in that state for a long time, right?  The latency becomes very high.  So the challenge what we want to solve,  especially if you want to ensure that users  who are using the system get their results  and the queries finish in a very timely fashion  and a quick fashion, right?  You want to reduce these scheduling weights  so that like, you know, as the latest stages  of a particular query start to run  and the tasks of them run, you don't end up like,  you know, forcing them to wait for a long time, right?  So the goal is how can we reduce these scheduling weight,  especially for the latest stages of queries  that have a decreasing resource profile.  The next observation again, might be intuitive,  but most of these scheduling systems do not account for  when they are scheduling tasks,  they don't really account for the actual resource usage.  The, like, you know, they don't account for the fact  that maybe a particular host is actually like  in a pretty resource intensive  or maybe is having an IO bottleneck or a CP bottleneck, right?  We carefully look at the graph I've shown here.  The X axis is a different host and the Y axis plots  the runtime of tasks, similar tasks on these different hosts.  You quickly see that there is a host number 10,  which is like way off the charts.  So task scheduling being scheduled  and that hosts are taking a long time.  If this task scheduling is not aware of this,  especially for those latest stages,  then you can end up with like, you know,  high resource weights, unpredictable resource, like weights.  So taking these resources and the contention  and maybe it's IO, maybe it's CPU,  maybe it's memory, maybe some combination into account,  again, especially for those latest stages  becomes very important.  Otherwise latency can really be unpredictable or pretty high.  All right, so that brings us to the second challenge.  So this is where we invested some time  because it's not an easy problem.  It's a problem which has been worked on for a long time,  scheduling is not a new problem, right?  As we started understanding how these workloads run  in these different, like, you know, data analytics platforms,  we identified a way in which something  that seems counterintuitive, right?  Like, you must have heard first and first out,  but here we are calling it, like, you know,  the most in first out is kind of like, you know,  give some priority to the latest stages,  to task the latest stages, how can we bring that  and make it work while, like, you know,  keeping all these other things like trooper,  and starvation, and resource fairness,  everything into account.  That's how we came up with this concept called MIFO  and the scheduling algorithm.  So it breaks down into, there is a,  although I'm calling it offline, right?  What I mean by offline here is a phase  that is not like, you know, running  when a query is actually running,  as opposed to a phase which kicks in,  just like the scheduling, the real-time adaptive stuff  that is actually kicking in when a query is running.  So there's an offline phase of identifying  which queries are, like, you know,  show this kind of a pattern,  this decreasing resource profile pattern.  So these become the MIFO eligible queries  and then find out a time.  When do they get to that stage?  Well, what is the latest stage in such a query, right?  Find a trigger age for coming up  with a different way of trading these queries  and then a third phase, which is really the phase  that runs when the query is actually running  of query acceleration, right?  I'll go through each of these things,  like one step at a time,  what I'm calling the MIFO techniques.  So the first and foremost thing is to identify  when a put like query or a pipeline,  if you kind of want to think it that way,  has a problem where it can suffer  from disproportional scheduling weight.  So a pictorial looks like this, right?  So the graph you see here,  the X axis is a fraction of work done in the query.  The Y axis is the fraction of scheduling weight time, right?  So what you really want is a straight line,  like in a 45 degree slope,  where as more and more fraction of work gets done,  you're actually proportionally spending the scheduling weight.  But a lot of queries,  if you really look at in a real system,  they will have a pattern that looks like this, right?  But this sort of skewed to the right  or to the bottom right corner,  where a lot of the work happens  and then most of the scheduling weight  sort of happens in the final phase, right?  So identifying queries,  not all queries have this behavior.  Of course you can have queries  which are looking much like what is on the left side.  But whenever such queries are identified,  you also want to identify what we call  the MIFO trigger age, right?  What does that knee of the curve  where you kind of want to start treating  and giving special attention to this query  so that it's not incurring a disproportional wait time.  So overall, this offline phase kind of looks like this.  I know it's pretty high level,  but more than happy to kind of go through the details  of this in the poster session, right?  First step is like, you know,  the over the entire workload,  identify these queries on  which you kind of want to apply MIFO algorithm.  So first step there is what we call the needle algorithm,  that algorithm that keeps analyzing the,  like, you know, the wait time to work done curve  and finds the point of maximum curvature, right?  And once you have identified such a query,  and of course you don't want to react to an individual run  and look at repeated runs, so identify such queries.  And then the next question is,  is the disproportionality high enough  that you want to apply MIFO?  Like, you know, again, a threshold is used for that.  And once you have identified such a query  and the disproportionality threshold,  then like, you know, you have that MIFO trigger age, right?  And then it becomes a question of applying the acceleration  like a leader.  And the high level of this acceleration itself  that happens when the query is really running,  we call it dynamic prioritization.  That's the main thing.  Again, the equation looks like this,  but intuitively the way to understand that is  MIFO is applied on top of a conventional scheduling algorithm  like FAIR.  So the FAIR algorithm will usually apply a wait per query  or per tenant or like, you know, whatever that,  like in a unit of generatis,  which is what I'm referring to here as the wait  defined by the user or the system.  Like, you know, that is the fairness wait.  And on top of that, based on the time  the query has been spending on top of the, like, you know,  the MIFO trigger time,  giving it a little more priority  so that it can actually be accelerated.  Now, this equation is pretty easy to state.  There are lots of, like, you know,  things you have to take care of,  you have to avoid starvation, right?  So you don't simply want to kind of treat it that way.  You need to have, like, you know,  some checks and balances  or some high and low water marks.  So the details, unfortunately,  I don't have time to get into them.  They are in the paper,  but there are also other factors you have to take into account.  And one of the things which I would say, like, you know,  five, six years back raised a lot of attention  is this concept called delay scheduling  to increase the chances that a task will be run on a node  where the data is local.  Now, with the, like, you know, ways in which  network speeds have increased,  the ways in which the decoupled storage and compute systems,  especially in the cloud, has basically kind of grown,  this has not become as important.  In fact, the delay scheduling  tends to hurt these queries  which have, like, you know,  the disproportionate wait times significantly.  So these are sort of things you kind of want to identify  and turn off as well.  So it's not just in the scheduling algorithm alone,  but it goes, like, you know,  it goes slightly broader than that.  So that was a scheduling wait.  Now coming to the resource wait times, right?  Constantly the same.  Ideally you want the resource wait times spread across  the runtime of the query,  but often you end up seeing the, like, you know,  the graphs I show on the right side where the latest stages  kind of run into disproportional wait time.  Same idea, identify the trigger age, right?  And then apply it.  Now, the way in which it applies at runtime,  the way it is taken into account is we actually,  as these queries are running,  the last stage is running, tasks are running  on every host, but every kind of resource,  keep, like, you know, computing the resource wait time  on average that is seen on that host.  Now when a host says, look, I have a slot to run a task,  right, compare the average resource wait time, for example,  with the resource kind of wait time  that we are actually seeing for that particular,  like, you know, the tasks of a particular stage.  And then if it is higher than average,  like, you know, just relinquish that.  Don't even schedule the task over there,  but if it is less than running over there, right?  That's a conceptual idea that needs to be incorporated  to ensure that, like, even the resource wait times  end up not being a significant fraction  of the latest stages, right?  So that's all I'm going to talk about,  the techniques themselves, more than happy to drill down  into details, and there are quite a bunch of details  in this to kind of really make it work in practice.  What I want to spend some time on is actually, like,  you know, how does it all work, right?  So we have been doing a lot of experiments,  so once I will actually share,  have been done in Spark 2.3.  Spark has, like, you know, this is done in Spark,  has its own standalone scheduler.  This is not Spark running on a, like,  a framework like Yarn, but the concept is the same.  We just chose this because it was easier for us to implement,  and, like, showcase the value.  We have been running mostly, like,  you know, SQL kind of queries, as I was going to mention,  a whole bunch of them chosen from the TPCDS benchmark, right?  So the key things to keep in mind  are when you're actually working on a scheduling problem,  it is sort of a multi-objective optimization problem, right?  From an end user's perspective,  the metrics they care about are things like,  how much time do I take for my workload to finish, right?  The mix span, or what is the mean response time  of my queries, right?  Or what is the, like, you know, speed up,  I'm actually getting with respect to a different technique.  So that is the end user's view.  End users don't care about throughput,  and, like, you know, resource utilization,  and all of that stuff.  But if you just transition to the operator, right,  of the, like, the so-called DBA in this environment, right,  the operations person,  they're not focusing on individual queries.  They want high throughput.  They want a CPU decision to be high.  Now, the big growth on the cloud, right?  Of course, you actually want, you're looking at cost.  End users never look at cost, right?  They're basically trying to make their queries faster.  So sort of becomes a kind of interesting trade-off.  So whenever you're developing algorithms,  you really have to think about their trade-off  and work with it, right?  And we have done extensive set of experiments  comparing the MIFO algorithm with the sort of the state-of-the-art  algorithm that's employed in most places,  the FAIR algorithm, as well as, like, you know,  things like FIFO or the shortest,  like, you know, remaining processing time,  first kind of algorithm.  And, of course, as you can imagine,  when you are implementing a scheduling algorithm,  especially one that depends on things like, you know,  some kind of prediction of the runtime,  you need to think about sensitivity,  you need to think about scalability,  because it's actually running, not in the context of few tasks,  but, like, you know, sometimes hundreds, if not,  like, you know, thousands of tasks concurrently, right?  Scalability and whatnot.  The results I'll show are primarily comparing  with the FAIR algorithm, as well as, like, you know,  as concurrency in the workload increases, right?  A quick, like, you know, overview of the results.  So the top, like, you know, the dashboard kind of numbers  you see are the numbers across the entire spectrum  of experiments that we have done,  the speedups and all these different metrics,  now, how much we have improved,  the FAIR algorithm improves over FAIR,  which is the sort of the state-of-the-art algorithm  in many of these systems.  As you can see, the makespan,  which is the completion time of the workload,  like, you know, 30% increase,  or better than, you know, what FAIR can actually offer  or the mean response time of the queries  in the workload has increased,  or actually in terms of absolute numbers has decreased,  but it has improved 48%, speedup is high, right?  So those are the end-user visible metrics.  Interestingly enough, the moment you can have taken  to account, like, you know, the wait time  and an attitude of, like, I want to minimize wait times,  it also ends up improving the, like, you know,  the resource metrics, like CPU utilization, right?  A huge fraction of that, which I'll kind of drill down  into a little bit more detail,  whose fraction of that comes because the moment you've taken  to account the resource usage of all of these different,  like, you know, hosts when you're scheduling tasks,  like, you know, and be intelligent to that,  resource utilization can increase,  or task throughput can increase, right?  Where we are paying some penalty is on this thing  called the resource fairness index, right?  Yep, resource fairness index is smaller,  but how many of you even know about resource fairness index?  Right? You probably don't even care, right?  The other things that would really make the most difference,  but fairness is something which in the grand scheme  of things is important to consider,  otherwise starvation can happen, right?  So on the bottom part, like, when you see some more results,  the bottom, like, you know, left, what you're seeing is  the overall mixpan of these four algorithms,  when it says fair five, what that means,  that five means is the concurrency,  like, you know, number of queries, like, you know,  coming in, right?  What you see is mixpan, all the other techniques  are better than fair.  Now, one thing to keep in mind is the FIFO  and the shortest remaining processing time first,  they come up with big disadvantages of their own, right?  FIFO can run into starvation,  FIFO often actually means, like, you know,  convoy effect and things like that,  but FIFO also has a, like, a behavior  that most people don't even think, right?  Tasks in a query or tasks in a stage  end up having a similar resource profile.  So if you're just letting things in a FIFO fashion,  you end up having to schedule tasks,  you end up having to schedule tasks,  which all have a similar resource profile.  And then it's pretty hard to, like,  you know, balance the resource usage across hosts, right?  So pretty interesting things to kind of keep in mind  as you really drill down into  how these different algorithms behave.  Like, you know, the, I'm just going to skip to the,  the plot on the bottom right,  which is actually showing, it's on the log scale.  So it shows how the,  across these different scheduling algorithms,  overall compute time remains the same,  but there are big, like, you know, improvements  in the scheduling and resource wait times.  So I want to use one more minute to wrap up.  Is that, that is not okay.  So, great.  So that is pretty much it.  So overall, I'm going to actually, like,  you know, summarize by saying,  this is a problem that has received a lot of attention,  but there's still a lot of room to improve.  And if you're interested in these sort of problems,  like, you know, we are actually working on these things,  not just in academia, but also in the industry.  There are interesting opportunities for great algorithms  to, like, you know, be applied  to application performance management.  Thank you.  Thank you.  Thank you.  Thank you.  Thank you.  Thank you.
Kalmegh_Babu_0_143.png Kalmegh_Babu_144_215.png Kalmegh_Babu_216_311.png Kalmegh_Babu_312_431.png Kalmegh_Babu_432_479.png Kalmegh_Babu_480_551.png Kalmegh_Babu_552_611.png Kalmegh_Babu_612_2015.png Kalmegh_Babu_2016_4727.png Kalmegh_Babu_4728_8783.png Kalmegh_Babu_8784_9215.png Kalmegh_Babu_9216_11495.png Kalmegh_Babu_11496_18203.png Kalmegh_Babu_18204_19511.png Kalmegh_Babu_19512_19763.png Kalmegh_Babu_19764_23219.png Kalmegh_Babu_23220_27899.png Kalmegh_Babu_27900_28511.png Kalmegh_Babu_28512_28559.png Kalmegh_Babu_28560_28619.png Kalmegh_Babu_28620_28739.png Kalmegh_Babu_28740_28882.png
So this is actually join work with my student, Prajakta. So she couldn't make it to the conference for some reason. So here I am giving the talk. I'm a professor of computer science now adjunct at Duke University and also doing a startup on some of these interesting kind of challenges that you'll see, right? I want to start off with a few observations just to bring everybody on the same page. We'll be talking about data analytics frameworks like Hadoop and massively parallel databases where you usually have a lot of different tenants who are submitting queries. Queries get converted into plans and with stages in there where each of these stages, it might be a map-reduced job or a spark, right? Or actually an operator and a query plan which will have a whole bunch of tasks that end up running on the system. So the scheduling of these tasks is usually handled by a resource manager where there are hosts, hosts will actually say, look, I have a slot where I can actually run a particular task and then it becomes a problem of how do we take all of these tasks that are ready to run and then figure out an order. I can figure out which ones should actually run on which host, like, you know, of course, tasks will get co-located on host. So that's usually how task scheduling works. If you really drill down and look at an individual task where it's spent time, usually it kind of goes into three phases. There is a initial wait to get a slot to actually get scheduled on the cluster and then it becomes, I'm computing, I'm actually waiting for resources and I'm computing. So basically there are these three phases that often get interleaved, especially the resource wait and the compute time. And the goal is to ensure that these tasks are actually spending the minimal amount of time waiting, be it on the scheduling part or be it on the resource wait part and then the highest fraction is the compute time. That is the ideal goal. So the first observation that we actually saw analyzing a very large number of, like, you know, data analytics platforms and the workloads that actually run is this notion of, which might be very intuitive, right? What we call decreasing resource profile. So at a high level, right? Tasks are basically queries. They will initially have stages which are reading a lot of data, like maybe doing some map and those kind of transformations on the data, right? And then there'll be joins and aggregates where there might be shuffle involved. And then finally, like in the results are getting written to some external store or back to our file system, right? We kind of look at this initial stages, intermediate stages and like, you know, later stages. The initial stages tend to often be IO intensive. The intermediate stages might be CPU intensive sometimes, like, you know, network heavy. And the later stages, right? They tend to be like, you know, very low on resources because by the time most of the data has been cut down significantly. So really, when you look at a graph like that where we are measuring the compute time spent in each of these stages, the initial stages will have a good fraction of like, you know, compute time. So it's progressively decreases and the latest stages will have very little compute time, right? Now the problem, this really kind of leads into the very first kind of problem that we saw which we call online slot hogging. What this means is like, if you really look at the time spent in the later stages and how much time they're actually waiting, right? Yes, the compute time is small, but they're actually end up waiting a lot of time, especially in the scheduling, right? And the intuitive reason is for this disproportionality is that like most of the scheduling is like in the use of a concept called fairness. That's fair scheduling, we've just heard of it, right? Where the key ideas as soon as these slots are basically the opportunities to kind of schedule a task, they free up, right? They're given to the most underserved, right? And if queries are coming continuously, if stages are running continuously, if tasks are actually being ready for scheduling continuously, often these underserved stages become the earliest stages or the intermediate stages, right? Or like, you know, a task in a new query. As opposed to that task, which is part of a query, which has been, which is a later stage, but the query itself has been running for a long time. So sort of fairness sort of plays in a very interesting way where these final like, you know, stages, the latest stages are task on those stages just end up having to wait for a long time, which is something that like most users don't like because you don't want to see your queries 99% complete, but it's stays in that state for a long time, right? The latency becomes very high. So the challenge what we want to solve, especially if you want to ensure that users who are using the system get their results and the queries finish in a very timely fashion and a quick fashion, right? You want to reduce these scheduling weights so that like, you know, as the latest stages of a particular query start to run and the tasks of them run, you don't end up like, you know, forcing them to wait for a long time, right? So the goal is how can we reduce these scheduling weight, especially for the latest stages of queries that have a decreasing resource profile. The next observation again, might be intuitive, but most of these scheduling systems do not account for when they are scheduling tasks, they don't really account for the actual resource usage. The, like, you know, they don't account for the fact that maybe a particular host is actually like in a pretty resource intensive or maybe is having an IO bottleneck or a CP bottleneck, right? We carefully look at the graph I've shown here. The X axis is a different host and the Y axis plots the runtime of tasks, similar tasks on these different hosts. You quickly see that there is a host number 10, which is like way off the charts. So task scheduling being scheduled and that hosts are taking a long time. If this task scheduling is not aware of this, especially for those latest stages, then you can end up with like, you know, high resource weights, unpredictable resource, like weights. So taking these resources and the contention and maybe it's IO, maybe it's CPU, maybe it's memory, maybe some combination into account, again, especially for those latest stages becomes very important. Otherwise latency can really be unpredictable or pretty high. All right, so that brings us to the second challenge. So this is where we invested some time because it's not an easy problem. It's a problem which has been worked on for a long time, scheduling is not a new problem, right? As we started understanding how these workloads run in these different, like, you know, data analytics platforms, we identified a way in which something that seems counterintuitive, right? Like, you must have heard first and first out, but here we are calling it, like, you know, the most in first out is kind of like, you know, give some priority to the latest stages, to task the latest stages, how can we bring that and make it work while, like, you know, keeping all these other things like trooper, and starvation, and resource fairness, everything into account. That's how we came up with this concept called MIFO and the scheduling algorithm. So it breaks down into, there is a, although I'm calling it offline, right? What I mean by offline here is a phase that is not like, you know, running when a query is actually running, as opposed to a phase which kicks in, just like the scheduling, the real-time adaptive stuff that is actually kicking in when a query is running. So there's an offline phase of identifying which queries are, like, you know, show this kind of a pattern, this decreasing resource profile pattern. So these become the MIFO eligible queries and then find out a time. When do they get to that stage? Well, what is the latest stage in such a query, right? Find a trigger age for coming up with a different way of trading these queries and then a third phase, which is really the phase that runs when the query is actually running of query acceleration, right? I'll go through each of these things, like one step at a time, what I'm calling the MIFO techniques. So the first and foremost thing is to identify when a put like query or a pipeline, if you kind of want to think it that way, has a problem where it can suffer from disproportional scheduling weight. So a pictorial looks like this, right? So the graph you see here, the X axis is a fraction of work done in the query. The Y axis is the fraction of scheduling weight time, right? So what you really want is a straight line, like in a 45 degree slope, where as more and more fraction of work gets done, you're actually proportionally spending the scheduling weight. But a lot of queries, if you really look at in a real system, they will have a pattern that looks like this, right? But this sort of skewed to the right or to the bottom right corner, where a lot of the work happens and then most of the scheduling weight sort of happens in the final phase, right? So identifying queries, not all queries have this behavior. Of course you can have queries which are looking much like what is on the left side. But whenever such queries are identified, you also want to identify what we call the MIFO trigger age, right? What does that knee of the curve where you kind of want to start treating and giving special attention to this query so that it's not incurring a disproportional wait time. So overall, this offline phase kind of looks like this. I know it's pretty high level, but more than happy to kind of go through the details of this in the poster session, right? First step is like, you know, the over the entire workload, identify these queries on which you kind of want to apply MIFO algorithm. So first step there is what we call the needle algorithm, that algorithm that keeps analyzing the, like, you know, the wait time to work done curve and finds the point of maximum curvature, right? And once you have identified such a query, and of course you don't want to react to an individual run and look at repeated runs, so identify such queries. And then the next question is, is the disproportionality high enough that you want to apply MIFO? Like, you know, again, a threshold is used for that. And once you have identified such a query and the disproportionality threshold, then like, you know, you have that MIFO trigger age, right? And then it becomes a question of applying the acceleration like a leader. And the high level of this acceleration itself that happens when the query is really running, we call it dynamic prioritization. That's the main thing. Again, the equation looks like this, but intuitively the way to understand that is MIFO is applied on top of a conventional scheduling algorithm like FAIR. So the FAIR algorithm will usually apply a wait per query or per tenant or like, you know, whatever that, like in a unit of generatis, which is what I'm referring to here as the wait defined by the user or the system. Like, you know, that is the fairness wait. And on top of that, based on the time the query has been spending on top of the, like, you know, the MIFO trigger time, giving it a little more priority so that it can actually be accelerated. Now, this equation is pretty easy to state. There are lots of, like, you know, things you have to take care of, you have to avoid starvation, right? So you don't simply want to kind of treat it that way. You need to have, like, you know, some checks and balances or some high and low water marks. So the details, unfortunately, I don't have time to get into them. They are in the paper, but there are also other factors you have to take into account. And one of the things which I would say, like, you know, five, six years back raised a lot of attention is this concept called delay scheduling to increase the chances that a task will be run on a node where the data is local. Now, with the, like, you know, ways in which network speeds have increased, the ways in which the decoupled storage and compute systems, especially in the cloud, has basically kind of grown, this has not become as important. In fact, the delay scheduling tends to hurt these queries which have, like, you know, the disproportionate wait times significantly. So these are sort of things you kind of want to identify and turn off as well. So it's not just in the scheduling algorithm alone, but it goes, like, you know, it goes slightly broader than that. So that was a scheduling wait. Now coming to the resource wait times, right? Constantly the same. Ideally you want the resource wait times spread across the runtime of the query, but often you end up seeing the, like, you know, the graphs I show on the right side where the latest stages kind of run into disproportional wait time. Same idea, identify the trigger age, right? And then apply it. Now, the way in which it applies at runtime, the way it is taken into account is we actually, as these queries are running, the last stage is running, tasks are running on every host, but every kind of resource, keep, like, you know, computing the resource wait time on average that is seen on that host. Now when a host says, look, I have a slot to run a task, right, compare the average resource wait time, for example, with the resource kind of wait time that we are actually seeing for that particular, like, you know, the tasks of a particular stage. And then if it is higher than average, like, you know, just relinquish that. Don't even schedule the task over there, but if it is less than running over there, right? That's a conceptual idea that needs to be incorporated to ensure that, like, even the resource wait times end up not being a significant fraction of the latest stages, right? So that's all I'm going to talk about, the techniques themselves, more than happy to drill down into details, and there are quite a bunch of details in this to kind of really make it work in practice. What I want to spend some time on is actually, like, you know, how does it all work, right? So we have been doing a lot of experiments, so once I will actually share, have been done in Spark 2.3. Spark has, like, you know, this is done in Spark, has its own standalone scheduler. This is not Spark running on a, like, a framework like Yarn, but the concept is the same. We just chose this because it was easier for us to implement, and, like, showcase the value. We have been running mostly, like, you know, SQL kind of queries, as I was going to mention, a whole bunch of them chosen from the TPCDS benchmark, right? So the key things to keep in mind are when you're actually working on a scheduling problem, it is sort of a multi-objective optimization problem, right? From an end user's perspective, the metrics they care about are things like, how much time do I take for my workload to finish, right? The mix span, or what is the mean response time of my queries, right? Or what is the, like, you know, speed up, I'm actually getting with respect to a different technique. So that is the end user's view. End users don't care about throughput, and, like, you know, resource utilization, and all of that stuff. But if you just transition to the operator, right, of the, like, the so-called DBA in this environment, right, the operations person, they're not focusing on individual queries. They want high throughput. They want a CPU decision to be high. Now, the big growth on the cloud, right? Of course, you actually want, you're looking at cost. End users never look at cost, right? They're basically trying to make their queries faster. So sort of becomes a kind of interesting trade-off. So whenever you're developing algorithms, you really have to think about their trade-off and work with it, right? And we have done extensive set of experiments comparing the MIFO algorithm with the sort of the state-of-the-art algorithm that's employed in most places, the FAIR algorithm, as well as, like, you know, things like FIFO or the shortest, like, you know, remaining processing time, first kind of algorithm. And, of course, as you can imagine, when you are implementing a scheduling algorithm, especially one that depends on things like, you know, some kind of prediction of the runtime, you need to think about sensitivity, you need to think about scalability, because it's actually running, not in the context of few tasks, but, like, you know, sometimes hundreds, if not, like, you know, thousands of tasks concurrently, right? Scalability and whatnot. The results I'll show are primarily comparing with the FAIR algorithm, as well as, like, you know, as concurrency in the workload increases, right? A quick, like, you know, overview of the results. So the top, like, you know, the dashboard kind of numbers you see are the numbers across the entire spectrum of experiments that we have done, the speedups and all these different metrics, now, how much we have improved, the FAIR algorithm improves over FAIR, which is the sort of the state-of-the-art algorithm in many of these systems. As you can see, the makespan, which is the completion time of the workload, like, you know, 30% increase, or better than, you know, what FAIR can actually offer or the mean response time of the queries in the workload has increased, or actually in terms of absolute numbers has decreased, but it has improved 48%, speedup is high, right? So those are the end-user visible metrics. Interestingly enough, the moment you can have taken to account, like, you know, the wait time and an attitude of, like, I want to minimize wait times, it also ends up improving the, like, you know, the resource metrics, like CPU utilization, right? A huge fraction of that, which I'll kind of drill down into a little bit more detail, whose fraction of that comes because the moment you've taken to account the resource usage of all of these different, like, you know, hosts when you're scheduling tasks, like, you know, and be intelligent to that, resource utilization can increase, or task throughput can increase, right? Where we are paying some penalty is on this thing called the resource fairness index, right? Yep, resource fairness index is smaller, but how many of you even know about resource fairness index? Right? You probably don't even care, right? The other things that would really make the most difference, but fairness is something which in the grand scheme of things is important to consider, otherwise starvation can happen, right? So on the bottom part, like, when you see some more results, the bottom, like, you know, left, what you're seeing is the overall mixpan of these four algorithms, when it says fair five, what that means, that five means is the concurrency, like, you know, number of queries, like, you know, coming in, right? What you see is mixpan, all the other techniques are better than fair. Now, one thing to keep in mind is the FIFO and the shortest remaining processing time first, they come up with big disadvantages of their own, right? FIFO can run into starvation, FIFO often actually means, like, you know, convoy effect and things like that, but FIFO also has a, like, a behavior that most people don't even think, right? Tasks in a query or tasks in a stage end up having a similar resource profile. So if you're just letting things in a FIFO fashion, you end up having to schedule tasks, you end up having to schedule tasks, which all have a similar resource profile. And then it's pretty hard to, like, you know, balance the resource usage across hosts, right? So pretty interesting things to kind of keep in mind as you really drill down into how these different algorithms behave. Like, you know, the, I'm just going to skip to the, the plot on the bottom right, which is actually showing, it's on the log scale. So it shows how the, across these different scheduling algorithms, overall compute time remains the same, but there are big, like, you know, improvements in the scheduling and resource wait times. So I want to use one more minute to wrap up. Is that, that is not okay. So, great. So that is pretty much it. So overall, I'm going to actually, like, you know, summarize by saying, this is a problem that has received a lot of attention, but there's still a lot of room to improve. And if you're interested in these sort of problems, like, you know, we are actually working on these things, not just in academia, but also in the industry. There are interesting opportunities for great algorithms to, like, you know, be applied to application performance management. Thank you. Thank you. Thank you. Thank you. Thank you. Thank you.
amount of time waiting, be it on the scheduling time. If this task scheduling is not aware of trooper, and starvation, and resource fairness, mean by offline here is a phase that is not like, latest stage in such a query, right? Find a disproportional wait time. So overall, this to understand that is MIFO is applied on top of a like, you know, the MIFO trigger time, giving it want the resource wait times spread across the the, like, you know, the graphs I show on the average resource wait time, for example, with the from the TPCDS benchmark, right? So the key or what is the mean response time of my queries, know, resource utilization, and all of that spectrum of experiments that we have done, the how much we have improved, the FAIR algorithm the completion time of the workload, like, you speedup is high, right? So those are the end-user the wait time and an attitude of, like, I want to know, be applied to application performance