[SPARK-2774] Set preferred locations for reduce tasks

[shivaram]

Set preferred locations for reduce tasks.
The basic design is that we maintain a map from reducerId to a list of (sizes, locations) for each shuffle. We then pick the machines with the top 5 sizes and set that as preferred locations.

Picking the top5 needs sorting (well its a little cheaper than sorting, but still), so we restrict this feature to cases where we have fewer than 1000 map tasks and 1000 reduce tasks.

Resubmission of #4576 and #1697. Hopefully third time is a charm ? cc @kayousterhout for review

[JoshRosen]

Why not use the Scaladoc syntax for this?

[shivaram]

Now using scaladoc syntax

[SparkQA]

Test build #34210 has finished for PR 6652 at commit e7d5449.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public final class UnsafeRow extends BaseMutableRow
  • public abstract class BaseMutableRow extends BaseRow implements MutableRow
  • public abstract class BaseRow implements Row
  • protected class CodeGenContext
  • abstract class BaseMutableProjection extends MutableProjection
  • class SpecificProjection extends $
  • class BaseOrdering extends Ordering[Row]
  • class SpecificOrdering extends $
  • abstract class Predicate
  • class SpecificPredicate extends $
  • abstract class BaseProject extends Projection
  • class SpecificProjection extends $
  • final class SpecificRow extends $

[SparkQA]

Test build #34220 has finished for PR 6652 at commit 0df3180.

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

[kayousterhout]

is there a reason to not use "sizes.map(_._2).sum" instead here? (just slightly easier to read)

[kayousterhout]

Actually what about:

val locs = statuses.groupBy(s.location).mapValues { statusesAtLocation =>
statusesAtLocation.map(_.getSizeForBlock(r)).sum }

[shivaram]

I made this imperative now per Josh's comment below. I think the new version is pretty readable, but let me now if you think its good

[kayousterhout]

Ok this looks mostly good, subject to @JoshRosen's comments about scalability -- just a few last comments to improve readability.

[shivaram]

@kayousterhout I updated the flag name to be a bit more specific. Let me know if you have any further comments

[SparkQA]

Test build #34374 has finished for PR 6652 at commit 1090b58.

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

[SparkQA]

Test build #34375 has finished for PR 6652 at commit 68bc29e.

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

[kayousterhout]

@shivaram and I discussed this offline and whether it was possible to incorporate @sryza's point about the fact that it only makes sense to do this when there's a lot of skew in the placement of the data read by one reduce task. We came up with a simple approach that seems strictly better than the current one (it's simpler to implement and also avoids setting preferred locations when there isn't much skew in the output). The new idea is to set the preferred locations for a reducer to be any locations that hold >20% of the data to be read by that reducer. This will result in at most 5 locations being set (the current approach always chooses 5 locations), but more typically fewer. 20% is totally a magic number here...but intuitively it seems unnecessary to set a preferred location and add that complexity when there will be less than a 20% perf. improvement from doing so.

Unless anyone objects, @shivaram is going to push a new version that uses that approach.

[shivaram]

Thanks @kayousterhout for taking a look and proposing this new design. I agree that its more cleaner than the top-5 locations and we only get up to 5 locations at the maximum with fraction as 0.2. I've updated the code to reflect this now.

BTW I also removed caching the preferred locations in the map output tracker. This made it messy to implement an API which accepted fractionThreshold as an argument and I think the caching only helps in case of speculative tasks / failures (so it shouldn't hurt us much).

[kayousterhout]

OOD comment

[shivaram]

Fixed now

[kayousterhout]

Mostly LGTM -- just a few readability comments to help whatever future poor soul wants to update the DAGScheduler code.

@sryza does this new approach adequately address your concerns?

[SparkQA]

Test build #34475 has finished for PR 6652 at commit 897a914.

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

[SparkQA]

Test build #34474 has finished for PR 6652 at commit f5be578.

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

[SparkQA]

Test build #34480 has finished for PR 6652 at commit 2ef2d39.

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

[SparkQA]

Test build #34519 has finished for PR 6652 at commit 492e25e.

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

[squito]

I think you can delete this pattern now, it shouldn't ever occur.

incidentally, this also suggests that Depedendency should be sealed. I just tried making that change, it turned up these other warnings, which actually seem legit:

warn] /Users/irashid/github/spark/core/src/main/scala/org/apache/spark/rdd/CoGroupedRDD.scala:136: non-variable type argument Product2[K,Any] in type pattern org.apache.spark.OneToOneDependency[Product2[K,Any]] is unchecked since it is eliminated by erasure
[warn]       case oneToOneDependency: OneToOneDependency[Product2[K, Any]] =>
[warn]                                ^
[warn] /Users/irashid/github/spark/core/src/main/scala/org/apache/spark/rdd/CoGroupedRDD.scala:135: match may not be exhaustive.
[warn] It would fail on the following input: NarrowDependency()
[warn]     for ((dep, depNum) <- dependencies.zipWithIndex) dep match {
[warn]                                                      ^
[warn] /Users/irashid/github/spark/core/src/main/scala/org/apache/spark/rdd/SubtractedRDD.scala:109: match may not be exhaustive.
[warn] It would fail on the following input: NarrowDependency()
[warn]       dependencies(depNum) match {
[warn]                   ^
[warn] three warnings found

It seems that code will break if they are ever given a RangeDependency

[squito]

lgtm too. I think you could go ahead and delete the extra case _, though maybe it makes sense to defer changing Dependency to sealed to a separate pr.

[shivaram]

Thanks @squito for taking a look. Yeah I think it makes to change Dependency to a sealed class but we can do it in a follow up PR. BTW in the specific case of the scheduler it might be interesting to set preferred locations for RangeDependency based on the size of the parent splits. I haven't run into many applications which use RangeDependency though.

[sryza]

@shivaram @kayousterhout this approach addresses my concerns. Thanks for updating!