[
https://issues.apache.org/jira/browse/SPARK-18886?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15765322#comment-15765322
]
Imran Rashid commented on SPARK-18886:
--------------------------------------
You understand correctly -- that is precisely what I'm proposing.
The scenario with multiple waves is a good example for why I think this is a
*good* change. If only 1% of your cluster can take advantage of locality, then
99% of your cluster goes unused across all those waves. That may be an extreme
(though a case I have actually seen in practice on large clusters). even if
its 50%, then you have 50% of your cluster going unused. Unless local tasks
are more than 2x faster, it would make more sense to make the change I'm
proposing. What's the worst case after this change? All but one executor are
local -- the result is that you have one task running slower. But the more
waves there are, the less the downside. Eg., you complete 10 waves on the
local executors, and only 8 waves on the non-local one.
The worst case is if there is only one wave, there is a huge gap (multiples Xs)
in runtime between local and non-local execution, and moments after you
schedule on non-local resources, some local resource would become available. I
think this situation is not very common -- in particular, there normally isn't
*such* an enormous gap between local and non-local that users would prefer
their non-local resources sit idle indefinitely. I'd argue that if such a use
case is important, we should add a special conf for that in particular.
> Delay scheduling should not delay some executors indefinitely if one task is
> scheduled before delay timeout
> -----------------------------------------------------------------------------------------------------------
>
> Key: SPARK-18886
> URL: https://issues.apache.org/jira/browse/SPARK-18886
> Project: Spark
> Issue Type: Bug
> Components: Scheduler
> Affects Versions: 2.1.0
> Reporter: Imran Rashid
>
> Delay scheduling can introduce an unbounded delay and underutilization of
> cluster resources under the following circumstances:
> 1. Tasks have locality preferences for a subset of available resources
> 2. Tasks finish in less time than the delay scheduling.
> Instead of having *one* delay to wait for resources with better locality,
> spark waits indefinitely.
> As an example, consider a cluster with 100 executors, and a taskset with 500
> tasks. Say all tasks have a preference for one executor, which is by itself
> on one host. Given the default locality wait of 3s per level, we end up with
> a 6s delay till we schedule on other hosts (process wait + host wait).
> If each task takes 5 seconds (under the 6 second delay), then _all 500_ tasks
> get scheduled on _only one_ executor. This means you're only using a 1% of
> your cluster, and you get a ~100x slowdown. You'd actually be better off if
> tasks took 7 seconds.
> *WORKAROUNDS*:
> (1) You can change the locality wait times so that it is shorter than the
> task execution time. You need to take into account the sum of all wait times
> to use all the resources on your cluster. For example, if you have resources
> on different racks, this will include the sum of
> "spark.locality.wait.process" + "spark.locality.wait.node" +
> "spark.locality.wait.rack". Those each default to "3s". The simplest way to
> be to set "spark.locality.wait.process" to your desired wait interval, and
> set both "spark.locality.wait.node" and "spark.locality.wait.rack" to "0".
> For example, if your tasks take ~3 seconds on average, you might set
> "spark.locality.wait.process" to "1s".
> Note that this workaround isn't perfect --with less delay scheduling, you may
> not get as good resource locality. After this issue is fixed, you'd most
> likely want to undo these configuration changes.
> (2) The worst case here will only happen if your tasks have extreme skew in
> their locality preferences. Users may be able to modify their job to
> controlling the distribution of the original input data.
> (2a) A shuffle may end up with very skewed locality preferences, especially
> if you do a repartition starting from a small number of partitions. (Shuffle
> locality preference is assigned if any node has more than 20% of the shuffle
> input data -- by chance, you may have one node just above that threshold, and
> all other nodes just below it.) In this case, you can turn off locality
> preference for shuffle data by setting
> {{spark.shuffle.reduceLocality.enabled=false}}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
