[CORE][SPARK-49788] Add optional TTL based cleaning for blocks to better support notebooks - #49032
Draft
holdenk wants to merge 29 commits into
Draft
[CORE][SPARK-49788] Add optional TTL based cleaning for blocks to better support notebooks#49032holdenk wants to merge 29 commits into
holdenk wants to merge 29 commits into
Conversation
holdenk
force-pushed
the
SPARK-49788-Restore-Cleaner-TTL-Functionality
branch
from
December 5, 2024 23:13
d763e01 to
907e2e2
Compare
holdenk
changed the title
|
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable. |
Contributor
Author
|
I've got some time to work on this again, lets see :) |
holdenk
force-pushed
the
SPARK-49788-Restore-Cleaner-TTL-Functionality
branch
from
November 3, 2025 18:42
26839a7 to
92ae9af
Compare
Contributor
Author
|
CC @JoshRosen what are your thoughts / do you have some cycles to review? |
Contributor
Author
|
CC @ivoson any thoughts? |
There was a problem hiding this comment.
Pull Request Overview
This pull request implements a Time-To-Live (TTL) mechanism for blocks in Spark, allowing automatic cleanup of RDD and shuffle blocks after a configurable time period. This addresses the issue where blocks persist indefinitely when DataFrames or RDDs are held at global scope and garbage collection doesn't occur.
Key Changes
- Added configurable TTL settings (
spark.cleaner.ttl.allandspark.cleaner.ttl.shuffle) for controlling block lifetimes - Implemented background cleaner threads in
BlockManagerMasterEndpointandMapOutputTrackerMasterto periodically remove expired blocks - Added access time tracking for RDD and shuffle blocks to support TTL-based cleanup
Reviewed Changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 16 comments.
Show a summary per file
| File | Description |
|---|---|
core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala |
New integration test suite validating TTL tracking and cleanup behavior for cache and shuffle blocks |
core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala |
Adds RDD access time tracking, TTL cleaner thread, and hooks to update access times on block operations |
core/src/main/scala/org/apache/spark/storage/BlockId.scala |
Introduces ShuffleId trait to enable uniform access to shuffle IDs across different shuffle block types |
core/src/main/scala/org/apache/spark/internal/config/package.scala |
Defines two new configuration options for TTL-based block cleanup |
core/src/main/scala/org/apache/spark/MapOutputTracker.scala |
Adds shuffle access time tracking, TTL cleaner thread, and integration with shuffle lifecycle operations |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| val maxAge = System.currentTimeMillis() - ttl | ||
| // Find the elements to be removed & update oldest remaining time (if any) | ||
| var oldest = System.currentTimeMillis() | ||
| val toBeRemoved = rddAccessTime.asScala.flatMap { case (rddId, atime) => |
There was a problem hiding this comment.
Potential race condition: The TTL cleaner thread reads from rddAccessTime and calls removeRdd(), which also modifies rddAccessTime. Since rddAccessTime is a non-concurrent JHashMap, concurrent access from the cleaner thread and other threads updating access times could lead to ConcurrentModificationException or data corruption. Consider using a ConcurrentHashMap or adding proper synchronization.
Contributor
Author
|
@ivoson yeah those are only for SQL jobs. |
|
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable. |
… printed out on failure inside of the eventual block, remove RDD/Shuffle from access time tracking regardless of if RDD/shuffle removal succeeds
…ance to check for them (mostly needed for GHA where we run in rather resource constrained machines).
… faster than we can get to them
…al through the endpoint loop Self-review of the rebased branch turned up two thread-safety bugs in the TTL cleaner threads (both pre-existing in the branch, surfaced by the rebase review): 1. shuffleAccessTime (MapOutputTrackerMaster) and rddAccessTime (BlockManagerMasterEndpoint) were plain java.util.HashMaps written from multiple threads (the map-output dispatcher pool, DAGScheduler, and the TTL cleaner thread) with no synchronization. Concurrent structural mutation of a plain HashMap can corrupt the map, not merely skew a "close enough" atime, so both are now ConcurrentHashMaps. 2. BlockManagerMasterEndpoint is an IsolatedThreadSafeRpcEndpoint whose non-concurrent maps (blockLocations, blockChecksums, sealedChecksums, ...) are only safe because a single dispatcher thread touches them. The cleaner thread called removeRdd directly, racing that dispatcher thread. It now routes removal through the endpoint's own message loop via self.askSync[Future[Seq[Int]]](RemoveRdd(rddId)), keeping removeRdd on the dispatcher thread. MapOutputTrackerMaster is concurrent-by-design (epochLock, ConcurrentHashMap statuses, per-ShuffleStatus locks) so its off-thread unregisterAllMapAndMergeOutput needs no equivalent change. Also drop a doubled shuffleAccessTime.remove in unregisterAllMapAndMergeOutput left over from the rebase. Verified: core main/test compile clean; BlockTTLIntegrationSuite 4/4 pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Bugs (Tier B):
- config: new spark.cleaner.ttl.{all,shuffle} tagged .version("5.0.0") (was 4.3.0;
project is 5.0.0-SNAPSHOT).
- MapOutputTracker.unregisterShuffle now drops the shuffleAccessTime entry (mirrors
removeRdd dropping rddAccessTime), so a GC-cleaned shuffle can't leave a stale entry
that later makes the cleaner call unregisterAllMapAndMergeOutput on an absent shuffle
(ShuffleStatusNotFoundException churn).
- BlockManagerMasterEndpoint.updateBlockAtime only records an RDD access when the block is
actually tracked, so a lookup of an absent/evicted block no longer creates a phantom
rddAccessTime entry that the cleaner would later broadcast a RemoveRdd for.
- BlockId.asShuffleId now gates on the ShuffleId trait (isInstanceOf[ShuffleId]) instead of
the narrower isShuffle, so chunk/push/merged shuffle ids are recognized.
- Both cleaners now use ConcurrentHashMap.remove(key, atime) so an entry refreshed between
the snapshot and the removal decision (a concurrent access) is not reaped (TOCTOU).
Cleanup (Tier C):
- Removed the now-dead catch ConcurrentModificationException + retry in both cleaners
(ConcurrentHashMap views are weakly consistent) and the misleading "reduce chance of CME"
comment; fixed a copy-pasted "shuffle" comment in the RDD cleaner.
- BlockManagerMasterEndpoint starts its TTLCleaner in onStart() rather than a val
initializer, since the cleaner uses self (valid only once registered); onStop uses foreach.
- BlockTTLIntegrationSuite no longer calls sc.setLogLevel("DEBUG") (JVM-wide, never reset)
and looks up the tracker via sc.env.mapOutputTracker, so the test-only
getMapOutputTrackerMaster() accessor is removed from the endpoint.
Verified: core main/test compile clean; BlockTTLIntegrationSuite 4/4 pass.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
holdenk
marked this pull request as draft
August 12, 2026 18:44
Contributor
Author
|
Converting to a draft while I get this back up on the latest branch |
… atime on map output (F2/F3) F3 (route shuffle TTL through the real removal path): - The shuffle TTL cleaner previously called unregisterAllMapAndMergeOutput, which only cleared driver-side outputs + bumped the epoch: it never sent RemoveShuffle to executors (so on-disk shuffle files were never freed -- the stated purpose unmet) and never removed the ShuffleStatus (leaking it). It now mirrors ContextCleaner.doCleanupShuffle: reclaim on-disk blocks on executors/ESS first, then drop the driver ShuffleStatus via unregisterShuffle. - MapOutputTrackerMaster lives in SparkEnv and has no ShuffleDriverComponents, so it exposes a shuffleFileRemover hook wired by SparkContext to shuffleDriverComponents.removeShuffle(id, false). This keeps the shuffle TTL feature independent of spark.cleaner.referenceTracking and honors a custom driver-shuffle-components plugin. If the hook is unset the cleaner still unregisters the status (fixing the leak) but cannot reclaim executor disk. F2 (refresh shuffle atime on produce side): - registerMapOutput now calls updateShuffleAtime, so a map stage running longer than the TTL is not reaped mid-production (registerShuffle only stamps the atime once, at stage submission; reduce fetches refresh via handleStatusMessage). This also matters now that F3 deletes files on reap. Test: - New BlockTTLIntegrationSuite case pins down the RDD TTL assumption: re-reading a cached RDD in a new job re-resolves cache locations at the driver (clearCacheLocs -> getCacheLocs -> BlockManagerMaster.getLocations -> updateBlockAtime) and thus refreshes the atime, so an actively-reused cached RDD is not reaped. Verified: core main/test compile clean; BlockTTLIntegrationSuite 5/5 pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Correctness: - X1 (regression I introduced with F3): the shuffle reaper now only reaps a shuffle that has actually produced output (numAvailableMapOutputs > 0). A registered-but-not-yet-produced shuffle has no files to reclaim, and removing its ShuffleStatus made the first registerMapOutput throw ShuffleStatusNotFoundException -> abortStage (hard job failure). The old code kept the status (leak) and didn't abort; this keeps F3's disk reclamation without the abort. - updateBlockAtime now branches on asRDDId / asShuffleId instead of the narrow isShuffle, so the ShuffleId trait (chunk/push/merged ids) is actually reachable; access-time UPDATES stay gated on the TTL config being set (so we don't populate maps no cleaner will drain). Cleanup: - Extracted the two byte-identical TTLCleaner loops into a shared BlockTtlCleaner(name, ttl, map, shouldReap, reap); the RDD cleaner passes shouldReap = _ => true, the shuffle cleaner the has-output gate above. - Cache the TTL config (Option[Long]) in a val instead of re-parsing conf.get on every updateBlockAtime/updateShuffleAtime hot-path call. - MapOutputTrackerMaster starts its cleaner after the fail-fast broadcast-size check (was in a field initializer, which orphaned the daemon if that check threw); removeRdd/unregister* use a bare unconditional access-map remove (a no-op when empty) instead of a guarded, dead-try-wrapped one; MOT stop() uses foreach not map. - Rename spark.cleaner.ttl.all -> spark.cleaner.ttl.rdd (it cleans only RDD cache blocks, not all; shuffle uses spark.cleaner.ttl.shuffle) and reword the doc. - Test: long TTL for the tracking/refresh-only tests (avoids racing the 5s reaper on a slow CI); drop dead constants (numExecs, TaskStarted/Ended, JobEnded). Per Holden: kept getShufflePushMergerLocations' updateShuffleAtime (needed for atime) and did not add a minimum-TTL floor (operator's responsibility). Verified: core main/test compile clean; BlockTTLIntegrationSuite 5/5 pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
holdenk
force-pushed
the
SPARK-49788-Restore-Cleaner-TTL-Functionality
branch
from
August 14, 2026 20:31
c7fc32d to
1e126c3
Compare
- Wrap 6 comment/doc lines that exceeded the 100-char limit; core/scalastyle and
core/Test/scalastyle now report 0 errors (they failed before, so CI would have rejected this).
- .version("5.0.0") -> .version("4.4.0") for both new configs. Per the repo versioning policy a
normally-backported PR against master ships first in branch-<N>.x;
dev/next_version_candidates.py reports master 5.0.0 / branch-4.x 4.4.0, and an additive pair of
createOptional configs is neither a breaking change nor a dependency upgrade.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…arkContext
Reaping a still-referenced shuffle via unregisterShuffle (introduced with F3) was fatal, and the
numAvailableMapOutputs gate did not cover it -- that gate handles "not yet produced", not "produced
and still needed". The two metadata calls differ in two safety-relevant ways:
unregisterShuffle removes the ShuffleStatus, does NOT incrementEpoch
unregisterAllMapAndMergeOutput keeps the (emptied) status, DOES incrementEpoch
So unregisterShuffle broke reaping twice over:
1. No epoch bump, so executors keep their cached map statuses and go on to fetch the files the
reaper just deleted -> FetchFailed.
2. With the status removed, the DAGScheduler's own FetchFailed recovery
(unregisterAllMapAndMergeOutput / unregisterMapOutput -> getShuffleStatusOrError) throws
ShuffleStatusNotFoundException on the event-loop thread, and
DAGSchedulerEventProcessLoop.onError responds with doCancelAllJobs() + sc.stopInNewThread():
reaping a live shuffle took down the whole application.
Reap now deletes the on-disk blocks (shuffleFileRemover, i.e. the normal removeShuffle path -- still
the point of F3) and then calls unregisterAllMapAndMergeOutput instead of unregisterShuffle. The
epoch bump makes executors re-ask rather than read deleted files; re-asking returns an empty status,
which surfaces as MetadataFetchFailedException -- a FetchFailed -- so the DAGScheduler recomputes the
map stage normally instead of hanging on an unanswered RPC. Keeping the status registered also means
the scheduler's error paths can no longer throw. The cost is that an emptied ShuffleStatus lingers
until the ContextCleaner collects it (the same small leak the original code had); emptying it makes
numAvailableMapOutputs 0, so it is not reaped again.
Verified: core/scalastyle + core/Test/scalastyle 0 errors; BlockTTLIntegrationSuite,
MapOutputTrackerSuite and ContextCleanerSuite 55 succeeded / 0 failed.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…reclamation path Four review findings, all reachable with the TTL configs enabled: 1. Data loss: the RDD cleaner reaped any tracked RDD, including a locally-checkpointed one. localCheckpoint truncates lineage and LocalCheckpointRDD.compute always throws, so those cache blocks are the only copy of the data -- reaping them loses it unrecoverably (RDD.localCheckpoint already warns about the same hazard with dynamicAllocation.cachedExecutorIdleTimeout). The endpoint now consults a wired `rddReapable` gate, which SparkContext implements by refusing any RDD whose live instance isLocallyCheckpointed. New test idles such an RDD past the TTL and asserts the data is still readable. 2. A zero or negative TTL was accepted: JavaUtils' time pattern is "(-?[0-9]+)([a-z]+)?", so ttl=-1s parses to -1000 and maxAge = now + 1000, making every entry instantly stale -- the cleaner then reaps everything roughly every 100ms (the delay clamp) for the life of the application. Both configs now .checkValue(_ > 0, ...). This is not a minimum-TTL floor (still deliberately absent, per the operator's-responsibility call): it just rejects input that inverts the comparison. 3. Reaping a shuffle skipped the CleanerListener.shuffleCleaned fan-out that ContextCleaner.doCleanupShuffle ends with. ExecutorMonitor uses it to let an executor holding only that shuffle go idle, so under dynamicAllocation.shuffleTracking the TTL deleted the files but never released the executor. The wired remover now calls ContextCleaner.notifyShuffleCleaned (new, private[spark]) after removeShuffle. 4. The RDD reap sent RemoveRdd directly, which is only the RPC half of SparkContext.unpersistRDD: persistentRdds was never cleared and no SparkListenerUnpersistRDD was posted, so the Storage UI showed the RDD as fully cached forever with stale per-executor block counts. The reap now routes through unpersistRDD (which still removes blocks via the same RemoveRdd RPC). Wiring: SparkEnv retains the driver's BlockManagerMasterEndpoint (captured inside the by-name endpointCreator, so it stays None on executors) so SparkContext can attach the two RDD hooks alongside the existing shuffle one. RDD.isLocallyCheckpointed widens from private[rdd] to private[spark]. Verified: core/scalastyle + core/Test/scalastyle 0 errors; BlockTTLIntegrationSuite, MapOutputTrackerSuite, ContextCleanerSuite and LocalCheckpointSuite 73 succeeded / 0 failed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ing it The RDD reap hook duplicated the first half of ContextCleaner.doCleanupRDD (sc.unpersistRDD) and dropped the second (the CleanerListener.rddCleaned fan-out). Call doCleanupRDD instead, so the TTL path and the GC-driven path share one entry point and its error handling. Both in-tree rddCleaned implementations are currently empty, so this is not a behavior fix like the shuffleCleaned one was -- it removes the duplication and stops a third-party CleanerListener from silently missing TTL reaps. Falls back to unpersistRDD when spark.cleaner.referenceTracking is off and there is no cleaner, so reaping still works in that configuration. Also pin the full-RDD removal path, since access times are recorded per RDD id but stamped by individual block accesses while a reap removes the whole RDD. The new test asserts every partition's block is gone from the master (not just idle ones), and that the RDD is still usable afterwards and comes back tracked: the reap frees blocks without resetting the RDD's storage level, so it is an eviction rather than an unpersist. Verified: core/scalastyle + core/Test/scalastyle 0 errors; BlockTTLIntegrationSuite (7 tests) and ContextCleanerSuite 21 succeeded / 0 failed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…nd cleaner shutdown The data-loss guard added earlier was defeated by the reap path it guards. rddReapable asked `!persistentRdds.get(rddId).exists(_.isLocallyCheckpointed)`, but reaping calls unpersistRDD, which does persistentRdds.remove(rddId), and RDD.persist only re-registers on the first transition out of StorageLevel.NONE -- which a reap never restores. So: cache, let it be reaped (entry gone), then localCheckpoint (persist(..., allowOverride = true) skips persistRDD), then let it idle -- the gate now reads None, decides the RDD is reapable, and deletes the only copy of the checkpointed data. Reading the live RDD from the cleaner thread was unsound anyway: RDD.checkpointData is not volatile, so a stale None was possible regardless. SparkContext now tracks locally-checkpointed ids in a dedicated concurrent set, written by RDD.localCheckpoint and cleared by unpersistRDD, and the gate reads that. RDD.isLocallyCheckpointed goes back to private[rdd]. Also: - updateBlockInfo stamped an access right after inserting the blockLocations entry, so the containsKey guard was trivially true and a *removal* report (the UpdateBlockInfo(NONE) replies that follow a RemoveRdd broadcast, or a late eviction report) resurrected an access time for an RDD with no blocks -- which one TTL later fires a pointless cluster-wide RemoveRdd. Only stamp when the reported level is valid. - BlockTtlCleaner swallowed the interrupt from shutdownNow when it landed inside a reap: the RPC wraps InterruptedException in a NonFatal SparkException, the throw clears the interrupt flag, and the loop carried on -- leaking a daemon thread per SparkContext in a long-lived JVM. The loop now tests isInterrupted. Reap failures move from logDebug to logWarning: a cleaner that reclaims nothing must not be invisible at the default log level. - updateBlockAtime returns before allocating when neither TTL is set (it runs per block-location lookup on the dispatcher thread, O(partitions) per stage, and both configs are off by default), and matches on the BlockId type instead of allocating two Options. - The unwired reap fallback observes the removal future, matching BlockManagerMaster.removeRdd, instead of discarding it and losing failures. - Document both configs in docs/configuration.md, including that removal is an eviction (the RDD re-caches on next use), that access is observed at the driver so a long single job may not look busy, and that locally checkpointed RDDs are never removed. - Test: use the SparkEnv accessor this PR adds instead of reflecting into RpcEnv internals (deleting a helper copy-pasted from MapOutputTrackerSuite); read atimes through Option, since a ConcurrentHashMap[Int, Long] miss unboxes to 0L and made the refresh assertion able to pass vacuously; wrap a racy tracking assertion in eventually; replace a clock-polling eventually with a plain sleep; drop the unused ResetSystemProperties mixin. Verified: core/scalastyle + core/Test/scalastyle 0 errors; BlockTTLIntegrationSuite, LocalCheckpointSuite, ContextCleanerSuite, MapOutputTrackerSuite 74 succeeded / 0 failed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Removing a shuffle that is still being read does not just cost a map-stage recompute: because executors cache map statuses per epoch, an actively-read shuffle may not look busy, and if the resulting fetch failures repeat they exhaust spark.stage.maxConsecutiveAttempts and fail the job. Setting the TTL safely is the operator's call, so say what the worst case actually is. Docs only; no behavior change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
This introduces two new config parameters for TTL based cleaning which remove blocks which are not used for a given period of time.
Why are the changes needed?
We depend on garbage collection happening to tell us when RDDs/DataFrames are no longer referenced, but when those are defined the global scope those will not be cleaned up. Rather than cleaning up the RDD/DataFrame based on the TTL we only clean up the blocks so that if the TTL is too short and the user does use that RDD or DataFrame again in the future their code will not fail, but instead re-create those blocks.
Does this PR introduce any user-facing change?
New config parameters are introduced. These are smiliar to an old Spark 1.X config parameter around TTLs but instead use reference time rather than a from creation time TTL.
How was this patch tested?
Work in progress.
Was this patch authored or co-authored using generative AI tooling?
Yes, github bot was used for some review, rebase on latest was done by Sonnet 5