diff --git a/core/src/main/scala/org/apache/spark/BlockTtlCleaner.scala b/core/src/main/scala/org/apache/spark/BlockTtlCleaner.scala
new file mode 100644
index 0000000000000..cf6e18b3bb18c
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/BlockTtlCleaner.scala
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark
+
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.Logging
+
+/**
+ * Shared periodic TTL sweep for a block/shuffle access-time map. Finds ids whose recorded access
+ * time is older than `ttlMillis`, reaps them, and sleeps until the next possible expiry. Used by
+ * both the RDD-cache TTL cleaner (in `BlockManagerMasterEndpoint`) and the shuffle TTL cleaner (in
+ * `MapOutputTrackerMaster`); they differ only in the map, the `shouldReap` gate, and the `reap`
+ * action, so the loop lives here once.
+ *
+ * The runnable loops until interrupted (its owner interrupts it via `shutdownNow` on stop).
+ *
+ * @param name label used in log messages (e.g. "RDD" / "shuffle")
+ * @param ttlMillis the TTL; only constructed when the corresponding config is set
+ * @param accessTimes id -> last-access-time (millis). Must be a `ConcurrentHashMap`: this sweep
+ * iterates it (weakly consistent) while other threads `put` to it, and plain
+ * HashMap structural mutation from multiple threads can corrupt the map.
+ * @param shouldReap gate checked before removal; return false to leave an id tracked this pass
+ * (e.g. a shuffle that has not produced output yet has nothing to reclaim, and
+ * removing its state would break a later registration).
+ * @param reap performs the actual removal for an id whose atime was still stale.
+ */
+private[spark] class BlockTtlCleaner(
+ name: String,
+ ttlMillis: Long,
+ accessTimes: ConcurrentHashMap[Int, Long],
+ shouldReap: Int => Boolean,
+ reap: Int => Unit) extends Runnable with Logging {
+
+ override def run(): Unit = {
+ try {
+ while (!Thread.currentThread().isInterrupted) {
+ val maxAge = System.currentTimeMillis() - ttlMillis
+ // Track the oldest still-live atime so we can sleep until the next possible expiry.
+ var oldest = System.currentTimeMillis()
+ val toBeRemoved = accessTimes.asScala.toList.flatMap { case (id, atime) =>
+ if (atime < maxAge) {
+ Some((id, atime))
+ } else {
+ if (atime < oldest) {
+ oldest = atime
+ }
+ None
+ }
+ }
+ toBeRemoved.foreach { case (id, atime) =>
+ try {
+ // `shouldReap` is checked before the removal so a skipped id stays tracked (its atime
+ // is unchanged). `remove(key, value)` only succeeds if the atime is unchanged since the
+ // snapshot, so a concurrent access in the window leaves the entry: it is back in use.
+ if (shouldReap(id) && accessTimes.remove(id, atime)) {
+ reap(id)
+ }
+ } catch {
+ // Warn, not debug: this loop's whole value is reclaiming space, so a reap that always
+ // fails (e.g. an unwired remover, or an RPC failure) must not be invisible at the
+ // default log level. The id has already been dropped from `accessTimes`, so it is not
+ // retried -- a persistent failure means that id is simply never reclaimed.
+ case NonFatal(e) =>
+ logWarning(s"Error reaping $id in the $name TTL cleaner", e)
+ }
+ }
+ // Wait until the next possible element to be removed.
+ val delay = math.max((oldest + ttlMillis) - System.currentTimeMillis(), 100)
+ Thread.sleep(delay)
+ }
+ logInfo(s"$name TTL cleaner thread interrupted, exiting.")
+ } catch {
+ case _: InterruptedException =>
+ logInfo(s"$name TTL cleaner thread interrupted, exiting.")
+ }
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/ContextCleaner.scala b/core/src/main/scala/org/apache/spark/ContextCleaner.scala
index 6c1b49157cc01..7224100a80907 100644
--- a/core/src/main/scala/org/apache/spark/ContextCleaner.scala
+++ b/core/src/main/scala/org/apache/spark/ContextCleaner.scala
@@ -124,6 +124,18 @@ private[spark] class ContextCleaner(
listeners.add(listener)
}
+ /**
+ * Notify listeners that a shuffle's data was reclaimed by something other than the
+ * reference-tracking path -- specifically the shuffle TTL cleaner, which removes the shuffle
+ * itself and so must issue this fan-out on its own. Without it `ExecutorMonitor` never sees the
+ * shuffle go away, so under `spark.dynamicAllocation.shuffleTracking.enabled` an executor holding
+ * only that shuffle would never become idle and the reclaimed space would not release the
+ * executor.
+ */
+ private[spark] def notifyShuffleCleaned(shuffleId: Int): Unit = {
+ listeners.asScala.foreach(_.shuffleCleaned(shuffleId))
+ }
+
/** Start the cleaner. */
def start(): Unit = {
cleaningThread.setDaemon(true)
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index 933da94772725..5ecde568d2faf 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -801,6 +801,24 @@ private[spark] class MapOutputTrackerMaster(
private[spark] val isLocal: Boolean)
extends MapOutputTracker(conf) with ShuffleOutputTrackerMaster {
+ // Keep track of last access times for shuffle based TTL. We don't care about overwriting times
+ // that are "close", but this is written concurrently by the (multi-threaded) map-output
+ // dispatcher, the DAGScheduler, and the TTL cleaner thread, so it must be a concurrent map:
+ // plain HashMap structural mutation from multiple threads can corrupt the map, not merely skew
+ // a timestamp.
+ private[spark] val shuffleAccessTime = new ConcurrentHashMap[Int, Long]
+
+ // Hook used by the shuffle TTL cleaner to reclaim a shuffle's on-disk blocks through the normal
+ // removal path (ShuffleDriverComponents.removeShuffle -> RemoveShuffle RPC to executors/ESS).
+ // Wired by SparkContext, which owns the ShuffleDriverComponents; the tracker itself is created in
+ // SparkEnv before those exist. When unset the cleaner still unregisters the driver-side status
+ // but cannot delete executor disk, so this must be wired for the shuffle TTL to reclaim space.
+ @volatile private[spark] var shuffleFileRemover: Option[Int => Unit] = None
+
+ // Cache the (immutable-after-start) shuffle TTL config once rather than re-parsing the time
+ // string on every updateShuffleAtime, which is on the hot path of serving map-output requests.
+ private val shuffleTtl: Option[Long] = conf.get(SPARK_TTL_SHUFFLE_BLOCK_CLEANER)
+
// The size at which we use Broadcast to send the map output statuses to the executors
private val minSizeForBroadcast = conf.get(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST).toInt
@@ -841,6 +859,16 @@ private[spark] class MapOutputTrackerMaster(
private val pushBasedShuffleEnabled = Utils.isPushBasedShuffleEnabled(conf, isDriver = true)
+ // The cleaner daemon is started at the end of construction (after the fail-fast broadcast-size
+ // check below), not here, so a failed construction can't orphan the thread.
+ private[spark] val cleanerThreadpool: Option[ThreadPoolExecutor] = {
+ if (shuffleTtl.isDefined) {
+ Some(ThreadUtils.newDaemonFixedThreadPool(1, "map-output-ttl-cleaner"))
+ } else {
+ None
+ }
+ }
+
// Thread pool used for handling map output status requests. This is a separate thread pool
// to ensure we don't block the normal dispatcher threads.
private val threadpool: ThreadPoolExecutor = {
@@ -854,6 +882,12 @@ private[spark] class MapOutputTrackerMaster(
private val availableProcessors = Runtime.getRuntime.availableProcessors()
+ def updateShuffleAtime(shuffleId: Int): Unit = {
+ if (shuffleTtl.isDefined) {
+ shuffleAccessTime.put(shuffleId, System.currentTimeMillis())
+ }
+ }
+
// Make sure that we aren't going to exceed the max RPC message size by making sure
// we use broadcast to send large map output statuses.
if (minSizeForBroadcast > maxRpcMessageSize) {
@@ -865,6 +899,40 @@ private[spark] class MapOutputTrackerMaster(
throw new IllegalArgumentException(logEntry.message)
}
+ // Start the shuffle TTL cleaner only after the fail-fast check above, so a failed construction
+ // can't leave the daemon running.
+ //
+ // Reap only a shuffle that has actually produced output: a registered-but-not-yet-produced
+ // shuffle (e.g. a map stage still waiting on parents) has no files to reclaim.
+ //
+ // Reaping reclaims the on-disk blocks on executors/ESS (shuffleFileRemover, i.e. the normal
+ // ShuffleDriverComponents.removeShuffle path) and then clears the driver-side outputs via
+ // unregisterAllMapAndMergeOutput. Deliberately NOT unregisterShuffle: keeping the (now empty)
+ // ShuffleStatus registered is what makes this safe against a shuffle that is still referenced.
+ // - unregisterAllMapAndMergeOutput calls incrementEpoch, so executors drop their cached
+ // statuses and re-ask rather than fetching files we just deleted. Re-asking yields an empty
+ // status -> MetadataFetchFailedException, which is a FetchFailed, so the DAGScheduler
+ // recomputes the map stage. unregisterShuffle bumps no epoch, so executors would instead
+ // fetch deleted files.
+ // - The DAGScheduler's own FetchFailed recovery calls unregisterAllMapAndMergeOutput /
+ // unregisterMapOutput, which go through getShuffleStatusOrError. With the status removed
+ // those throw ShuffleStatusNotFoundException on the event-loop thread, and
+ // DAGSchedulerEventProcessLoop.onError responds by cancelling all jobs and stopping the
+ // SparkContext -- i.e. reaping a still-referenced shuffle would kill the application.
+ // The cost is that an emptied ShuffleStatus stays in shuffleStatuses until the ContextCleaner
+ // collects it; emptying it also makes numAvailableMapOutputs 0, so it is not reaped again.
+ cleanerThreadpool.foreach { pool =>
+ pool.execute(new BlockTtlCleaner(
+ name = "shuffle",
+ ttlMillis = shuffleTtl.get,
+ accessTimes = shuffleAccessTime,
+ shouldReap = shuffleId => shuffleStatuses.get(shuffleId).exists(_.numAvailableMapOutputs > 0),
+ reap = shuffleId => {
+ shuffleFileRemover.foreach(_(shuffleId))
+ unregisterAllMapAndMergeOutput(shuffleId)
+ }))
+ }
+
def post(message: MapOutputTrackerMasterMessage): Unit = {
mapOutputTrackerMasterMessages.offer(message)
}
@@ -879,6 +947,7 @@ private[spark] class MapOutputTrackerMaster(
val shuffleStatus = shuffleStatuses.get(shuffleId).head
logDebug(s"Handling request to send ${if (needMergeOutput) "map/merge" else "map"}" +
s" output locations for shuffle $shuffleId to $hostPort")
+ updateShuffleAtime(shuffleId)
if (needMergeOutput) {
context.reply(
shuffleStatus.
@@ -930,6 +999,7 @@ private[spark] class MapOutputTrackerMaster(
}
def registerShuffle(shuffleId: Int, numMaps: Int, numReduces: Int): Unit = {
+ updateShuffleAtime(shuffleId)
if (pushBasedShuffleEnabled) {
if (shuffleStatuses.put(shuffleId,
new ShuffleStatus(numMaps, numReduces, bufferRacingMigrations)).isDefined) {
@@ -967,6 +1037,11 @@ private[spark] class MapOutputTrackerMaster(
}
def registerMapOutput(shuffleId: Int, mapIndex: Int, status: MapStatus): Boolean = {
+ // A map task completing output for this shuffle is an active use of it: refresh the atime so a
+ // map stage that runs longer than the TTL is not reaped mid-production. Reduce-side fetches
+ // refresh via handleStatusMessage; this covers the produce side (registerShuffle only stamps
+ // the atime once, at stage submission).
+ updateShuffleAtime(shuffleId)
getShuffleStatusOrError(shuffleId, "registerMapOutput").addMapOutput(mapIndex, status)
}
@@ -978,6 +1053,8 @@ private[spark] class MapOutputTrackerMaster(
/** Unregister all map and merge output information of the given shuffle. */
def unregisterAllMapAndMergeOutput(shuffleId: Int): Unit = {
+ // Drop any TTL tracking (a bare no-op when the map is empty / TTL disabled).
+ shuffleAccessTime.remove(shuffleId)
val shuffleStatus = getShuffleStatusOrError(shuffleId, "unregisterAllMapAndMergeOutput")
shuffleStatus.removeOutputsByFilter(x => true)
shuffleStatus.removeMergeResultsByFilter(x => true)
@@ -1035,6 +1112,10 @@ private[spark] class MapOutputTrackerMaster(
/** Unregister shuffle data */
override def unregisterShuffle(shuffleId: Int): Unit = {
+ // Drop any TTL tracking so the cleaner doesn't later wake up and try to unregister a shuffle
+ // that has already been GC-cleaned (which would throw ShuffleStatusNotFoundException). This
+ // mirrors removeRdd dropping rddAccessTime. A no-op when TTL tracking is disabled (map empty).
+ shuffleAccessTime.remove(shuffleId)
shuffleStatuses.remove(shuffleId).foreach { shuffleStatus =>
shuffleStatus.invalidateSerializedMapOutputStatusCache()
shuffleStatus.invalidateSerializedMergeOutputStatusCache()
@@ -1379,6 +1460,7 @@ private[spark] class MapOutputTrackerMaster(
// This method is only called in local-mode.
override def getShufflePushMergerLocations(shuffleId: Int): Seq[BlockManagerId] = {
+ updateShuffleAtime(shuffleId)
shuffleStatuses.get(shuffleId).map(_.getShufflePushMergerLocations).getOrElse(Seq.empty)
}
@@ -1389,6 +1471,7 @@ private[spark] class MapOutputTrackerMaster(
override def stop(): Unit = {
mapOutputTrackerMasterMessages.offer(PoisonPill)
threadpool.shutdown()
+ cleanerThreadpool.foreach(_.shutdownNow())
try {
sendTracker(StopMapOutputTracker)
} catch {
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 3555491370779..a83e9783a3fd9 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -314,6 +314,19 @@ class SparkContext(config: SparkConf) extends Logging {
val map: ConcurrentMap[Int, RDD[_]] = new MapMaker().weakValues().makeMap[Int, RDD[_]]()
map.asScala
}
+
+ // Ids of RDDs that have been locally checkpointed. A local checkpoint truncates lineage, so its
+ // cache blocks are the only copy of the data and the RDD-cache TTL cleaner must never reap it.
+ // Kept here rather than derived from the live RDD because the cleaner runs on its own thread:
+ // this set gives it a safely-published answer, whereas RDD.checkpointData is not volatile. Not
+ // derived from `persistentRdds` either, since unpersisting drops that entry permanently
+ // (RDD.persist only registers on the first transition out of StorageLevel.NONE).
+ private[spark] val locallyCheckpointedRddIds = ConcurrentHashMap.newKeySet[Int]()
+
+ private[spark] def registerLocallyCheckpointedRdd(rddId: Int): Unit = {
+ locallyCheckpointedRddIds.add(rddId)
+ }
+
def statusTracker: SparkStatusTracker = _statusTracker
private[spark] def progressBar: Option[ConsoleProgressBar] = _progressBar
@@ -645,6 +658,42 @@ class SparkContext(config: SparkConf) extends Logging {
_conf.set(ShuffleDataIOUtils.SHUFFLE_SPARK_CONF_PREFIX + k, v)
}
+ // Wire the TTL cleaners, which live in SparkEnv components created before the pieces they need
+ // exist (ShuffleDriverComponents, the ContextCleaner's listeners, and this SparkContext).
+ // Without this wiring a reap still frees blocks, but reclaims no executor disk for shuffles and
+ // leaves the unpersist bookkeeping stale.
+ _env.mapOutputTracker match {
+ case mapOutputTrackerMaster: MapOutputTrackerMaster =>
+ mapOutputTrackerMaster.shuffleFileRemover = Some { shuffleId =>
+ // Reclaim the on-disk blocks on executors/ESS, then tell the CleanerListeners, which is
+ // what lets dynamic allocation's shuffle tracking release an executor that only held this
+ // shuffle. Mirrors the tail of ContextCleaner.doCleanupShuffle.
+ _shuffleDriverComponents.removeShuffle(shuffleId, false)
+ _cleaner.foreach(_.notifyShuffleCleaned(shuffleId))
+ }
+ case _ =>
+ }
+ _env.blockManagerMasterEndpoint.foreach { endpoint =>
+ // Never TTL-reap a locally-checkpointed RDD: localCheckpoint truncates lineage, so its cache
+ // blocks are the only copy of the data and losing them is unrecoverable. Read from the
+ // dedicated id set, not from persistentRdds: the reap below unpersists, which drops the
+ // persistentRdds entry permanently (RDD.persist only registers on the first transition out of
+ // StorageLevel.NONE), so a persistentRdds-based gate would stop protecting the RDD after the
+ // first reap.
+ endpoint.rddReapable = rddId => !locallyCheckpointedRddIds.contains(rddId)
+ // Reap through ContextCleaner.doCleanupRDD, the same entry point the GC-driven cleanup uses,
+ // rather than re-implementing it: it unpersists the RDD's blocks and notifies the
+ // CleanerListeners. Note this frees the blocks but does not reset the RDD's storage level
+ // (the cleaner only has an id), so a later action re-caches it.
+ // Falls back to unpersistRDD when reference tracking is disabled and there is no cleaner.
+ endpoint.rddReaper = Some { rddId =>
+ _cleaner match {
+ case Some(contextCleaner) => contextCleaner.doCleanupRDD(rddId, blocking = false)
+ case None => unpersistRDD(rddId, blocking = false)
+ }
+ }
+ }
+
if (_conf.get(UI_REVERSE_PROXY)) {
val proxyUrl = _conf.get(UI_REVERSE_PROXY_URL).getOrElse("").stripSuffix("/")
System.setProperty("spark.ui.proxyBase", proxyUrl + "/proxy/" + _applicationId)
@@ -2162,6 +2211,7 @@ class SparkContext(config: SparkConf) extends Logging {
private[spark] def unpersistRDD(rddId: Int, blocking: Boolean): Unit = {
env.blockManager.master.removeRdd(rddId, blocking)
persistentRdds.remove(rddId)
+ locallyCheckpointedRddIds.remove(rddId)
listenerBus.post(SparkListenerUnpersistRDD(rddId))
}
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index d48640d9469b4..47e6bbb07e00e 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -78,6 +78,11 @@ class SparkEnv (
val outputCommitCoordinator: OutputCommitCoordinator,
val conf: SparkConf) extends Logging {
+ // The driver's BlockManagerMasterEndpoint instance, retained so SparkContext can wire the RDD TTL
+ // cleaner's hooks (which need to consult live RDDs and SparkContext.unpersistRDD, neither of
+ // which exists when SparkEnv builds the endpoint). None on executors, where only a ref exists.
+ @volatile private[spark] var blockManagerMasterEndpoint: Option[BlockManagerMasterEndpoint] = None
+
// The two shuffle managers are peers keyed by kind, not a default and an override: a shuffle is
// routed to one or the other by its dependency type via `shuffleManagerFor`, so neither is ever
// installed "behind" the other.
@@ -694,21 +699,28 @@ object SparkEnv extends Logging {
// Mapping from block manager id to the block manager's information.
val blockManagerInfo = new concurrent.TrieMap[BlockManagerId, BlockManagerInfo]()
+ // Captured inside the by-name endpointCreator below so it is only set on the driver (on an
+ // executor the endpoint is never constructed, only looked up).
+ var driverBlockManagerMasterEndpoint: Option[BlockManagerMasterEndpoint] = None
val blockManagerMaster = new BlockManagerMaster(
registerOrLookupEndpoint(
BlockManagerMaster.DRIVER_ENDPOINT_NAME,
- new BlockManagerMasterEndpoint(
- rpcEnv,
- isLocal,
- conf,
- listenerBus,
- if (conf.get(config.SHUFFLE_SERVICE_ENABLED)) {
- externalShuffleClient
- } else {
- None
- }, blockManagerInfo,
- mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
- isDriver)),
+ {
+ val endpoint = new BlockManagerMasterEndpoint(
+ rpcEnv,
+ isLocal,
+ conf,
+ listenerBus,
+ if (conf.get(config.SHUFFLE_SERVICE_ENABLED)) {
+ externalShuffleClient
+ } else {
+ None
+ }, blockManagerInfo,
+ mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
+ isDriver)
+ driverBlockManagerMasterEndpoint = Some(endpoint)
+ endpoint
+ }),
registerOrLookupEndpoint(
BlockManagerMaster.DRIVER_HEARTBEAT_ENDPOINT_NAME,
new BlockManagerMasterHeartbeatEndpoint(rpcEnv, isLocal, blockManagerInfo)),
@@ -773,6 +785,8 @@ object SparkEnv extends Logging {
outputCommitCoordinator,
conf)
+ envInstance.blockManagerMasterEndpoint = driverBlockManagerMasterEndpoint
+
// Add a reference to tmp dir created by driver, we will delete this tmp dir when stop() is
// called, and we only need to do it for driver. Because driver may run as a service, and if we
// don't delete this tmp dir when sc is stopped, then will create too many tmp dirs.
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 87a2da236a91a..d5082e24e6838 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -3262,4 +3262,29 @@ package object config {
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.booleanConf
.createWithDefault(false)
+
+ private[spark] val SPARK_TTL_RDD_CLEANER =
+ ConfigBuilder("spark.cleaner.ttl.rdd")
+ .doc("Add a TTL for RDD cache blocks tracked in Spark (broadcast and other block types " +
+ "are not TTL-cleaned; shuffle blocks use spark.cleaner.ttl.shuffle). By default blocks " +
+ "are only removed after GC on driver, which with DataFrames or RDDs at the global scope " +
+ "will not occur. This must be configured before starting the SparkContext (e.g. can not " +
+ "be added to a running Spark instance).")
+ .version("4.4.0")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .checkValue(_ > 0, "The RDD block TTL must be positive. A zero or negative TTL would make " +
+ "every block immediately eligible for removal.")
+ .createOptional
+
+ private[spark] val SPARK_TTL_SHUFFLE_BLOCK_CLEANER =
+ ConfigBuilder("spark.cleaner.ttl.shuffle")
+ .doc("Add a TTL for shuffle blocks tracked in Spark. By default blocks are only removed " +
+ "after GC on driver, which with DataFrames or RDDs at the global scope will not occur. " +
+ "This must be configured before starting the SparkContext (e.g. can not be added to " +
+ "a running Spark instance).")
+ .version("4.4.0")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .checkValue(_ > 0, "The shuffle block TTL must be positive. A zero or negative TTL would " +
+ "make every shuffle immediately eligible for removal.")
+ .createOptional
}
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index e4cbbe6302af4..88e43d507263e 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -1817,6 +1817,11 @@ abstract class RDD[T: ClassTag](
case _ =>
}
checkpointData = Some(new LocalRDDCheckpointData(this))
+ // Tell the SparkContext, so the RDD-cache TTL cleaner can refuse to reap this RDD: a local
+ // checkpoint truncates lineage, making its cache blocks the only copy of the data. Recorded
+ // here rather than read off `checkpointData` later because the cleaner runs on another thread
+ // and `checkpointData` is not volatile.
+ sc.registerLocallyCheckpointedRdd(id)
// Mark for checksum + seal only when the checkpoint's storage level is serialized: a
// deserialized level keeps in-memory objects with no bytes to checksum, so there is
// nothing to verify and marking would only add cost. (A deserialized default is expected,
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockId.scala b/core/src/main/scala/org/apache/spark/storage/BlockId.scala
index 3e46a53ee082c..8d2d205fb5801 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockId.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockId.scala
@@ -45,12 +45,22 @@ sealed abstract class BlockId {
(isInstanceOf[ShuffleBlockId] || isInstanceOf[ShuffleBlockBatchId] ||
isInstanceOf[ShuffleDataBlockId] || isInstanceOf[ShuffleIndexBlockId])
}
+ // Gate on the ShuffleId trait itself rather than the narrower isShuffle (which covers only the
+ // four plain shuffle block ids), so chunk / push / merged shuffle ids -- which also mix in
+ // ShuffleId -- are recognized here too.
+ def asShuffleId: Option[ShuffleId] =
+ if (isInstanceOf[ShuffleId]) Some(asInstanceOf[ShuffleId]) else None
def isShuffleChunk: Boolean = isInstanceOf[ShuffleBlockChunkId]
def isBroadcast: Boolean = isInstanceOf[BroadcastBlockId]
override def toString: String = name
}
+@DeveloperApi
+trait ShuffleId {
+ def shuffleId: Int
+}
+
@DeveloperApi
case class RDDBlockId(rddId: Int, splitIndex: Int) extends BlockId {
override def name: String = "rdd_" + rddId + "_" + splitIndex
@@ -59,7 +69,8 @@ case class RDDBlockId(rddId: Int, splitIndex: Int) extends BlockId {
// Format of the shuffle block ids (including data and index) should be kept in sync with
// org.apache.spark.network.shuffle.ExternalShuffleBlockResolver#getBlockData().
@DeveloperApi
-case class ShuffleBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId {
+case class ShuffleBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId
+ with ShuffleId {
override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId
}
@@ -69,7 +80,7 @@ case class ShuffleBlockBatchId(
shuffleId: Int,
mapId: Long,
startReduceId: Int,
- endReduceId: Int) extends BlockId {
+ endReduceId: Int) extends BlockId with ShuffleId {
override def name: String = {
"shuffle_" + shuffleId + "_" + mapId + "_" + startReduceId + "_" + endReduceId
}
@@ -81,18 +92,20 @@ case class ShuffleBlockChunkId(
shuffleId: Int,
shuffleMergeId: Int,
reduceId: Int,
- chunkId: Int) extends BlockId {
+ chunkId: Int) extends BlockId with ShuffleId {
override def name: String =
"shuffleChunk_" + shuffleId + "_" + shuffleMergeId + "_" + reduceId + "_" + chunkId
}
@DeveloperApi
-case class ShuffleDataBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId {
+case class ShuffleDataBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId
+ with ShuffleId {
override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId + ".data"
}
@DeveloperApi
-case class ShuffleIndexBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId {
+case class ShuffleIndexBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId
+ with ShuffleId {
override def name: String = "shuffle_" + shuffleId + "_" + mapId + "_" + reduceId + ".index"
}
@@ -108,7 +121,7 @@ case class ShufflePushBlockId(
shuffleId: Int,
shuffleMergeId: Int,
mapIndex: Int,
- reduceId: Int) extends BlockId {
+ reduceId: Int) extends BlockId with ShuffleId {
override def name: String = "shufflePush_" + shuffleId + "_" +
shuffleMergeId + "_" + mapIndex + "_" + reduceId + ""
}
@@ -118,7 +131,7 @@ case class ShufflePushBlockId(
case class ShuffleMergedBlockId(
shuffleId: Int,
shuffleMergeId: Int,
- reduceId: Int) extends BlockId {
+ reduceId: Int) extends BlockId with ShuffleId {
override def name: String = "shuffleMerged_" + shuffleId + "_" +
shuffleMergeId + "_" + reduceId
}
@@ -129,7 +142,7 @@ case class ShuffleMergedDataBlockId(
appId: String,
shuffleId: Int,
shuffleMergeId: Int,
- reduceId: Int) extends BlockId {
+ reduceId: Int) extends BlockId with ShuffleId {
override def name: String = RemoteBlockPushResolver.MERGED_SHUFFLE_FILE_NAME_PREFIX + "_" +
appId + "_" + shuffleId + "_" + shuffleMergeId + "_" + reduceId + ".data"
}
@@ -140,7 +153,7 @@ case class ShuffleMergedIndexBlockId(
appId: String,
shuffleId: Int,
shuffleMergeId: Int,
- reduceId: Int) extends BlockId {
+ reduceId: Int) extends BlockId with ShuffleId {
override def name: String = RemoteBlockPushResolver.MERGED_SHUFFLE_FILE_NAME_PREFIX + "_" +
appId + "_" + shuffleId + "_" + shuffleMergeId + "_" + reduceId + ".index"
}
@@ -151,7 +164,7 @@ case class ShuffleMergedMetaBlockId(
appId: String,
shuffleId: Int,
shuffleMergeId: Int,
- reduceId: Int) extends BlockId {
+ reduceId: Int) extends BlockId with ShuffleId {
override def name: String = RemoteBlockPushResolver.MERGED_SHUFFLE_FILE_NAME_PREFIX + "_" +
appId + "_" + shuffleId + "_" + shuffleMergeId + "_" + reduceId + ".meta"
}
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
index 08d8027406188..1f35d165d7710 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
@@ -19,7 +19,7 @@ package org.apache.spark.storage
import java.io.IOException
import java.util.{HashMap => JHashMap}
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{ConcurrentHashMap, ThreadPoolExecutor, TimeUnit}
import scala.collection.mutable
import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService, Future, TimeoutException}
@@ -29,7 +29,7 @@ import scala.util.control.NonFatal
import com.google.common.cache.CacheBuilder
-import org.apache.spark.{MapOutputTrackerMaster, SparkConf, SparkContext, SparkEnv}
+import org.apache.spark.{BlockTtlCleaner, MapOutputTrackerMaster, SparkConf, SparkContext, SparkEnv}
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.internal.{config, Logging}
import org.apache.spark.internal.LogKeys._
@@ -96,6 +96,12 @@ class BlockManagerMasterEndpoint(
// rddId without scanning the whole map.
private val sealedBlocksByRdd = new mutable.HashMap[Int, mutable.HashSet[RDDBlockId]]
+ // Keep track of last access times if we're using block TTLs. "Close" is good enough for atimes,
+ // but the TTL cleaner thread iterates and removes from this map while the dispatcher thread puts
+ // to it, so it must be concurrent: plain HashMap structural mutation from two threads can corrupt
+ // the map, not merely skew a timestamp.
+ private[spark] val rddAccessTime = new ConcurrentHashMap[Int, Long]
+
// Mapping from task id to the set of rdd blocks which are generated from the task.
private val tidToRddBlockIds = new mutable.HashMap[Long, mutable.HashSet[RDDBlockId]]
// Record the RDD blocks which are not visible yet, a block will be removed from this collection
@@ -115,6 +121,40 @@ class BlockManagerMasterEndpoint(
private implicit val askExecutionContext: ExecutionContextExecutorService =
ExecutionContext.fromExecutorService(askThreadPool)
+
+ // Cache the (immutable-after-start) RDD TTL config once rather than re-parsing the time string on
+ // every updateBlockAtime, which is on the hot path of block-location lookups.
+ private val rddTtl: Option[Long] = conf.get(config.SPARK_TTL_RDD_CLEANER)
+
+ // Whether either TTL is set, so updateBlockAtime can bail out before allocating on the dispatcher
+ // thread's hot path in the default (both-unset) configuration.
+ private val anyTtlEnabled: Boolean =
+ rddTtl.isDefined || conf.get(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER).isDefined
+
+ // Gate consulted before the TTL cleaner reaps an RDD. Wired by SparkContext (which can see the
+ // live RDDs) to refuse locally-checkpointed RDDs: localCheckpoint truncates lineage, so its cache
+ // blocks are the only copy of the data and reaping them loses it unrecoverably. Defaults to
+ // allowing everything for the (test-only) case where nothing wired it.
+ @volatile private[spark] var rddReapable: Int => Boolean = _ => true
+
+ // How the TTL cleaner reaps an RDD. Wired by SparkContext to SparkContext.unpersistRDD so the
+ // reap also clears `persistentRdds` and posts SparkListenerUnpersistRDD (keeping the Storage UI
+ // and getPersistentRDDs accurate); it still removes the blocks through the same RemoveRdd RPC.
+ // When unwired we fall back to sending ourselves RemoveRdd, which frees the blocks but leaves
+ // that bookkeeping stale.
+ @volatile private[spark] var rddReaper: Option[Int => Unit] = None
+
+ // Created here so onStop can shut it down, but the cleaner is not started until onStart: it
+ // routes removals through self.ask(RemoveRdd(...)), and self is only valid once the endpoint is
+ // registered (i.e. from onStart onward).
+ private[spark] val cleanerThreadpool: Option[ThreadPoolExecutor] = {
+ if (rddTtl.isDefined) {
+ Some(ThreadUtils.newDaemonFixedThreadPool(1, "rdd-ttl-cleaner"))
+ } else {
+ None
+ }
+ }
+
private val topologyMapper = {
val topologyMapperClassName = conf.get(
config.STORAGE_REPLICATION_TOPOLOGY_MAPPER)
@@ -154,6 +194,8 @@ class BlockManagerMasterEndpoint(
case _updateBlockInfo @
UpdateBlockInfo(blockManagerId, blockId, storageLevel, deserializedSize, size, checksum) =>
+ // We don't update the block access times here because the update block infos are triggered by
+ // migrations rather than actual access.
@inline def handleResult(success: Boolean): Unit = {
// SPARK-30594: we should not post `SparkListenerBlockUpdated` when updateBlockInfo
// returns false since the block info would be updated again later.
@@ -270,6 +312,30 @@ class BlockManagerMasterEndpoint(
context.reply(updateRDDBlockVisibility(taskId, visible))
}
+ private def updateBlockAtime(blockId: BlockId): Unit = {
+ // Both TTLs are unset by default, and this runs on the dispatcher thread for every block
+ // location lookup (O(partitions) per stage via DAGScheduler.getCacheLocs), so return before
+ // doing any work -- including the Option allocations below -- when the feature is off.
+ if (anyTtlEnabled) {
+ blockId match {
+ case rddBlockId: RDDBlockId =>
+ // Only RDD cache blocks are TTL-tracked (not broadcast). Record an access only for a
+ // block we actually track: a lookup of an absent/evicted block must not create a phantom
+ // rddAccessTime entry that the cleaner would later try to reap (broadcasting a RemoveRdd
+ // for an RDD that has nothing to remove). Timestamp races are "close enough" so we don't
+ // check the return value.
+ if (rddTtl.isDefined && blockLocations.containsKey(blockId)) {
+ rddAccessTime.put(rddBlockId.rddId, System.currentTimeMillis())
+ }
+ case shuffleBlockId: ShuffleId =>
+ // Shuffle blocks are tracked in the map output tracker (which self-guards on its own TTL
+ // config); everything else (e.g. broadcast) is not TTL-tracked.
+ mapOutputTracker.updateShuffleAtime(shuffleBlockId.shuffleId)
+ case _ =>
+ }
+ }
+ }
+
private def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
if (trackingCacheVisibility) {
blockLocations.containsKey(blockId) &&
@@ -366,7 +432,10 @@ class BlockManagerMasterEndpoint(
}
private def removeRdd(rddId: Int): Future[Seq[Int]] = {
- // First remove the metadata for the given RDD, and then asynchronously remove the blocks
+ // Drop the RDD from TTL tracking (a bare no-op when the map is empty / TTL disabled).
+ rddAccessTime.remove(rddId)
+
+ // Then remove the metadata for the given RDD, and then asynchronously remove the blocks
// from the storage endpoints.
// The message sent to the storage endpoints to remove the RDD
@@ -438,6 +507,7 @@ class BlockManagerMasterEndpoint(
}
private def removeShuffle(shuffleId: Int): Future[Seq[Boolean]] = {
+ // Start with removing shuffle blocks without an associated executor (e.g. ESS only).
// Find all shuffle blocks on executors that are no longer running
val blocksToDeleteByShuffleService =
new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
@@ -495,6 +565,7 @@ class BlockManagerMasterEndpoint(
}
}.getOrElse(Seq.empty)
+ // Remove shuffle blocks from running executors.
val removeMsg = RemoveShuffle(shuffleId)
val removeShuffleFromExecutorsFutures = blockManagerInfo.values.map { bm =>
bm.storageEndpoint.ask[Boolean](removeMsg).recover {
@@ -580,6 +651,7 @@ class BlockManagerMasterEndpoint(
}
private def addMergerLocation(blockManagerId: BlockManagerId): Unit = {
+ logDebug(log"Adding merger location ${MDC(BLOCK_MANAGER_ID, blockManagerId)}")
if (!blockManagerId.isDriver && !shuffleMergerLocations.contains(blockManagerId.host)) {
val shuffleServerId = BlockManagerId(BlockManagerId.SHUFFLE_MERGER_IDENTIFIER,
blockManagerId.host, externalShuffleServicePort)
@@ -799,7 +871,8 @@ class BlockManagerMasterEndpoint(
}
private def updateShuffleBlockInfo(blockId: BlockId, blockManagerId: BlockManagerId)
- : Future[Boolean] = {
+ : Future[Boolean] = {
+ logDebug(s"Updating shuffle block info ${blockId} on ${blockManagerId}")
blockId match {
case ShuffleIndexBlockId(shuffleId, mapId, _) =>
// SPARK-36782: Invoke `MapOutputTracker.updateMapOutput` within the thread
@@ -857,6 +930,15 @@ class BlockManagerMasterEndpoint(
} else {
locations = new mutable.HashSet[BlockManagerId]
blockLocations.put(blockId, locations)
+ // Since it's the initial put we register this as an access as well -- but only for a report
+ // that actually stores the block. A report with an invalid level is a *removal* (e.g. the
+ // UpdateBlockInfo replies that follow a RemoveRdd broadcast, or an eviction report arriving
+ // after the last replica is gone); those also land here, and stamping them would resurrect an
+ // access time for an RDD with no blocks left, which one TTL later fires a pointless
+ // cluster-wide RemoveRdd.
+ if (storageLevel.isValid) {
+ updateBlockAtime(blockId)
+ }
}
if (storageLevel.isValid) {
@@ -1012,12 +1094,14 @@ class BlockManagerMasterEndpoint(
}
private def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
+ updateBlockAtime(blockId)
if (blockLocations.containsKey(blockId)) blockLocations.get(blockId).toSeq else Seq.empty
}
private def getLocationsAndStatus(
blockId: BlockId,
requesterHost: String): Option[BlockLocationsAndStatus] = {
+ updateBlockAtime(blockId)
val allLocations = Option(blockLocations.get(blockId)).map(_.toSeq).getOrElse(Seq.empty)
val blockStatusWithBlockManagerId: Option[(BlockStatus, BlockManagerId)] =
(if (externalShuffleServiceRddFetchEnabled && blockId.isRDD) {
@@ -1130,8 +1214,33 @@ class BlockManagerMasterEndpoint(
}
}
+ override def onStart(): Unit = {
+ // Start the TTL cleaner only now that the endpoint is registered: it reaps by routing through
+ // self.ask(RemoveRdd(...)) (so blockLocations et al. are mutated only on the single dispatcher
+ // thread, preserving the IsolatedThreadSafeRpcEndpoint invariant), and self is only valid from
+ // onStart onward. RemoveRdd is the existing message; its reply is the async removal Future.
+ cleanerThreadpool.foreach { pool =>
+ pool.execute(new BlockTtlCleaner(
+ name = "RDD",
+ ttlMillis = rddTtl.get,
+ accessTimes = rddAccessTime,
+ shouldReap = rddId => rddReapable(rddId),
+ reap = rddId => rddReaper match {
+ case Some(reaper) => reaper(rddId)
+ case None =>
+ // Unwired fallback (nothing attached the SparkContext hook). Observe the returned
+ // removal future the way BlockManagerMaster.removeRdd does, so a failure to remove
+ // blocks on some executor is not silently dropped.
+ self.askSync[Future[Seq[Int]]](RemoveRdd(rddId)).failed.foreach { e =>
+ logWarning(log"Failed to remove RDD ${MDC(RDD_ID, rddId)} in the TTL cleaner", e)
+ }(ThreadUtils.sameThread)
+ }))
+ }
+ }
+
override def onStop(): Unit = {
askThreadPool.shutdownNow()
+ cleanerThreadpool.foreach(_.shutdownNow())
}
}
diff --git a/core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala
new file mode 100644
index 0000000000000..18a46a1516f6c
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import org.scalatest.concurrent.Eventually
+import org.scalatest.time._
+
+import org.apache.spark._
+import org.apache.spark.internal.config
+
+class BlockTTLIntegrationSuite extends SparkFunSuite with LocalSparkContext
+ with Eventually {
+
+ implicit override val patienceConfig: PatienceConfig =
+ PatienceConfig(timeout = scaled(Span(20, Seconds)), interval = scaled(Span(5, Millis)))
+
+ val blockTTL = 5000L
+ // Long enough that the cleaner won't reap mid-test; used by the tests that only check that
+ // tracking happens / atime is refreshed (not the removal-after-TTL tests).
+ val longTTL = 60000L
+
+ val numParts = 3
+
+ private def lookupBlockManagerMasterEndpoint(sc: SparkContext): BlockManagerMasterEndpoint = {
+ // The driver retains its endpoint instance precisely so this is reachable without reflection.
+ sc.env.blockManagerMasterEndpoint.get
+ }
+
+ private def lookupMapOutputTrackerMaster(sc: SparkContext): MapOutputTrackerMaster = {
+ // On the driver the tracker is always a MapOutputTrackerMaster.
+ sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
+ }
+
+ test("Test that cache blocks are recorded.") {
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-decommissioner")
+ .setMaster("local-cluster[2, 1, 1024]")
+ .set(config.SPARK_TTL_RDD_CLEANER, longTTL)
+ .set(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER, longTTL)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty)
+ // Make some cache blocks
+ val input = sc.parallelize(1.to(100)).cache()
+ input.count()
+ // Check that the blocks were registered with the TTL tracker. Wrapped in eventually because the
+ // executors' UpdateBlockInfo reports can land just after count() returns.
+ eventually { assert(managerMasterEndpoint.rddAccessTime.containsKey(input.id)) }
+ }
+
+ test("Test that re-reading a cached RDD in a new job refreshes its access time") {
+ // This pins down the assumption behind the RDD TTL: an actively-reused cached RDD is not
+ // reaped, because every job that reads it re-resolves cache locations at the driver
+ // (DAGScheduler.clearCacheLocs -> getCacheLocs -> BlockManagerMaster.getLocations ->
+ // updateBlockAtime), refreshing the atime -- independent of whether the block read itself is
+ // served locally on the executor.
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-ttls-rdd-refresh")
+ .setMaster("local-cluster[2, 1, 1024]")
+ .set(config.SPARK_TTL_RDD_CLEANER, longTTL)
+ .set(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER, longTTL)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ val input = sc.parallelize(1.to(100), numParts).cache()
+ input.count()
+ // The cached blocks are tracked, keyed by RDD id.
+ eventually { assert(managerMasterEndpoint.rddAccessTime.containsKey(input.id)) }
+ // Read via Option, not get: a ConcurrentHashMap[Int, Long] miss unboxes null to 0L rather than
+ // throwing, which would let the "atime advanced" assertion below pass vacuously.
+ def atimeOf(rddId: Int): Option[Long] =
+ Option(managerMasterEndpoint.rddAccessTime.get(rddId)).map(_.longValue)
+ val firstAtime = atimeOf(input.id).getOrElse(
+ fail("the cached RDD should be TTL-tracked before we test the refresh"))
+ // Re-reading the cached RDD in a new job must refresh (advance) its access time. Re-running
+ // inside eventually guards against the clock not having ticked past firstAtime yet; if the
+ // cleaner had already reaped it, count() re-materializes it and the atime still advances.
+ eventually {
+ input.count()
+ assert(atimeOf(input.id).exists(_ > firstAtime),
+ s"a new job reading the cached RDD should refresh its atime (was $firstAtime)")
+ }
+ }
+
+ test("Test that shuffle blocks are tracked properly and removed after TTL") {
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-ttls-shuffle-only")
+ .setMaster("local-cluster[2, 1, 1024]")
+ .set(config.SPARK_TTL_RDD_CLEANER, blockTTL)
+ .set(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER, blockTTL)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ val mapOutputTracker = lookupMapOutputTrackerMaster(sc)
+ // Make sure it's empty at the start
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty)
+ assert(mapOutputTracker.shuffleAccessTime.isEmpty)
+ // Make some cache blocks
+ val input = sc.parallelize(1.to(100)).groupBy(_ % 10)
+ input.count()
+ // Make sure we've got the tracker threads defined
+ assert(mapOutputTracker.cleanerThreadpool.isDefined)
+ // Check that the shuffle blocks were NOT registered with the RDD TTL tracker.
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty)
+ // Check that the shuffle blocks are registered with the map output TTL
+ eventually { assert(!mapOutputTracker.shuffleAccessTime.isEmpty) }
+ // It should be expired!
+ eventually {
+ val t = System.currentTimeMillis()
+ assert(
+ mapOutputTracker.shuffleAccessTime.isEmpty,
+ s"We should have no blocks since we are now at time ${t} with ttl of ${blockTTL}")
+ }
+ }
+
+
+ test(s"Test that all blocks are tracked properly and removed after TTL") {
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-ttls-enabled")
+ .setMaster("local-cluster[2, 1, 1024]")
+ .set(config.SPARK_TTL_RDD_CLEANER, blockTTL)
+ .set(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER, blockTTL)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ val mapOutputTracker = lookupMapOutputTrackerMaster(sc)
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty)
+ // Make some cache blocks
+ val input = sc.parallelize(1.to(100)).groupBy(_ % 10)
+ val cachedInput = input.cache()
+ cachedInput.count()
+ // Check that we have both shuffle & RDD blocks registered
+ eventually { assert(!managerMasterEndpoint.rddAccessTime.isEmpty) }
+ eventually { assert(!mapOutputTracker.shuffleAccessTime.isEmpty) }
+ // Both should be expired!
+ eventually {
+ val t = System.currentTimeMillis()
+ assert(mapOutputTracker.shuffleAccessTime.isEmpty,
+ s"We should have no blocks since we are now at time ${t} with ttl of ${blockTTL}")
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty,
+ s"We should have no blocks since we are now at time ${t} with ttl of ${blockTTL}")
+ }
+ // And redoing the count should work and everything should come back.
+ input.count()
+ eventually {
+ assert(!managerMasterEndpoint.rddAccessTime.isEmpty)
+ assert(!mapOutputTracker.shuffleAccessTime.isEmpty)
+ }
+ }
+
+ test("Test that a locally-checkpointed RDD is never reaped by the TTL cleaner") {
+ // localCheckpoint truncates lineage, so the cache blocks are the only copy of the data: reaping
+ // them loses it unrecoverably (LocalCheckpointRDD.compute always throws). The cleaner must skip
+ // such RDDs no matter how long they sit idle.
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-ttls-local-checkpoint")
+ .setMaster("local-cluster[2, 1, 1024]")
+ .set(config.SPARK_TTL_RDD_CLEANER, blockTTL)
+ .set(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER, blockTTL)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ val checkpointed = sc.parallelize(1.to(100), numParts)
+ checkpointed.localCheckpoint()
+ assert(checkpointed.count() === 100)
+ // Sit idle for longer than the TTL; a plain cached RDD would be reaped in this window.
+ Thread.sleep(blockTTL * 2)
+ // The data must still be readable -- this is the assertion that would fail on data loss.
+ assert(checkpointed.count() === 100,
+ "a locally-checkpointed RDD must survive the TTL: its blocks are the only copy")
+ assert(managerMasterEndpoint.rddReapable(checkpointed.id) === false,
+ "the TTL cleaner must refuse to reap a locally-checkpointed RDD")
+ }
+
+ test("Test that reaping an RDD removes every partition and leaves the RDD usable") {
+ // Access times are recorded per RDD id but stamped by individual block accesses, and a reap
+ // removes the whole RDD. Pin both halves of that: every partition's block goes away, and since
+ // the reap frees blocks without resetting the RDD's storage level it is an eviction, not an
+ // unpersist -- the RDD still computes the right answer and re-caches on the next action.
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-ttls-full-rdd-removal")
+ .setMaster("local-cluster[2, 1, 1024]")
+ .set(config.SPARK_TTL_RDD_CLEANER, blockTTL)
+ .set(config.SPARK_TTL_SHUFFLE_BLOCK_CLEANER, longTTL)
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ val input = sc.parallelize(1.to(100), numParts).cache()
+ assert(input.count() === 100)
+
+ def cachedPartitionsOf(rddId: Int): Int =
+ sc.env.blockManager.master.getMatchingBlockIds({
+ case RDDBlockId(id, _) => id == rddId
+ case _ => false
+ }, askStorageEndpoints = true).size
+
+ // All partitions are cached and the RDD is tracked.
+ eventually {
+ assert(managerMasterEndpoint.rddAccessTime.containsKey(input.id))
+ assert(cachedPartitionsOf(input.id) >= numParts,
+ s"expected all $numParts partitions cached")
+ }
+
+ // After the TTL every one of this RDD's blocks is gone from the master's directory, and it is
+ // no longer tracked.
+ eventually {
+ assert(!managerMasterEndpoint.rddAccessTime.containsKey(input.id),
+ "the reaped RDD should no longer be TTL-tracked")
+ assert(cachedPartitionsOf(input.id) === 0,
+ "every partition of the reaped RDD should be removed, not just the idle ones")
+ }
+
+ // Eviction, not unpersist: the RDD still produces the right answer and comes back tracked.
+ assert(input.count() === 100, "a reaped RDD must still be usable (recomputed)")
+ eventually { assert(managerMasterEndpoint.rddAccessTime.containsKey(input.id)) }
+ }
+
+ test("Test that blocks TTLS are not tracked when not enabled") {
+ val conf = new SparkConf()
+ .setAppName("test-blockmanager-decommissioner")
+ .setMaster("local-cluster[2, 1, 1024]")
+ sc = new SparkContext(conf)
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+ val managerMasterEndpoint = lookupBlockManagerMasterEndpoint(sc)
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty)
+ // Make some cache blocks
+ val input = sc.parallelize(1.to(100)).groupBy(_ % 10).cache()
+ input.count()
+ // Check that no RDD blocks are tracked
+ assert(managerMasterEndpoint.rddAccessTime.isEmpty)
+ // Check that the no shuffle blocks are tracked.
+ val mapOutputTracker = lookupMapOutputTrackerMaster(sc)
+ assert(mapOutputTracker.shuffleAccessTime.isEmpty)
+ }
+}
diff --git a/docs/configuration.md b/docs/configuration.md
index 38460273a823b..f2d3a17205c18 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -2266,6 +2266,38 @@ Apart from these, the following properties are also available, and may be useful
| 1.4.0 |
+
+ spark.cleaner.ttl.rdd |
+ (none) |
+
+ If set, RDD cache blocks that have not been accessed for this duration are removed. By default
+ cached blocks are only removed once the RDD is garbage collected on the driver, which does not
+ happen for a DataFrame or RDD held at global scope. Broadcast and other block types are not
+ covered; shuffle blocks use spark.cleaner.ttl.shuffle. An access is recorded when the
+ driver resolves the block's locations, so an RDD read only from executor-local caches within a
+ single long job may not appear to be in use: set this comfortably longer than the longest gap
+ between uses and the longest stage runtime, or the RDD will be removed and recomputed. Removal
+ frees the blocks but does not un-persist the RDD, so a later action re-caches it. Locally
+ checkpointed RDDs are never removed. Must be set before the SparkContext is created.
+ |
+ 4.4.0 |
+
+
+ spark.cleaner.ttl.shuffle |
+ (none) |
+
+ If set, shuffles that have not been accessed for this duration have their map output removed from
+ the driver and their shuffle files deleted on the executors. By default shuffle data is only
+ removed once the shuffle dependency is garbage collected on the driver. An access is recorded when
+ a map task registers output or the driver serves map output locations; executors cache those
+ locations for an epoch, so a shuffle that is actively being read may not appear to be in use.
+ Set this comfortably longer than the longest gap between uses and the longest stage runtime:
+ removing a shuffle that is still needed makes its readers fail and the map stage recompute, and
+ if that repeats it will exhaust spark.stage.maxConsecutiveAttempts and fail the job.
+ Must be set before the SparkContext is created.
+ |
+ 4.4.0 |
+
### Execution Behavior