[SPARK-34942][API][CORE] Abstract Location in MapStatus to enable support for custom storage #31876
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,8 +21,7 @@ import java.io.{ObjectInputStream, ObjectOutputStream} | |
|
|
||
| import org.apache.spark.annotation.DeveloperApi | ||
| import org.apache.spark.internal.Logging | ||
| import org.apache.spark.scheduler.{AccumulableInfo, Location} | ||
| import org.apache.spark.util.{AccumulatorV2, Utils} | ||
|
|
||
| // ============================================================================================== | ||
|
|
@@ -81,7 +80,7 @@ case object Resubmitted extends TaskFailedReason { | |
| */ | ||
| @DeveloperApi | ||
| case class FetchFailed( | ||
| bmAddress: Location, // Note that bmAddress can be null | ||
|
Contributor
There was a problem hiding this comment. This is a backwardly incompatible change, which also impacts event files.
Member
Author
There was a problem hiding this comment. For the event files, we still log the json of Besides, do you worry about the possible backward incompatibility due to the reference out of Spark?
Contributor
There was a problem hiding this comment. There are two changes here: a) change to public api (programmatic, REST api, etc), b) change to generated/persisted data. Specifically about (b), with Spark history server, REST consumers, other apps depending on event files - will need a way to identify Location type and serde all valid Location types.
Member
Author
There was a problem hiding this comment.
Technically, it's really a problem. Although, I can't imagine how users would use it as an API. And I have a new idea that we can introduce a new fetch failed class for the custom location and leave this one unchanged. For example, we can have
I think we won't change the data of The only problem is the custom location. It's new data, e.g.,
Contributor
There was a problem hiding this comment.
There are couple of issues here:
I actually dont have good solutions on this - other than adding some metadata per location record to indicate the 'type'.
Member
There was a problem hiding this comment. +1 for a new class without introducing a breaking change. Thank you for pinging me, @mridulm .
Member
Author
There was a problem hiding this comment. I think we'll have to add interfaces to public JValue serializeToJson();
public Location deserilaizeFromJson(json: JValue);
Adding metadata is good idea, we can have the format like, "Mapstatus Location": {
"type": "xx.yy.zz", // qualified class name
"content": { // content is generate by `Location.serializeToJson`
"aaa":"bbb"
}
}with the constant format, end-users and SHS are able to parse them as well. (I had an idea about SHS is to add the location type as the extension of the event log file. That's the way what compression does now. But I think it doesn't solve the problem of REST case.) BTW, I have added https://issues.apache.org/jira/browse/SPARK-35188 for this support. |
||
| shuffleId: Int, | ||
| mapId: Long, | ||
| mapIndex: Int, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -28,16 +28,23 @@ import org.apache.spark.internal.config | |
| import org.apache.spark.storage.BlockManagerId | ||
| import org.apache.spark.util.Utils | ||
|
|
||
| trait Location extends Externalizable { | ||
|
Member
Author
There was a problem hiding this comment.
There was a problem hiding this comment. Original BlockManagerId extends
Contributor
There was a problem hiding this comment. I might have missed some context here, but why
Member
Author
There was a problem hiding this comment. Because the A possible way to let
Contributor
There was a problem hiding this comment. Using
Contributor
There was a problem hiding this comment. To clarify: |
||
| def host: String | ||
| def port: Int | ||
| def hostPort: String | ||
| def executorId: String = "unknown" | ||
|
Member
Author
There was a problem hiding this comment. I added this only for the convenient purpose, doesn't mean I have any preference for the interface. There was a problem hiding this comment. Yeah, adding this here helps for this initial version. Could we add something like
Member
Author
There was a problem hiding this comment. For "store extra information", I think implementors can add whatever they want only if they're serializable. e.g., extra info can be But I did think about adding a common There was a problem hiding this comment. Or could we add
Contributor
There was a problem hiding this comment. Based on discussion, if we are going to mirror everything in
Member
Author
There was a problem hiding this comment. I think it (exposing Adding
Member
Author
There was a problem hiding this comment.
Do you know other cases that need it? There was a problem hiding this comment. Uber Remote Shuffle Service uses topologyInfo to store shuffle server information. Since we kind of mirror things in
Contributor
There was a problem hiding this comment. I think in |
||
| } | ||
|
|
||
| /** | ||
| * Result returned by a ShuffleMapTask to a scheduler. Includes the block manager address that the | ||
| * task has shuffle files stored on as well as the sizes of outputs for each reducer, for passing | ||
| * on to the reduce tasks. | ||
| */ | ||
| private[spark] sealed trait MapStatus { | ||
| /** Location where this task output is. */ | ||
| def location: Location | ||
|
|
||
| def updateLocation(newLoc: Location): Unit | ||
|
|
||
| /** | ||
| * Estimated size for the reduce block, in bytes. | ||
|
|
@@ -66,7 +73,7 @@ private[spark] object MapStatus { | |
| .getOrElse(config.SHUFFLE_MIN_NUM_PARTS_TO_HIGHLY_COMPRESS.defaultValue.get) | ||
|
|
||
| def apply( | ||
| loc: Location, | ||
| uncompressedSizes: Array[Long], | ||
| mapTaskId: Long): MapStatus = { | ||
| if (uncompressedSizes.length > minPartitionsToUseHighlyCompressMapStatus) { | ||
|
|
@@ -115,21 +122,21 @@ private[spark] object MapStatus { | |
| * @param _mapTaskId unique task id for the task | ||
| */ | ||
| private[spark] class CompressedMapStatus( | ||
| private[this] var loc: Location, | ||
| private[this] var compressedSizes: Array[Byte], | ||
| private[this] var _mapTaskId: Long) | ||
| extends MapStatus with Externalizable { | ||
|
|
||
| // For deserialization only | ||
| protected def this() = this(null, null.asInstanceOf[Array[Byte]], -1) | ||
|
|
||
| def this(loc: Location, uncompressedSizes: Array[Long], mapTaskId: Long) = { | ||
| this(loc, uncompressedSizes.map(MapStatus.compressSize), mapTaskId) | ||
| } | ||
|
|
||
| override def location: Location = loc | ||
|
|
||
| override def updateLocation(newLoc: Location): Unit = { | ||
| loc = newLoc | ||
| } | ||
|
|
||
|
|
@@ -168,7 +175,7 @@ private[spark] class CompressedMapStatus( | |
| * @param _mapTaskId unique task id for the task | ||
| */ | ||
| private[spark] class HighlyCompressedMapStatus private ( | ||
| private[this] var loc: Location, | ||
| private[this] var numNonEmptyBlocks: Int, | ||
| private[this] var emptyBlocks: RoaringBitmap, | ||
| private[this] var avgSize: Long, | ||
|
|
@@ -183,9 +190,9 @@ private[spark] class HighlyCompressedMapStatus private ( | |
|
|
||
| protected def this() = this(null, -1, null, -1, null, -1) // For deserialization only | ||
|
|
||
| override def location: Location = loc | ||
|
|
||
| override def updateLocation(newLoc: Location): Unit = { | ||
| loc = newLoc | ||
| } | ||
|
|
||
|
|
@@ -216,7 +223,10 @@ private[spark] class HighlyCompressedMapStatus private ( | |
| } | ||
|
|
||
| override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException { | ||
| // TODO(wuyi): config | ||
| val location = "org.apache.spark.storage.BlockManagerId" | ||
| loc = Utils.classForName(location).newInstance().asInstanceOf[Location] | ||
| loc.readExternal(in) | ||
|
Ngone51 marked this conversation as resolved.
Outdated
|
||
| numNonEmptyBlocks = -1 // SPARK-32436 Scala 2.13 doesn't initialize this during deserialization | ||
| emptyBlocks = new RoaringBitmap() | ||
| emptyBlocks.deserialize(in) | ||
|
|
@@ -235,7 +245,7 @@ private[spark] class HighlyCompressedMapStatus private ( | |
|
|
||
| private[spark] object HighlyCompressedMapStatus { | ||
| def apply( | ||
| loc: Location, | ||
| uncompressedSizes: Array[Long], | ||
| mapTaskId: Long): HighlyCompressedMapStatus = { | ||
| // We must keep track of which blocks are empty so that we don't report a zero-sized | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.