[SPARK-2044] Pluggable interface for shuffles #1009

mateiz · 2014-06-08T06:07:14Z

This is a first cut at moving shuffle logic behind a pluggable interface, as described at https://issues.apache.org/jira/browse/SPARK-2044, to let us more easily experiment with new shuffle implementations. It moves the existing shuffle code to a class HashShuffleManager behind a general ShuffleManager interface.

Two things are still missing to make this complete:

MapOutputTracker needs to be hidden behind the ShuffleManager interface; this will also require adding methods to ShuffleManager that will let the DAGScheduler interact with it as it does with the MapOutputTracker today
The code to do map-sides and reduce-side combine in ShuffledRDD, PairRDDFunctions, etc needs to be moved into the ShuffleManager's readers and writers

However, some of these may also be done later after we merge the current interface.

AmplabJenkins · 2014-06-08T06:07:49Z

Merged build triggered.

AmplabJenkins · 2014-06-08T06:07:58Z

Merged build started.

AmplabJenkins · 2014-06-08T06:09:00Z

Merged build finished.

AmplabJenkins · 2014-06-08T06:09:00Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15536/

mateiz · 2014-06-08T06:12:12Z

Jenkins, retest this please

AmplabJenkins · 2014-06-08T06:12:49Z

Merged build triggered.

AmplabJenkins · 2014-06-08T06:12:58Z

Merged build started.

AmplabJenkins · 2014-06-08T06:53:50Z

Merged build finished.

AmplabJenkins · 2014-06-08T06:53:51Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15537/

jerryshao · 2014-06-08T06:57:39Z

Hi Matei, since we already have ShuffleWriter and ShuffleManager, do we still need to keep ShuffleBlockManager, I think two functionalities of ShuffleBlockManager as getting shuffle writer and managing shuffle block location can separately move into HashShuffleWriter and HashShuffleManager or some others. I think it will be more clear and loose-couple for the shuffle module and BlockManager.

aarondav · 2014-06-08T07:17:11Z

core/src/main/scala/org/apache/spark/shuffle/hash/HashShuffleReader.scala

+    context: TaskContext)
+  extends ShuffleReader[K, C]
+{
+  require(endPartition == startPartition + 1,


Who enforces this? I thought the API allows users to create a ShuffleReader with an arbitrary partition range, and it's up to the ShuffleReader to implement it correctly (perhaps by chunking itself up).

Yup, this is just not yet supported in the first pass.

(The reason is that for this to work we need to move aggregation into the ShuffleReader.)

AmplabJenkins · 2014-06-10T19:02:06Z

Merged build triggered.

AmplabJenkins · 2014-06-10T19:02:12Z

Merged build started.

AmplabJenkins · 2014-06-10T19:48:06Z

Merged build finished.

AmplabJenkins · 2014-06-10T19:48:06Z

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15631/

andrewor14 · 2014-06-11T00:31:36Z

core/src/main/scala/org/apache/spark/shuffle/hash/HashShuffleWriter.scala

+    shuffleMetrics.shuffleWriteTime = totalTime
+    metrics.shuffleWriteMetrics = Some(shuffleMetrics)
+
+    success = true


I don't think this is used anywhere?

mateiz · 2014-06-12T01:10:21Z

Thanks Andrew for the comments. @pwendell @rxin I think this is good to merge now if it passes tests because other people want to contribute some other things (e.g. moving aggregation into the shuffle implementation) and I'd like to let them do that in parallel.

AmplabJenkins · 2014-06-12T01:12:07Z

Merged build triggered.

AmplabJenkins · 2014-06-12T01:12:17Z

Merged build started.

AmplabJenkins · 2014-06-12T01:55:18Z

Merged build finished. All automated tests passed.

AmplabJenkins · 2014-06-12T01:55:18Z

All automated tests passed.
Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15699/

rxin · 2014-06-12T03:45:16Z

Ok merging this in master. Thanks!

mateiz · 2014-06-12T04:32:50Z

Thanks Reynold. I've now moved the other TODOs into sub-tasks on https://issues.apache.org/jira/browse/SPARK-2044.

cloud-fan · 2014-06-12T06:15:21Z

core/src/main/scala/org/apache/spark/rdd/ShuffledRDD.scala

  def setSerializer(serializer: Serializer): ShuffledRDD[K, V, P] = {
-    this.serializer = serializer
+    this.serializer = Option(serializer)


should this be Some(serializer)?

Not if serializer is null

for null it will be None, isn't that the expect result?

scala> var serializer: Option[java.lang.Integer] = null serializer: Option[Integer] = null scala> serializer = Some(1) serializer: Option[Integer] = Some(1) scala> serializer.get res0: Integer = 1 scala> serializer = Some(null) serializer: Option[Integer] = Some(null) scala> serializer.get res1: Integer = null scala> serializer.isDefined res2: Boolean = true scala> serializer = Option(null) serializer: Option[Integer] = None scala> serializer.isDefined res3: Boolean = false

This is a first cut at moving shuffle logic behind a pluggable interface, as described at https://issues.apache.org/jira/browse/SPARK-2044, to let us more easily experiment with new shuffle implementations. It moves the existing shuffle code to a class HashShuffleManager behind a general ShuffleManager interface. Two things are still missing to make this complete: * MapOutputTracker needs to be hidden behind the ShuffleManager interface; this will also require adding methods to ShuffleManager that will let the DAGScheduler interact with it as it does with the MapOutputTracker today * The code to do map-sides and reduce-side combine in ShuffledRDD, PairRDDFunctions, etc needs to be moved into the ShuffleManager's readers and writers However, some of these may also be done later after we merge the current interface. Author: Matei Zaharia <matei@databricks.com> Closes apache#1009 from mateiz/pluggable-shuffle and squashes the following commits: 7a09862 [Matei Zaharia] review comments be33d3f [Matei Zaharia] review comments 1513d4e [Matei Zaharia] Add ASF header ac56831 [Matei Zaharia] Bug fix and better error message 4f681ba [Matei Zaharia] Move write part of ShuffleMapTask to ShuffleManager f6f011d [Matei Zaharia] Move hash shuffle reader behind ShuffleManager interface 55c7717 [Matei Zaharia] Changed RDD code to use ShuffleReader 75cc044 [Matei Zaharia] Partial work to move hash shuffle in

mateiz added 5 commits June 5, 2014 22:09

Partial work to move hash shuffle in

75cc044

Changed RDD code to use ShuffleReader

55c7717

Move hash shuffle reader behind ShuffleManager interface

f6f011d

Move write part of ShuffleMapTask to ShuffleManager

4f681ba

Bug fix and better error message

ac56831

Add ASF header

1513d4e

aarondav reviewed Jun 8, 2014
View reviewed changes

review comments

be33d3f

andrewor14 reviewed Jun 11, 2014
View reviewed changes

review comments

7a09862

asfgit closed this in 508fd37 Jun 12, 2014

cloud-fan reviewed Jun 12, 2014
View reviewed changes

wangyum pushed a commit that referenced this pull request May 26, 2023

[CARMEL-6087] User is continually blocked by circuit breaker (#1009)

de7f557

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-2044] Pluggable interface for shuffles #1009

[SPARK-2044] Pluggable interface for shuffles #1009

mateiz commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

mateiz commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

jerryshao commented Jun 8, 2014

aarondav Jun 8, 2014

mateiz Jun 8, 2014

mateiz Jun 8, 2014

AmplabJenkins commented Jun 10, 2014

AmplabJenkins commented Jun 10, 2014

AmplabJenkins commented Jun 10, 2014

AmplabJenkins commented Jun 10, 2014

andrewor14 Jun 11, 2014

mateiz commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

rxin commented Jun 12, 2014

mateiz commented Jun 12, 2014

cloud-fan Jun 12, 2014

rxin Jun 12, 2014

colorant Jun 12, 2014

rxin Jun 12, 2014

[SPARK-2044] Pluggable interface for shuffles #1009

[SPARK-2044] Pluggable interface for shuffles #1009

Conversation

mateiz commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

mateiz commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

AmplabJenkins commented Jun 8, 2014

jerryshao commented Jun 8, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AmplabJenkins commented Jun 10, 2014

AmplabJenkins commented Jun 10, 2014

AmplabJenkins commented Jun 10, 2014

AmplabJenkins commented Jun 10, 2014

Choose a reason for hiding this comment

mateiz commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

AmplabJenkins commented Jun 12, 2014

rxin commented Jun 12, 2014

mateiz commented Jun 12, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment