Shuffle tracking prevents dynamic allocation from removing executors after `collect()` / `toPandas()`

[aalopatin]

When using dynamic allocation with shuffle tracking enabled, executors that participated in a shuffle remain alive indefinitely after a collect() or toPandas() action has completed.

Configuration:

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.executorIdleTimeout=60s
spark.dynamicAllocation.cachedExecutorIdleTimeout=1m
spark.dynamicAllocation.shuffleTracking.enabled=true
spark.dynamicAllocation.shuffleTracking.timeout=infinity

For example:

pandas_df = spark.sql("...").toPandas()

After the action completes, the executors that produced shuffle data are not removed after spark.dynamicAllocation.executorIdleTimeout.

The same behavior can be reproduced with:

df.collect()

I also observed that shuffle files are present in $SPARK_LOCAL_DIRS on the executor pods (shuffle_*.data, shuffle_*.index, etc.).

Interestingly, setting a finite value for:

spark.dynamicAllocation.shuffleTracking.timeout

allows the executors to be removed, which suggests that shuffle tracking is what prevents their removal.

My question is whether this behavior is expected.

collect() and toPandas() are terminal actions that materialize their result on the driver. Once the action has completed successfully, there should presumably be no further consumer of the shuffle output produced by the query.

Therefore, I would expect the executors to become eligible for removal after spark.dynamicAllocation.executorIdleTimeout, even if shuffle files produced during the query still physically exist on the executors.

Is there a reason why the shuffle is still considered active after the collect() / toPandas() action has completed? Or is this an expected limitation/behavior of shuffle tracking?

For comparison, setting:

spark.dynamicAllocation.shuffleTracking.timeout=1m

allows the executors to be removed, whereas:

spark.dynamicAllocation.shuffleTracking.timeout=infinity

keeps them alive.

[rcpeken]

Hi, I'd like to investigate this issue.

I'll first trace how shuffle tracking determines whether an executor is still needed after terminal actions such as collect() and toPandas(), and check whether the current behavior is intentional.

If the behavior is not expected, I'd like to work on a fix and add regression coverage for dynamic allocation with shuffle tracking enabled.

Thanks!

[rcpeken]

I traced this through ExecutorMonitor and it looks like expected (documented) behavior rather than a bug.

When an executor writes shuffle output, the shuffle id is added to its shuffleIds set. onJobEnd only clears
hasActiveShuffle; the id itself is removed only in cleanupShuffle, which runs on ShuffleCleanedEvent, i.e.
after the ShuffleDependency has been garbage collected on the driver and ContextCleaner removes the shuffle.

While shuffleIds is non-empty, updateTimeout() uses max(idleTimeout, shuffleTimeout), so with
spark.dynamicAllocation.shuffleTracking.timeout=infinity the executor never times out until driver GC cleans the
shuffle. collect() / toPandas() being terminal doesn't change this: the DataFrame (and its plan) can still be
re-executed and reuse the shuffle output, so Spark can't release it at the end of the action.

The config docs describe this: "The default value means that Spark will rely on the shuffles being garbage
collected to be able to release executors."

Workarounds:

• set a finite spark.dynamicAllocation.shuffleTracking.timeout (as you observed), or
• make driver GC run more often, e.g. lower spark.cleaner.periodicGC.interval (default 30min), and drop
  references to DataFrames you no longer need.

If maintainers think it's useful, I'd be happy to open a small docs PR clarifying that shuffle data from completed
terminal actions is only released after driver GC.