code: def load_dataset_files(
      client, data_dir, file_path, nbytes, npartitions
  ):
      num_bytes_per_partition = nbytes // npartitions
      filenames = []
  
      @dask.delayed
      def generate_file(i, data_dir, file_path):
          key = "{}/df-{}-{}.parquet.gzip".format(
              file_path, num_bytes_per_partition, i
          )
          from os import path
  
          if path.exists(key):
              print(f"The file {key} already exists. Do nothing")
              return
  
          filename = os.path.join(data_dir, key)
          if not os.path.exists(filename):
              print("Generating partition", filename)
              nrows = num_bytes_per_partition // 8
              dataset = pd.DataFrame(
                  np.random.randint(
                      0,
                      np.iinfo(np.int64).max,
                      size=(nrows, 1),
                      dtype=np.int64,
                  ),
                  columns=["a"],
              )
              dataset.to_parquet(filename, compression="gzip")
          print("Writing partition to a file", filename)
  
      x = []
      for i in range(npartitions):
          x.append(generate_file(i, data_dir, file_path))
      # from ray.util.dask import ProgressBarCallback
      # with ProgressBarCallback():
      #     dask.compute(x, _ray_enable_progress_bar=True)
      dask.compute(x)
  
      filenames = [
          f"{file_path}/df-{num_bytes_per_partition}-{i}.parquet.gzip"
          for i in range(npartitions)
      ]
      df = dd.read_parquet(filenames)
      return df
  
file_path: ray/release/nightly_tests/dask_on_ray/dask_on_ray_sort.py
imports: import csv
  import json
  import os.path
  import time
  import boto3
  import dask
  import dask.dataframe
  import numpy
  import pandas
  from dask.distributed import Client
  import traceback
label_desc: Use image node_id 2 for calling a global function with example usage: load_dataset_files(client, data_dir, file_path, nbytes, npartitions) and returns: df