[data] log progress bar to data logs

[Zandew]

Why are these changes needed?

Log each operator's progress bar to ray-data.log after each _scheduling_loop_step if DEBUG_TRACE_SCHEDULING is not set.

Example: ray.data.range(100).map_batches(lambda x: x).materialize():

2023-10-31 15:41:26,646 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:26,646 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:26,646 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 10 active, 10 queued, 0.02 MiB objects, Blocks Outputted: 0/20
2023-10-31 15:41:26,646 INFO streaming_executor.py:430 --
2023-10-31 15:41:26,646 INFO streaming_executor.py:312 -- Operator InputDataBuffer[Input] completed. Operator Metrics:
{'num_inputs_received': 20, 'bytes_inputs_received': 45900, 'num_outputs_taken': 20, 'bytes_outputs_taken': 45900, 'cpu_usage': 0, 'gpu_usage': 0}
2023-10-31 15:41:26,757 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:26,757 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:26,757 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 10 active, 9 queued, 0.02 MiB objects, Blocks Outputted: 1/20
2023-10-31 15:41:26,757 INFO streaming_executor.py:430 --
2023-10-31 15:41:26,866 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:26,866 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:26,866 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 10 active, 8 queued, 0.02 MiB objects, Blocks Outputted: 2/20
2023-10-31 15:41:26,866 INFO streaming_executor.py:430 --
2023-10-31 15:41:26,972 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:26,972 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:26,972 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 10 active, 7 queued, 0.02 MiB objects, Blocks Outputted: 3/20
2023-10-31 15:41:26,972 INFO streaming_executor.py:430 --
2023-10-31 15:41:27,079 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:27,079 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:27,079 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 10 active, 6 queued, 0.02 MiB objects, Blocks Outputted: 4/20
2023-10-31 15:41:27,079 INFO streaming_executor.py:430 --
2023-10-31 15:41:27,187 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:27,188 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:27,188 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 10 active, 5 queued, 0.02 MiB objects, Blocks Outputted: 5/20
2023-10-31 15:41:27,188 INFO streaming_executor.py:430 --
2023-10-31 15:41:27,290 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:27,290 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:27,290 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 5 active, 0 queued, 0.01 MiB objects, Blocks Outputted: 15/20
2023-10-31 15:41:27,290 INFO streaming_executor.py:430 --
2023-10-31 15:41:27,293 INFO streaming_executor.py:424 -- Execution Progress:
2023-10-31 15:41:27,293 INFO streaming_executor.py:426 -- 0: - Input: 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:27,293 INFO streaming_executor.py:426 -- 1: - ReadRange->MapBatches(<lambda>): 0 active, 0 queued, 0.0 MiB objects, Blocks Outputted: 20/20
2023-10-31 15:41:27,293 INFO streaming_executor.py:430 --

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 :(

[c21]

cc @raulchen as well.