code: def test_sort_with_progress_bar(ray_start_1_cpu):
      npartitions = 10
      df = dd.from_pandas(
          pd.DataFrame(
              np.random.randint(0, 100, size=(100, 2)),
              columns=["age", "grade"],
          ),
          npartitions=npartitions,
      )
      # We set max_branch=npartitions in order to ensure that the task-based
      # shuffle happens in a single stage, which is required in order for our
      # optimization to work.
      sorted_with_pb = None
      sorted_without_pb = None
      with ProgressBarCallback():
          sorted_with_pb = df.set_index(
              ["age"], shuffle="tasks", max_branch=npartitions
          ).compute(
              scheduler=ray_dask_get, _ray_enable_progress_bar=True
          )
      sorted_without_pb = df.set_index(
          ["age"], shuffle="tasks", max_branch=npartitions
      ).compute(scheduler=ray_dask_get)
      assert sorted_with_pb.equals(sorted_without_pb)
  
file_path: ray/python/ray/util/dask/tests/test_dask_scheduler.py
imports: import sys
  import unittest
  import dask
  import dask.array
  import dask.dataframe
  import numpy
  import pandas
  import pytest
  import ray
  from ray.tests.conftest import *
  from ray.util.client.common import ClientObjectRef
  from ray.util.dask import disable_dask_on_ray, enable_dask_on_ray, ray_dask_get
  from ray.util.dask.callbacks import ProgressBarCallback
label_desc: Use image node_id 6 for calling a global function with example usage: test_sort_with_progress_bar(ray_start_1_cpu) without return types