code: def divide_files(self, folder, sizes):
      import glob
  
      # Each file is of format: /path/to/processed/part-id-***.csv
      files = glob.glob(folder + "/*.csv")
      file_with_sizes = [
          (file, sizes[int(file.split("/")[-1].split("-")[1])])
          for file in files
      ]
      num_files_per_worker = len(files) // self.num_workers
      num_remain_files = len(files) % self.num_workers
      extra_files = []
      if num_remain_files > 0:
          extra_files = file_with_sizes[-num_remain_files:]
      files_dict = dict()
      for worker in self.remote_workers:
          index = ray.get(worker.get_task_index.remote())
          worker_files = file_with_sizes[
              index
              * num_files_per_worker : (index + 1)
              * num_files_per_worker
          ]
          if extra_files:
              worker_files += [extra_files.pop()]
          files_dict[index] = worker_files
      # key is worker id; value is list of (file, size) tuple.
      return files_dict
  
imports: import time
  import argparse
  import tensorflow
  import os
  import sys
  import math
  import collections
  from tensorflow.python.client import timeline
  import json
  from tensorflow.python.ops import partitioned_variables
  import ray
  from bigdl.dllib.utils.log4Error import *
  from bigdl.orca import init_orca_context, stop_orca_context, OrcaContext
  from bigdl.orca.data.utils import process_spark_xshards
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards
  from bigdl.friesian.feature import FeatureTable
label_desc: Use image node_id 4 for calling the RayDeepRecCluster obj's underlying member method code with example usage: obj.divide_files(folder, sizes) and returns: files_dict