Github user markhamstra commented on a diff in the pull request:
https://github.com/apache/spark/pull/126#discussion_r10586047
--- Diff:
core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala ---
@@ -17,28 +17,24 @@
package org.apache.spark.scheduler
+import scala.collection.mutable.HashMap
+
import java.io._
import java.util.zip.{GZIPInputStream, GZIPOutputStream}
-import scala.collection.mutable.HashMap
-
import org.apache.spark._
import org.apache.spark.executor.ShuffleWriteMetrics
-import org.apache.spark.rdd.RDD
-import org.apache.spark.rdd.RDDCheckpointData
+import org.apache.spark.rdd.{RDD, RDDCheckpointData}
import org.apache.spark.storage._
-import org.apache.spark.util.{MetadataCleaner, MetadataCleanerType,
TimeStampedHashMap}
+import org.apache.spark.util.BoundedHashMap
private[spark] object ShuffleMapTask {
// A simple map between the stage id to the serialized byte array of a
task.
// Served as a cache for task serialization because serialization can be
// expensive on the master node if it needs to launch thousands of tasks.
- val serializedInfoCache = new TimeStampedHashMap[Int, Array[Byte]]
-
- // TODO: This object shouldn't have global variables
- val metadataCleaner = new MetadataCleaner(
- MetadataCleanerType.SHUFFLE_MAP_TASK,
serializedInfoCache.clearOldValues, new SparkConf)
+ val MAX_CACHE_SIZE = 100
+ val serializedInfoCache = new BoundedHashMap[Int,
Array[Byte]](MAX_CACHE_SIZE, true)
--- End diff --
"This is because by the time the dependency or RDD goes out of scope, the
stage will already have been removed."
Right, but do be aware that it doesn't work the other way around. A stage
and stageId can be created and associated with a ShuffleDependency when a job
runs, then that stage and stageId can disappear from the DAGScheduler when the
job completes (finished, canceled or aborted); but metadata, cached data, etc.
for the associated ShuffleDependency should stick around as long as that
ShuffleDependency is in scope, since DAGScheduler#newOrUsedStage will want to
make use of prior mapOutputs (now associated with a fresh stageId) when it can
instead of forcing re-evaluation of those results.
Just because one job and stage is done with a shuffleDep, and as long as
that shuffleDep is in scope from some RDD, that doesn't me that another job
will not want to make use of that shuffleDep.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
