[SPARK-18886][CORE] Make Locality wait time measure resource under utilization due to delay scheduling.

[bmarcott]

What changes were proposed in this pull request?

Delay scheduling is an optimization that sacrifices fairness for data locality in order to improve cluster and workload throughput.

One useful definition of "delay" here is how much time has passed since the TaskSet was using its fair share of resources.

However it is impractical to calculate this delay, as it would require running simulations assuming no delay scheduling. Tasks would be run in different orders with different run times.

Currently the heuristic used to estimate this delay is the time since a task was last launched for a TaskSet. The problem is that it essentially does not account for resource utilization, potentially leaving the cluster heavily underutilized.

This PR modifies the heuristic in an attempt to move closer to the useful definition of delay above.
The newly proposed delay is the time since a TasksSet last launched a task and did not reject any resources due to delay scheduling when offered its "fair share".

See the last comments of #26696 for more discussion.

Why are the changes needed?

cluster can become heavily underutilized as described in SPARK-18886

How was this patch tested?

TaskSchedulerImplSuite

@cloud-fan
@tgravescs
@squito

[cloud-fan]

ok to test

[cloud-fan]

can we ask each TSM to track it? We can pass the isAllFreeResources parameter to TSM when calling resourceOffer

[bmarcott]

what do you see as pros/cons of keeping this map vs putting in TSM?
The TaskSchedulerImpl is the only one setting/updating the variable based on knowledge only it has (no schedule delay reject).

[SparkQA]

Test build #116771 has finished for PR 27207 at commit d52de6b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[tgravescs]

so please update the description with information from the other PR. The description should have basically a high level design of this approach with enough details before someone reads the code to make sure the code is doing what you are proposing. What cases it covers and what cases you know it doesn't.

So the case you mentioned it doesn't cover:

The case I am referring to is: imagine you have 2 resources and an "all resource offer" is scheduled every second. when TSM1 is submitted, it'll also get an "all resource offer", and assume it rejects both, causing a prexisting TSM2 to utilize them. Assume those 2 tasks finish, and the freed resources are offered one by one to TSM1, which accepts both, all within 1 second (before any "all resource offer"). This should reset the timer, but it won't in the implementation.

So the issue here is that we aren't really tracking when all resources are used we are proxying that.
To really calculate the free slots though is pretty complex when you take into account blacklisting (have both application and taskset level).
I'm kind of thinking at this point the above case is ok, it favors not delaying and it will be fixed up on the next "all resource offer"

One thing I don't think I like is that if you are fully scheduled, we keep trying to schedule "all resources" but if there are no resources, then we continue to reset the timer. This means that it takes a long time to fall back in case where you may have multiple tasksets and the first task set rejects it and the second one takes it and the tasks are finishing such that you get an all resources offer in between the task finishes. In this scenario the first taskset can get starved. We would need to perhaps track this separately.

I need to take another walk through all the scenarios again as well

[tgravescs]

nit - I think the variable names are a bit confusing because you have noDelay and then hadAnyDelay, I think it would be easier to read if they were the same.

[bmarcott]

@tgravescs
Thanks for the comments.

so please update the description with information from the other PR

Which one of my snippets from the previous PR was most clear to you? I can put that one in the description.

One thing I don't think I like....

Really good point on this scenario. It's bad even for the same scenario that you described but where the all resource offer has only 1 executor, and the first taskset accepts it.
Let me know if you have any good ideas here 😉

Is it ok I do follow up changes, such as variable names, unit tests, and other backend schedulers only once we iron out the problematic scenarios?

[tgravescs]

yes its fine not to do any minor changes until we decide on design.
I'm not so worried about the 1 executor case if the first taskset takes it because that one should be higher priority and that should work as expected, when there isn't anywhere to put the tasks, then have them wait a bit to try to get locality. Its the case with the fair scheduler where the one that isn't the highest priority that I'm more concerned with.

I wonder if we can add in a separate tracking/check in TaskSchedulerImpl that tracks to see if it keeps rejecting it on the non all resource offers but then resets on the all offers, then after some number of those we stop resetting it, thoughts on that?

[bmarcott]

That may be reasonable to do, but I'd like to avoid adding more tracking/accounting if possible. I already don't like the boolean map I added.
What do you think about adding back the old condition of "must have launched a task".
so the new condition for reset would be must have no rejects and launch a task on an all resource offer
there is starvation still, but should be no more starvation than current master code I believe.

[cloud-fan]

I'm fine with a non-perfect solution as long as it's no worse than the existing one and fixes the local wait problem.

[tgravescs]

So the conditions would be like:

offer 1 resource that was rejected - no timer reset
offer all resources with no rejects and launched a task - timer is reset
offer 1 resource, no reject - timer is reset
offer 1 resource that was rejected - no timer reset
offer 1 resource, no reject - no timer reset because previous offer was rejected

I think that makes sense, it definitely addresses the case I was talking about with 2 task sets.

I was also wondering about leaving the old logic in there but configured off by default. While I don't like it, it would be safer if we haven't thought of some corner case and user could turn it back on if necessary, thoughts?

[bmarcott]

@cloud-fan
thanks for the input

@tgravescs
yep, that sequence and explanation matches with my understanding

I was also wondering about leaving the old logic in there but configured off by default. While I don't like it, it would be safer if we haven't thought of some corner case and user could turn it back on if necessary, thoughts?

I'm not opposed to adding a new config that disables new way and enables old way, but is this standard practice in spark for the more risky changes?

I would feel more comfortable knowing there was a fallback too ;)

[tgravescs]

its a case by case basis. While we don't like adding more configs and code for maintenance, if its somewhat risky change and with all the corner cases and different ways to run I think its warranted. Actually what we can do is leave the config undocumented and default to new algorithm and put a deprecated message by it so we can remove it in like a 3.1 if we want.

[tgravescs]

@bmarcott will you be able to update this?

[bmarcott]

yea, I'll try to take a look this weekend. Thanks for checking in.

[bmarcott]

@tgravescs
I updated with the new config to switch to legacy behavior as well as some doc/variable renaming.
I also updated the description of this PR.

After reading more through SchedulerBackend impls:

1. decided to assume resource offers are full resource by default, to match previous behavior for any SchedulerBackends not touched in this PR.
2. I only found MesosFineGrainedSchedulerBackend which doesn't use CoarseGrainedSchedulerBackend and decided not to touch it since it is deprecated and from the design looks like "all free resources" can't be tracked.

Test updates are still needed, but wanted to get your feedback first.

[SparkQA]

Test build #117728 has finished for PR 27207 at commit e0ac12e.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #117727 has finished for PR 27207 at commit 2924c5b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[Ngone51]

This makes me feel that we're trying to put cluster utilized before data locality while assigning resources. Right?

[bmarcott]

@Ngone51 thanks for taking a look.
I wouldn't say it puts one thing before another. The idea is to move closer to what a reasonable definition of scheduling delay (locality.wait time) is: how long you want to sacrifice "fairness" (using your fair share of resources) in favor of data locality.

I'll do the touch ups you mentioned after there is enough support on the design.

[cloud-fan]

The newly proposed delay is the time since a TasksSet last launched a task and did not reject any resources due to delay scheduling when offered its "fair share".

Thanks for updating the PR description! The proposal makes sense to me.

Is it possible to centralize the delay scheduling code? Now it's in both TaskSchedulerImpl and TaskSetManager, which makes it a bit hard to understand as you need to think about the interactions between them.

[tgravescs]

I'll take a closer look later, I think its fine not to support mesos fine grain scheduler

[bmarcott]

I appreciate everyones feedback. I've updated some nits locally and will follow up on tests later.

[bmarcott]

👍
What are your thoughts on:
I'm thinking instead of making assumptions about what taskDescription.isEmpty means, maybe it'd be better to pass maxLocality into dequeueTask and then change its logic near the bottom to be something like:

    if (TaskLocality.isAllowed(allowedLocality, TaskLocality.ANY)) {
      dequeue(pendingTaskSetToUse.all).foreach { index =>
        return Some((index, TaskLocality.ANY, speculative))
      }
    } else if (maxLocality == TaskLocality.ANY && pendingTasks.all.nonEmpty) {
      hasReject = true
    }

[bmarcott]

@cloud-fan

Is it possible to centralize the delay scheduling code? Now it's in both TaskSchedulerImpl and TaskSetManager, which makes it a bit hard to understand as you need to think about the interactions between them.

I am not sure a good way to centralize because

1. The TSM is called multiple times with various offers and we need to keep track of what happened across those calls
2. TSM today doesn't differentiate whether it didn't launch a task due to blacklisting or due to delay scheduling, hence the new boolean returned.

[SparkQA]

Test build #118111 has finished for PR 27207 at commit 61671b3.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #118113 has finished for PR 27207 at commit f341ebb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #118214 has finished for PR 27207 at commit e39680b.

• This patch fails to generate documentation.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #118223 has finished for PR 27207 at commit 894bebb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #120771 has finished for PR 27207 at commit 24c8ad9.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[tgravescs]

@bmarcott accepted your linked in but we can discuss here as well. what all manual testing have you done? I was hoping to do some myself but haven't had time yet. We can continue to test after its merged in.

[bmarcott]

I have not tested this manually yet, nor have played around with using multiple executors locally before.
Maybe I can just use spark standalone: https://spark.apache.org/docs/latest/spark-standalone.html

[bmarcott]

running into problems running standalone. posted question on stack

....updated...
was using wrong port. Answered my own question

[tgravescs]

@bmarcott did you get it to work on standalone cluster?

[bmarcott]

@tgravescs I did setup everything and can run jobs. Now I am trying to create a use case where the tasks prefer a particular executor. Have any ideas on a way to easily repro this issue?

[cloud-fan]

Can you create a custom RDD which generates random data and set the preferred location to one executor?

[bmarcott]

Thanks @cloud-fan
That is the approach I was going to start with.
In the wild/production I hit this issue with a join, but I forget the exact reason/context.

[bmarcott]

First manual test looks good. Ran a simple app which creates 1000 partitions all preferring executor 0.
Each task sleeps 1 second. Here is the app's code .
New code utilized all executors, whereas with legacy flag enabled, only process local tasks were run, making it much slower.

Run with new command:
../../projects/apache/spark/bin/spark-submit --class "TestLocalityWait" --master spark://localhost:7077 --conf spark.executor.instances=4 --conf spark.executor.cores=2 target/scala-2.12/simple-project_2.12-1.0.jar

Processed 326 partitions in 35 seconds:

Many tasks run at ANY locality level:

---

Run with legacy flag set to true:
../../projects/apache/spark/bin/spark-submit --class "TestLocalityWait" --master spark://localhost:7077 --conf spark.executor.instances=4 --conf spark.executor.cores=2 --conf spark.locality.wait.legacyResetOnTaskLaunch=true target/scala-2.12/simple-project_2.12-1.0.jar

Processed 146 partitions in 1.3 min:

All tasks run at PROCESS_LOCAL locality level

[cloud-fan]

Thanks for the manual testing! Great Job!

Merging to master!

[cloud-fan]

It conflicts with 3.0, can you send a new PR for 3.0?

[dongjoon-hyun]

Hi, All.
The last test was 7 days ago. This causes a UT failure on all master Jenkins jobs which is added by another PR. I made a follow-up to recover master branch.

• [SPARK-18886][CORE][TESTS][FOLLOWUP] Fix a test failure due to InvalidUseOfMatchersException #28174

[cloud-fan]

thanks for fixing!

[bmarcott]

@cloud-fan I thought 3.0 was already branched and frozen? Will open a PR for it if I am given the go ahead.
Thanks for all the review and feedback along the way (@tgravescs as well!)

I am further manual testing and finding either I'm testing wrong, or locality wait isn't being respected even when I set spark.locality.wait higher . Looking into it...(may have several day delay)
I believe I may need to add a special case for when there is an all resource offer with 0 offers (there will be no launched task, so no reset).

[cloud-fan]

This is a high-value perf fix, but seems too risky for 3.0 after a hindsight. @tgravescs are you OK with having this fix in master only?

[tgravescs]

yeah at this point I'm fine with leaving it in master only.

[dongjoon-hyun]

Hi, @cloud-fan . If the decision is final, could you resolve SPARK-18886 with the fixed version, 3.1.0? The JIRA issue is still open. Thanks.

[cloud-fan]

done

[gatorsmile]

Can we update the description of this function and explain the parameter "isAllFreeResources"?

[bmarcott]

yes I can add something like: if true, then the parameter offers contains all workers and their free resources. See delay scheduling comments in class description.

[tgravescs]

is there are changes suggested and required, please file a separate new jira for this and link them. this pr has been merged and we have had to many followups at this point.

[gatorsmile]

Do we need to set the default value?

[bmarcott]

I think I originally did this to not break the api + maintain something closer to previous behavior for callers who hadn't migrated to setting it to false.
Lemme know if this is the wrong approach.