andygrove opened a new issue, #6727:
URL: https://github.com/apache/datafusion-comet/issues/6727
### What is the problem the feature request solves?
Sizing `spark.executor.memoryOverhead` for Comet is a guess made before the
job runs. The tuning guide says the memory Comet allocates without reserving it
(per-batch working memory, Parquet and decompression buffers, allocator
fragmentation and retained pages, JVM-side Arrow buffers) has to fit in the
overhead, and that executors that outgrow it are killed with exit code 137.
Measurements in #4576 found jemalloc's resident memory at 6x to 8x `allocated`
at the median, so the right overhead depends on the workload and is hard to
know up front. Operators over-provision every executor for the worst case, or
under-provision and lose executors mid-query.
Spark 4.2.0 adds a built-in
`org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin` (SPARK-55432)
that uses Kubernetes in-place pod resize (beta in 1.33, GA in 1.35) to grow an
executor's memory without restarting it. This looks like a good fit for the
untracked part of Comet's footprint, and nobody has looked at it from Comet's
side yet.
### What the Spark plugin does
Read from `branch-4.2` of `ExecutorResizePlugin.scala`:
- It runs in the driver, registered through `spark.plugins`. Despite the
docs showing `spark.kubernetes.executor.resizeInterval` as `0s`, the config
default in code is `1m`.
- Every interval it reads each executor container's memory usage from the
Kubernetes metrics API (metrics-server) and compares it with the container's
memory limit.
- If `usage > resizeThreshold * limit` (default `0.9`), it patches the pod's
`resize` subresource, setting both the memory request and limit to `limit * (1
+ resizeFactor)` (default `0.1`).
- It only grows memory. There is no scale-down, and 4.2.0 has no cap. A
`resizeMaxMemory` cap and support for the `deployment` pods allocator are in
apache/spark#59180. In 4.2.0 the plugin only starts with the `direct` pods
allocator.
- Spark's own pools are not resized. Heap and `spark.memory.offHeap.size`
are fixed when the JVM starts, so the extra container memory is headroom for
allocations that no Spark pool accounts for. That is exactly the memory Comet
cannot size ahead of time.
### Why this might help Comet
- A smaller baseline overhead, with the plugin as a safety net that grows it
on demand, instead of sizing every executor for the worst case.
- Fewer exit-137 executor losses when Comet's untracked memory exceeds a
static overhead.
- It needs no Comet code to try. Comet already documents `memoryOverhead` as
the place this memory lives, so the first step is just measuring whether the
combination works.
### Open questions and likely gaps
1. **Reaction time.** The plugin polls every minute by default and grows 10%
of the limit per step. Native memory in a sort, join, or shuffle can climb
faster than that. How often does an executor die before the plugin reacts, and
does a shorter interval or larger factor fix it?
2. **What the plugin measures.** It sees the container working set from
metrics-server, which includes memory the allocator retains but could return,
and it never shrinks. A long-running application could ratchet its limit up
after one bad stage and keep it. Comet could reduce this by purging or decaying
allocator pages between plans, which depends on the allocator work in #6167 and
#4576.
3. **A stale overhead in the warning.** `CometExecIterator` computes the
container's native budget from `spark.executor.memoryOverhead` (or the derived
default) and warns when native memory exceeds it. After an in-place resize the
real limit is larger than that, so the warning can fire on an executor that is
fine. Reading the actual cgroup limit (`memory.max` on cgroup v2) instead of
the conf would be correct with and without resizing.
4. **A Comet-side trigger.** The executor knows more than metrics-server
does: `RssAnon` (proposed in #6167), `allocated`, `reserved`, and the pool
state. A Comet `SparkPlugin` could request a resize from the executor, either
by patching its own pod (needs RBAC for `pods/resize`) or by sending the
request to the driver over the plugin RPC channel and having the driver patch
it. This would react faster and use a better signal than the generic plugin, at
the cost of more code. It is worth building only if the experiment shows the
generic plugin is too slow or too coarse.
5. **Scale down.** The Spark plugin never gives memory back. Lowering a
memory limit in place is only safe once usage has dropped below the new limit,
so a Comet-aware plugin would need to coordinate with allocator purging. This
is probably out of scope for a first pass.
6. **Prerequisites and failure modes.** The cluster needs Kubernetes 1.33+
(1.35 for GA), metrics-server, and node capacity. A resize can be deferred or
infeasible if the node is full, and the failure is only logged by the plugin.
The docs should say what the user can expect when that happens.
7. **Interaction with other memory work.** This is complementary to #5116
(per-stage ResourceProfiles), which changes what is requested at stage
boundaries, while resize changes it during a stage. It also overlaps with the
circuit-breaker idea in #4576, which is about acting on RSS inside the process.
### Describe the potential solution
1. **Experiment first.** On a Kubernetes 1.35 cluster (kind or minikube is
enough) with Spark 4.2.0 and a current Comet build, run a TPC-H or TPC-DS
subset that is known to need a large untracked footprint, with
`spark.executor.memoryOverhead` set well below what the memory usage log says
is needed, in three configurations:
- low overhead, no plugin
- low overhead,
`spark.plugins=org.apache.spark.scheduler.cluster.k8s.ExecutorResizePlugin`
- correctly sized overhead, no plugin
Record whether the query completes, the number of executor losses and
exit codes, the resize events and time from threshold crossing to resize, final
limit versus peak `RssAnon`, wall-clock time, and total memory-seconds. Repeat
with a shorter `resizeInterval` and a larger `resizeFactor`.
2. **Fix the stale limit in the native memory warning** (open question 3),
independent of the experiment's outcome.
3. **Document it** in the tuning guide and Kubernetes guide if the
experiment shows it helps: when to use it, the recommended settings, the
Kubernetes version and metrics-server prerequisites, and the `direct` allocator
limit in 4.2.0.
4. **Decide on a Comet-side plugin** (open question 4) from the experiment
results.
### Additional context
- Spark: SPARK-55432 (`ExecutorResizePlugin`), SPARK-59918 (`deployment`
allocator support), [Spark 4.2.0 release
notes](https://spark.apache.org/releases/spark-release-4-2-0.html), [Kubernetes
1.35 in-place resize
GA](https://kubernetes.io/blog/2025/12/19/kubernetes-v1-35-in-place-pod-resize-ga).
- Related Comet issues: #6167 (report `RssAnon` in the memory usage log),
#4576 (allocator-level accounting and the RSS circuit breaker), #5116
(ResourceProfiles for native versus JVM stages), #6190 (Kubernetes guide
example never enables Comet).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
