code: def _prepare_split(
      self,
      split_generator: "datasets.SplitGenerator",
      file_format: str = "arrow",
      max_shard_size: Optional[Union[str, int]] = None,
      num_proc: Optional[int] = None,
      **kwargs,
  ):
      self._validate_cache_dir()
  
      max_shard_size = convert_file_size_to_int(
          max_shard_size or MAX_SHARD_SIZE
      )
      self._repartition_df_if_needed(max_shard_size)
      is_local = not is_remote_filesystem(self._fs)
      path_join = os.path.join if is_local else posixpath.join
  
      SUFFIX = "-TTTTT-SSSSS-of-NNNNN"
      fname = (
          f"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
      )
      fpath = path_join(self._output_dir, fname)
  
      total_num_examples = 0
      total_num_bytes = 0
      total_shards = 0
      task_id_and_num_shards = []
      all_shard_lengths = []
  
      for task_id, content in self._prepare_split_single(
          fpath, file_format, max_shard_size
      ):
          (
              num_examples,
              num_bytes,
              num_shards,
              shard_lengths,
          ) = content
          if num_bytes > 0:
              total_num_examples += num_examples
              total_num_bytes += num_bytes
              total_shards += num_shards
              task_id_and_num_shards.append((task_id, num_shards))
              all_shard_lengths.extend(shard_lengths)
  
      split_generator.split_info.num_examples = total_num_examples
      split_generator.split_info.num_bytes = total_num_bytes
  
      # should rename everything at the end
      logger.debug(f"Renaming {total_shards} shards.")
      if total_shards > 1:
          split_generator.split_info.shard_lengths = all_shard_lengths
  
          # Define fs outside of _rename_shard so that we don't reference self in the function, which will result in a
          # pickling error due to pickling the SparkContext.
          fs = self._fs
  
          # use the -SSSSS-of-NNNNN pattern
          def _rename_shard(
              task_id: int,
              shard_id: int,
              global_shard_id: int,
          ):
              rename(
                  fs,
                  fpath.replace("SSSSS", f"{shard_id:05d}").replace(
                      "TTTTT", f"{task_id:05d}"
                  ),
                  fpath.replace(
                      "TTTTT-SSSSS", f"{global_shard_id:05d}"
                  ).replace("NNNNN", f"{total_shards:05d}"),
              )
  
          args = []
          global_shard_id = 0
          for i in range(len(task_id_and_num_shards)):
              task_id, num_shards = task_id_and_num_shards[i]
              for shard_id in range(num_shards):
                  args.append([task_id, shard_id, global_shard_id])
                  global_shard_id += 1
          self._spark.sparkContext.parallelize(args, len(args)).map(
              lambda args: _rename_shard(*args)
          ).collect()
      else:
          # don't use any pattern
          shard_id = 0
          task_id = task_id_and_num_shards[0][0]
          self._rename(
              fpath.replace("SSSSS", f"{shard_id:05d}").replace(
                  "TTTTT", f"{task_id:05d}"
              ),
              fpath.replace(SUFFIX, ""),
          )
  
imports: import os
  import posixpath
  import uuid
  from dataclasses import dataclass
  from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union
  import numpy
  import pyarrow
  import datasets
  from datasets.arrow_writer import ArrowWriter, ParquetWriter
  from datasets.config import MAX_SHARD_SIZE
  from datasets.filesystems import is_remote_filesystem, rename
  from datasets.iterable_dataset import _BaseExamplesIterable
  from datasets.utils.py_utils import convert_file_size_to_int
label_desc: Use image node_id 7 for calling the Spark obj's underlying member method code with example usage: obj._prepare_split(split_generator, file_format, max_shard_size, num_proc) without return types