[data] Refractor progress bar clearer metrics

[iamjustinhsu]

Before:
https://github.com/user-attachments/assets/9db00f37-0c37-4e99-874a-a14481878e4a
In before, the progress bar won't update until the first tasks finishes.

~~After:
https://github.com/user-attachments/assets/99877a3f-7b52-4293-aae5-7702edfaabec

In After, the progress bar won't update until the first task generates output. If a task generates 10 blocks, we will update the progress bar while it's generating blocks, even if the task hasn't finished. Once the task finishes, we default back to the way it was before.

This is better because the very 1st progress bar update will occur sooner, and won't feel abrupt to the user.

Refractoring the progress bar estimates using known metrics.

Why are these changes needed?

Currently we use number of finished tasks. This is OK, but since we use streaming geneator, 1 task = thousands of blocks. This is troublesome for additional split factor (split blocks) in read parquet

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run pre-commit jobs to lint the changes in this PR. (pre-commit setup)
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[iamjustinhsu]

not being used anywhere

[iamjustinhsu]

drive-by

[bveeramani]

@iamjustinhsu I felt confused after reading the PR description about what the change is, and why it's an improvement. Could you update the description to make it clearer?

Is this understanding correct?

• Before: Ray Data doesn't render a progress bar if no tasks have finished because it can't estimate how many rows each task will produce, and therefore can't estimate the total number of rows the operator will produce.
• After: Ray Data uses the number of rows already outputted as an estimate of the total number of rows that will be produced by the operator.

And this is better because it appears smoother (?)

[bveeramani]

@iamjustinhsu what's the code you used for the example recordings?

[iamjustinhsu]

The code is above the sample recordings:

import time

def random_blocks(x):
    bits_to_allocate = 1 * 1024 * 1024
    for i in range(5):
        time.sleep(1)
        yield {'item': [0] * bits_to_allocate} # After the first yield, it the progress bar will update

ray.data.DataContext.get_current().target_max_block_size = 1 * 1024 * 1024
ray.data.range(100, override_num_blocks=2).map_batches(random_blocks, concurrency=1).materialize()

[bveeramani]

Gotcha.

I'm worried that some users might feel confused if they see the progress bar stuck at 100% for a while before it abruptly updates to <100%. Might be clearer to show no progress bar than an inaccurate one, though I can understand the argument that always showing a progress bar might look nicer.

@alexeykudinkin what's your take?

[iamjustinhsu]

@bveeramani for more context, this was mainly annoying because of ReadParquet->SplitBlocks(N), because you don't get any updates until after the 1st task finishes. If N is large (which I have seen to be in the several hundreds), nothing is updating since SplitBlocks is in a single task.

[bveeramani]

@bveeramani for more context, this was mainly annoying because of ReadParquet->SplitBlocks(N), because you don't get any updates until after the 1st task finishes. If N is large (which I have seen to be in the several hundreds), nothing is updating since SplitBlocks is in a single task.

Yeah, this can happen whenever tasks produce multiple outputs, which can be especially common with something like split blocks

[alexeykudinkin]

Hold on, this is what we estimate total # of rows/bundles to be not what has been gen'd so far, right?

[iamjustinhsu]

_estimated_num_output_bundles and _estimated_output_num_rows estimate the total number. But as a crude heuristic, I wanted these to update sooner, rather than later. These variables are mainly used for user facing progress indicators (like in progress bar). After the 1st task finishes, the progress bar will behave as before.

[alexeykudinkin]

But these values don't make sense, right?

As reader of the code i'm scratching my head as we're now messing up their semantic

[iamjustinhsu]

i understand, if the name of variable is named estimation, then semantics imply that it's not exact but a guess. Whether that guess is wrong, is also intrinsic for the current estimation. IMO a crude estimation is still better than None. Incidentally, I had originally thought to do something like this:

if self._metrics.num_tasks_finished == 0:
      estimated_num_tasks = (
            self.upstream_op_num_outputs
            / metrics.num_inputs_received
            * num_tasks_submitted
      )
      ratio = estimated_num_tasks / self._metrics.num_tasks_running
      self._estimated_num_output_bundles = (
                self._metrics.num_task_outputs_generated * ratio
      )
      self._estimated_output_num_rows = (
            self._metrics.rows_task_outputs_generated * ratio
     )

but at this point the estimation complexity outweighs it's temporal use when num_tasks_finished = 0.

Regardless, I'm gonna change the intent of the PR to refractor stuff instead, because there are still stuff that needs to change in this PR.

[bveeramani]

If these metrics are None, this code will raise a TypeError:

• average_num_inputs_per_task
• average_num_outputs_per_task
• average_rows_outputs_per_task

This can't happen right now because we check num_tasks_finished > 0, and the three metrics aren't None when num_tasks_finished > 0. But, that's an implementation detail, and isn't guaranteed by their interface.

This code would be more robust if replaced this check:

        and metrics.num_inputs_received > 0
        and metrics.num_tasks_finished > 0

With explicit checks that the metrics aren't None.