[Data] Backlog-aware Actor Autoscaler

[rayhhome]

Description

This PR implements BacklogAwareActorAutoscaler, the backlog-aware actor-pool autoscaler. Where DefaultActorAutoscaler sizes the upscale delta proportionally to the pool's current size, this autoscaler sizes it from the work already visible: it converts the operator's enqueued input blocks into a task count, adds the in-flight tasks, and solves for the pool size that brings utilization back under the upscaling threshold in a single step.

RAY_DATA_ACTOR_AUTOSCALER=DEFAULT restores the previous autoscaler.

Additional information

Adds backlog_aware_actor_autoscaler.py and registers BACKLOG_AWARE in a new ActorAutoscalerVersion, following the RAY_DATA_CLUSTER_AUTOSCALER precedent.

Enabling it also requires taking a second variable: DEFAULT_ACTOR_POOL_MAX_UPSCALING_DELTA now defaults to None. If we keep the old value of 1, this cap will truncate every scaling decision to a single actor, which would make the backlog-derived delta unobservable.

[gemini-code-assist]

Code Review

This pull request introduces a new BacklogAwareActorAutoscaler that sizes the actor pool based on the enqueued input backlog rather than just current utilization, allowing the pool to scale up to the required size in a single step. It also adds configuration options to select the autoscaler version via an environment variable and includes comprehensive unit tests. The review feedback highlights several code quality improvements, including removing an unused logging import and logger instance, removing a redundant override of _compute_downscale_delta, and correcting the return type annotation of _estimate_expected_tasks from float to int.

[gemini-code-assist]

The function _estimate_expected_tasks returns an int (since math.ceil returns an int in Python 3), but its return type is annotated as float. Please update the return type annotation to int.

Suggested change

	def _estimate_expected_tasks(
	op_state: "OpState",
	) -> float:
	def _estimate_expected_tasks(
	op_state: "OpState",
	) -> int:

[Copilot]

🟢 Approval recommended

The change is opt-in, well-scoped to the actor autoscaler internals, and includes targeted test coverage, with only a minor type-annotation nit noted.

Pull request overview

Ports an internal backlog-aware actor-pool autoscaler to open source Ray Data, enabling opt-in scaling decisions based on observed backlog (enqueued input blocks) plus in-flight tasks, selected via RAY_DATA_ACTOR_AUTOSCALER=BACKLOG_AWARE.

Changes:

• Add BacklogAwareActorAutoscaler, computing scale-up deltas from backlog + in-flight tasks to reach the target utilization threshold in one step.
• Add unit tests covering backlog-derived sizing, budget/allocation constraints, and create_actor_autoscaler version selection.
• Register the new autoscaler version behind RAY_DATA_ACTOR_AUTOSCALER and add a Bazel py_test target.

File summaries

File	Description
python/ray/data/tests/test_backlog_aware_actor_autoscaler.py	New tests for backlog-aware scaling behavior and env-based autoscaler selection.
python/ray/data/BUILD.bazel	Adds Bazel py_test target for the new test module.
python/ray/data/_internal/actor_autoscaler/backlog_aware_actor_autoscaler.py	Introduces the backlog-aware autoscaler implementation and backlog→task estimation helper.
python/ray/data/_internal/actor_autoscaler/init.py	Adds ActorAutoscalerVersion and selects autoscaler implementation via RAY_DATA_ACTOR_AUTOSCALER.

Review details

• Files reviewed: 4/4 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 39715ff. Configure here.}

[cursor]

Unknown average over-scales actor pool

Medium Severity

_estimate_expected_tasks treats a missing average_num_inputs_per_task as one block per task. That metric stays None until the first task finishes, so a queued map_batches workload is counted as far more tasks than it will become. With BACKLOG_AWARE and an uncapped actor_pool_max_upscaling_delta now the defaults, the first scale-up can launch a large burst of actors that later have to be shed one at a time.

 

^{Reviewed by Cursor Bugbot for commit 39715ff. Configure here.}