[Data] Disable Ray Data operator progress bars when running in Ray job

[scottjlee]

Why are these changes needed?

When using Ray Data in a Ray job, logs are spammed in an incoherent fashion. This is due to how the ray_tqdm module handles multiple progress bars (or rather, how it does not handle them) -- with each progress bar writing independently and the outputs including \r (normally used to overwrite the progress bar in console in real time), this results in empty spaces/lines/overlapping progress bar outputs.

To clean up the logs from Ray jobs, we disable individual operator progress bars, and only keep the "global" overall progress bar for the Ray Dataset. Existing progress bar behavior outside of Ray jobs is unaffected (e.g. using Ray Data outside Ray jobs in console / Jupyter notebooks) and shows all progress bars.

For the following script:

import ray
import time
import os

def f(batch):
    time.sleep(10)
    return batch

paths = ["s3://anonymous@air-example-data/iris.csv"]
ds = ray.data.read_csv(paths, override_num_blocks=20)

ds = ds.map_batches(f)

start_t = time.time()
batch_count = 0
for batch in ds.iter_batches(batch_size=None):
    batch_count += 1

print(f"All done! {time.time() - start_t} seconds, {batch_count} batches")

Output in console, not in Ray Job:

2024-07-26 19:02:44,883 INFO streaming_executor.py:108 -- Starting execution of Dataset. Full logs are in /tmp/ray/session_2024-07-26_12-21-02_283749_7367/logs/ray-data
2024-07-26 19:02:44,883 INFO streaming_executor.py:109 -- Execution plan of Dataset: InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV] -> TaskPoolMapOperator[MapBatches(f)]
✔️  Dataset execution finished in 31.51 seconds: 100%|█████████████████████████████████████████████████████████████| 20/20 [00:31<00:00,  1.58s/ bundle]]
- ReadCSV->SplitBlocks(20): 0 active, 0 queued, [cpu: 0.0, objects: 0.0B]: : 20 bundle [00:31,  1.58s/ bundle]]
- MapBatches(f): 0 active, 0 queued, [cpu: 0.0, objects: 840.0B]: 100%|████████████████████████████████████████████| 20/20 [00:31<00:00,  1.58s/ bundle]
All done! 31.516483783721924 seconds, 20 batches

Output from Ray Job:

2024-07-26 19:03:37,857	INFO streaming_executor.py:108 -- Starting execution of Dataset. Full logs are in /tmp/ray/session_2024-07-26_12-21-02_283749_7367/logs/ray-data
2024-07-26 19:03:37,857	INFO streaming_executor.py:109 -- Execution plan of Dataset: InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV] -> TaskPoolMapOperator[MapBatches(f)]

Running 0: 0 bundle [00:00, ? bundle/s]
Running: 2/10 CPU, 0/0 GPU, 512.0MB/1.0GB object_store_memory: : 0 bundle [00:01, ? bundle/s]
Running: 2/10 CPU, 0/0 GPU, 6.5KB/1.0GB object_store_memory: : 0 bundle [00:11, ? bundle/s]  
Running: 9/10 CPU, 0/0 GPU, 7.7KB/1.0GB object_store_memory: : 0 bundle [00:21, ? bundle/s]
Running: 9/10 CPU, 0/0 GPU, 7.7KB/1.0GB object_store_memory:   0%|          | 0/20 [00:21<?, ? bundle/s]
Running: 9/10 CPU, 0/0 GPU, 7.7KB/1.0GB object_store_memory:  15%|█▌        | 3/20 [00:21<01:59,  7.01s/ bundle]
Running: 7/10 CPU, 0/0 GPU, 4.0KB/1.0GB object_store_memory:  15%|█▌        | 3/20 [00:22<01:59,  7.01s/ bundle]
Running: 7/10 CPU, 0/0 GPU, 4.0KB/1.0GB object_store_memory:  65%|██████▌   | 13/20 [00:22<00:09,  1.30s/ bundle]
Running: 6/10 CPU, 0/0 GPU, 3.7KB/1.0GB object_store_memory:  65%|██████▌   | 13/20 [00:31<00:09,  1.30s/ bundle]
✔️  Dataset execution finished in 32.02 seconds:  65%|██████▌   | 13/20 [00:32<00:09,  1.30s/ bundle]            
✔️  Dataset execution finished in 32.02 seconds: 100%|██████████| 20/20 [00:32<00:00,  1.36s/ bundle]
✔️  Dataset execution finished in 32.02 seconds: 100%|██████████| 20/20 [00:32<00:00,  1.60s/ bundle]
All done! 32.02574801445007 seconds, 20 batches

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 scripts/format.sh to lint the changes in this PR.
• 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 :(

[raulchen]

can we instead make the default enable_progress_bars False for ray jobs?
So users can still set it to True in case needed.

[scottjlee]

always enables the global progress bar.