import collections
import os
import re
from pathlib import Path


a_    					=       'src/transformers'


# Matches is_xxx_available()
a_    					=       re.compile(r'is\_([a-z_]*)_available()')
# Catches a one-line _import_struct = {xxx}
a_    					=       re.compile(r'^_import_structure\s+=\s+\{([^\}]+)\}')
# Catches a line with a key-values pattern: "bla": ["foo", "bar"]
a_    					=       re.compile(r'\s+"\S*":\s+\[([^\]]*)\]')
# Catches a line if not is_foo_available
a_    					=       re.compile(r'^\s*if\s+not\s+is\_[a-z_]*\_available\(\)')
# Catches a line _import_struct["bla"].append("foo")
a_    					=       re.compile(r'^\s*_import_structure\["\S*"\]\.append\("(\S*)"\)')
# Catches a line _import_struct["bla"].extend(["foo", "bar"]) or _import_struct["bla"] = ["foo", "bar"]
a_    					=       re.compile(r'^\s*_import_structure\[\S*\](?:\.extend\(|\s*=\s+)\[([^\]]*)\]')
# Catches a line with an object between quotes and a comma:     "MyModel",
a_    					=       re.compile(r'^\s+"([^"]+)",')
# Catches a line with objects between brackets only:    ["foo", "bar"],
a_    					=       re.compile(r'^\s+\[([^\]]+)\]')
# Catches a line with from foo import bar, bla, boo
a_    					=       re.compile(r'\s+from\s+\S*\s+import\s+([^\(\s].*)\n')
# Catches a line with try:
a_    					=       re.compile(r'^\s*try:')
# Catches a line with else:
a_    					=       re.compile(r'^\s*else:')





def    __lowercase     (						lowerCamelCase					:     Dict	):
 if _re_test_backend.search(lowerCamelCase	) is None:
  return None
 UpperCamelCase_					:						int =					[b[0] for b in _re_backend.findall(lowerCamelCase	)]
 backends.sort()
 return "_and_".join(lowerCamelCase	)





def    __lowercase     (						lowerCamelCase					:     Optional[int]	):
 with open(lowerCamelCase  ,  'r'  ,  encoding='utf-8'  ,  newline='\n'	) as f:
  UpperCamelCase_					:						Optional[Any] =					f.readlines()

 UpperCamelCase_					:						Optional[Any] =					0
 while line_index < len(lowerCamelCase	) and not lines[line_index].startswith('_import_structure = {'	):
  line_index += 1

 # If this is a traditional init, just return.
 if line_index >= len(lowerCamelCase	):
  return None

 # First grab the objects without a specific backend in _import_structure
 UpperCamelCase_					:						Tuple =					[]
 while not lines[line_index].startswith('if TYPE_CHECKING'	) and find_backend(lines[line_index]	) is None:
  UpperCamelCase_					:						str =					lines[line_index]
  # If we have everything on a single line, let's deal with it.
  if _re_one_line_import_struct.search(lowerCamelCase	):
   UpperCamelCase_					:						int =					_re_one_line_import_struct.search(lowerCamelCase	).groups()[0]
   UpperCamelCase_					:						List[Any] =					re.findall(R'\[([^\]]+)\]'  ,  lowerCamelCase	)
   for imp in imports:
    objects.extend([obj[1:-1] for obj in imp.split(', '	)]	)
   line_index += 1
   continue
  UpperCamelCase_					:						List[str] =					_re_import_struct_key_value.search(lowerCamelCase	)
  if single_line_import_search is not None:
   UpperCamelCase_					:						Optional[Any] =					[obj[1:-1] for obj in single_line_import_search.groups()[0].split(', '	) if len(lowerCamelCase	) > 0]
   objects.extend(lowerCamelCase	)
  elif line.startswith(' ' * 8 + '"'	):
   objects.append(line[9:-3]	)
  line_index += 1

 UpperCamelCase_					:						Any =					{'none': objects}
 # Let's continue with backend-specific objects in _import_structure
 while not lines[line_index].startswith('if TYPE_CHECKING'	):
  # If the line is an if not is_backend_available, we grab all objects associated.
  UpperCamelCase_					:						Any =					find_backend(lines[line_index]	)
  # Check if the backend declaration is inside a try block:
  if _re_try.search(lines[line_index - 1]	) is None:
   UpperCamelCase_					:						Dict =					None

  if backend is not None:
   line_index += 1

   # Scroll until we hit the else block of try-except-else
   while _re_else.search(lines[line_index]	) is None:
    line_index += 1

   line_index += 1

   UpperCamelCase_					:						Any =					[]
   # Until we unindent, add backend objects to the list
   while len(lines[line_index]	) <= 1 or lines[line_index].startswith(' ' * 4	):
    UpperCamelCase_					:						int =					lines[line_index]
    if _re_import_struct_add_one.search(lowerCamelCase	) is not None:
     objects.append(_re_import_struct_add_one.search(lowerCamelCase	).groups()[0]	)
    elif _re_import_struct_add_many.search(lowerCamelCase	) is not None:
     UpperCamelCase_					:						Dict =					_re_import_struct_add_many.search(lowerCamelCase	).groups()[0].split(', '	)
     UpperCamelCase_					:						int =					[obj[1:-1] for obj in imports if len(lowerCamelCase	) > 0]
     objects.extend(lowerCamelCase	)
    elif _re_between_brackets.search(lowerCamelCase	) is not None:
     UpperCamelCase_					:						Optional[Any] =					_re_between_brackets.search(lowerCamelCase	).groups()[0].split(', '	)
     UpperCamelCase_					:						int =					[obj[1:-1] for obj in imports if len(lowerCamelCase	) > 0]
     objects.extend(lowerCamelCase	)
    elif _re_quote_object.search(lowerCamelCase	) is not None:
     objects.append(_re_quote_object.search(lowerCamelCase	).groups()[0]	)
    elif line.startswith(' ' * 8 + '"'	):
     objects.append(line[9:-3]	)
    elif line.startswith(' ' * 12 + '"'	):
     objects.append(line[13:-3]	)
    line_index += 1

   UpperCamelCase_					:						int =					objects
  else:
   line_index += 1

    # At this stage we are in the TYPE_CHECKING part, first grab the objects without a specific backend
 UpperCamelCase_					:						Optional[Any] =					[]
 while (
     line_index < len(lowerCamelCase	)
     and find_backend(lines[line_index]	) is None
     and not lines[line_index].startswith('else'	)
 ):
  UpperCamelCase_					:						Tuple =					lines[line_index]
  UpperCamelCase_					:						int =					_re_import.search(lowerCamelCase	)
  if single_line_import_search is not None:
   objects.extend(single_line_import_search.groups()[0].split(', '	)	)
  elif line.startswith(' ' * 8	):
   objects.append(line[8:-2]	)
  line_index += 1

 UpperCamelCase_					:						int =					{'none': objects}
 # Let's continue with backend-specific objects
 while line_index < len(lowerCamelCase	):
  # If the line is an if is_backend_available, we grab all objects associated.
  UpperCamelCase_					:						Optional[Any] =					find_backend(lines[line_index]	)
  # Check if the backend declaration is inside a try block:
  if _re_try.search(lines[line_index - 1]	) is None:
   UpperCamelCase_					:						Any =					None

  if backend is not None:
   line_index += 1

   # Scroll until we hit the else block of try-except-else
   while _re_else.search(lines[line_index]	) is None:
    line_index += 1

   line_index += 1

   UpperCamelCase_					:						Union[str, Any] =					[]
   # Until we unindent, add backend objects to the list
   while len(lines[line_index]	) <= 1 or lines[line_index].startswith(' ' * 8	):
    UpperCamelCase_					:						int =					lines[line_index]
    UpperCamelCase_					:						Optional[Any] =					_re_import.search(lowerCamelCase	)
    if single_line_import_search is not None:
     objects.extend(single_line_import_search.groups()[0].split(', '	)	)
    elif line.startswith(' ' * 12	):
     objects.append(line[12:-2]	)
    line_index += 1

   UpperCamelCase_					:						Dict =					objects
  else:
   line_index += 1

 return import_dict_objects, type_hint_objects





def    __lowercase     (						lowerCamelCase					:     str  ,  lowerCamelCase					:     Optional[Any]	):

 def find_duplicates(lowerCamelCase					:     Optional[int]	):
  return [k for k, v in collections.Counter(lowerCamelCase	).items() if v > 1]

 if list(import_dict_objects.keys()	) != list(type_hint_objects.keys()	):
  return ["Both sides of the init do not have the same backends!"]

 UpperCamelCase_					:						Optional[int] =					[]
 for key in import_dict_objects.keys():
  UpperCamelCase_					:						Optional[Any] =					find_duplicates(import_dict_objects[key]	)
  if duplicate_imports:
   errors.append(F"Duplicate _import_structure definitions for: {duplicate_imports}"	)
  UpperCamelCase_					:						Dict =					find_duplicates(type_hint_objects[key]	)
  if duplicate_type_hints:
   errors.append(F"Duplicate TYPE_CHECKING objects for: {duplicate_type_hints}"	)

  if sorted(set(import_dict_objects[key]	)	) != sorted(set(type_hint_objects[key]	)	):
   UpperCamelCase_					:						Any =					'base imports' if key == 'none' else F"{key} backend"
   errors.append(F"Differences for {name}:"	)
   for a in type_hint_objects[key]:
    if a not in import_dict_objects[key]:
     errors.append(F"  {a} in TYPE_HINT but not in _import_structure."	)
   for a in import_dict_objects[key]:
    if a not in type_hint_objects[key]:
     errors.append(F"  {a} in _import_structure but not in TYPE_HINT."	)
 return errors





def    __lowercase     (						):
 UpperCamelCase_					:						str =					[]
 for root, _, files in os.walk(lowerCamelCase	):
  if "__init__.py" in files:
   UpperCamelCase_					:						Union[str, Any] =					os.path.join(lowerCamelCase  ,  '__init__.py'	)
   UpperCamelCase_					:						Any =					parse_init(lowerCamelCase	)
   if objects is not None:
    UpperCamelCase_					:						List[Any] =					analyze_results(*lowerCamelCase	)
    if len(lowerCamelCase	) > 0:
     UpperCamelCase_					:						Any =					F"Problem in {fname}, both halves do not define the same objects.\n{errors[0]}"
     failures.append('\n'.join(lowerCamelCase	)	)
 if len(lowerCamelCase	) > 0:
  raise ValueError('\n\n'.join(lowerCamelCase	)	)





def    __lowercase     (						):
 UpperCamelCase_					:						str =					[]
 for path, directories, files in os.walk(lowerCamelCase	):
  for folder in directories:
   # Ignore private modules
   if folder.startswith('_'	):
    directories.remove(lowerCamelCase	)
    continue
   # Ignore leftovers from branches (empty folders apart from pycache)
   if len(list((Path(lowerCamelCase	) / folder).glob('*.py'	)	)	) == 0:
    continue
   UpperCamelCase_					:						Union[str, Any] =					str((Path(lowerCamelCase	) / folder).relative_to(lowerCamelCase	)	)
   UpperCamelCase_					:						Union[str, Any] =					short_path.replace(os.path.sep  ,  '.'	)
   submodules.append(lowerCamelCase	)
  for fname in files:
   if fname == "__init__.py":
    continue
   UpperCamelCase_					:						Optional[int] =					str((Path(lowerCamelCase	) / fname).relative_to(lowerCamelCase	)	)
   UpperCamelCase_					:						Union[str, Any] =					short_path.replace('.py'  ,  ''	).replace(os.path.sep  ,  '.'	)
   if len(submodule.split('.'	)	) == 1:
    submodules.append(lowerCamelCase	)
 return submodules


a_    					=       [
    'convert_pytorch_checkpoint_to_tf2',
    'modeling_flax_pytorch_utils',
    'models.esm.openfold_utils',
]





def    __lowercase     (						):
 # This is to make sure the transformers module imported is the one in the repo.
 from transformers.utils import direct_transformers_import

 UpperCamelCase_					:						List[str] =					direct_transformers_import(lowerCamelCase	)

 UpperCamelCase_					:						Tuple =					set(transformers._import_structure.keys()	)
 # This contains all the base keys of the _import_structure object defined in the init, but if the user is missing
 # some optional dependencies, they may not have all of them. Thus we read the init to read all additions and
 # (potentiall re-) add them.
 with open(os.path.join(lowerCamelCase  ,  '__init__.py'	)  ,  'r'	) as f:
  UpperCamelCase_					:						Optional[Any] =					f.read()
 import_structure_keys.update(set(re.findall(R'import_structure\[\"([^\"]*)\"\]'  ,  lowerCamelCase	)	)	)

 UpperCamelCase_					:						Tuple =					[
     module
     for module in get_transformers_submodules()
     if module not in IGNORE_SUBMODULES and module not in import_structure_keys
 ]

 if len(lowerCamelCase	) > 0:
  UpperCamelCase_					:						Dict =					'\n'.join(F"- {module}" for module in module_not_registered	)
  raise ValueError(
      'The following submodules are not properly registed in the main init of Transformers:\n'
      F"{list_of_modules}\n"
      'Make sure they appear somewhere in the keys of `_import_structure` with an empty list as value.'	)


if __name__ == "__main__":
   check_all_inits()
   check_submodules()




import os
import posixpath
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union

import numpy as np
import pyarrow as pa

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


a_    					=       datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
   import pyspark





@dataclass
class 		_lowercase						(					datasets.BuilderConfig       ):
 lowercase				     =   None





def    __lowercase     (						lowerCamelCase					:     "pyspark.sql.DataFrame"  ,  lowerCamelCase					:     List[int]  ,  ):
 import pyspark

 def generate_fn():
  UpperCamelCase_					:						Dict =					df.select('*'  ,  pyspark.sql.functions.spark_partition_id().alias('part_id'	)	)
  for partition_id in partition_order:
   UpperCamelCase_					:						Tuple =					df_with_partition_id.select('*'	).where(F"part_id = {partition_id}"	).drop('part_id'	)
   UpperCamelCase_					:						Union[str, Any] =					partition_df.collect()
   UpperCamelCase_					:						Any =					0
   for row in rows:
    yield F"{partition_id}_{row_id}", row.asDict()
    row_id += 1

 return generate_fn





class 		_lowercase						(					_BaseExamplesIterable       ):

 def __init__( self	:							Optional[int]   ,      snake_case	:							"pyspark.sql.DataFrame"   ,      snake_case	:							Tuple=None   ,      )   ->						Tuple:

  """simple docstring"""





  UpperCamelCase_					:						Dict =					df
  UpperCamelCase_					:						int =					partition_order or range(self.df.rdd.getNumPartitions()			)
  UpperCamelCase_					:						Optional[Any] =					_generate_iterable_examples(self.df   ,      self.partition_order			)

 def __iter__( self	:							Optional[int]			)   ->						Any:

  """simple docstring"""





  yield from self.generate_examples_fn()

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Any   ,      snake_case	:							np.random.Generator			)   ->						"SparkExamplesIterable":

  """simple docstring"""





  UpperCamelCase_					:						Optional[Any] =					list(range(self.df.rdd.getNumPartitions()			)			)
  generator.shuffle(snake_case			)
  return SparkExamplesIterable(self.df   ,      partition_order=snake_case			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							int   ,      snake_case	:							int   ,      snake_case	:							int			)   ->						"SparkExamplesIterable":

  """simple docstring"""





  UpperCamelCase_					:						Tuple =					self.split_shard_indices_by_worker(snake_case   ,      snake_case			)
  return SparkExamplesIterable(self.df   ,      partition_order=snake_case			)


 @property
 def 						SCREAMING_SNAKE_CASE__   ( self	:							int			)   ->						int:

  """simple docstring"""





  return len(self.partition_order			)





class 		_lowercase						(					datasets.DatasetBuilder       ):
 lowercase				     =   SparkConfig

 def __init__( self	:							List[Any]   ,      snake_case	:							"pyspark.sql.DataFrame"   ,      snake_case	:							str = None   ,      snake_case	:							str = None   ,      **snake_case	:							Optional[Any]   ,      )   ->						List[str]:

  """simple docstring"""





  import pyspark

  UpperCamelCase_					:						List[Any] =					pyspark.sql.SparkSession.builder.getOrCreate()
  UpperCamelCase_					:						str =					df
  UpperCamelCase_					:						Tuple =					working_dir

  super().__init__(
      cache_dir=snake_case   ,      config_name=str(self.df.semanticHash()			)   ,      **snake_case   ,      )

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Any			)   ->						Dict:

  """simple docstring"""





  def create_cache_and_write_probe(snake_case	:							str			):
   # makedirs with exist_ok will recursively create the directory. It will not throw an error if directories
   # already exist.
   os.makedirs(self._cache_dir   ,      exist_ok=snake_case			)
   UpperCamelCase_					:						Tuple =					os.path.join(self._cache_dir   ,      'fs_test' + uuid.uuida().hex			)
   # Opening the file in append mode will create a new file unless it already exists, in which case it will not
   # change the file contents.
   open(snake_case   ,      'a'			)
   return [probe_file]

  if self._spark.conf.get('spark.master'   ,      ''			).startswith('local'			):
   return

  # If the cluster is multi-node, make sure that the user provided a cache_dir and that it is on an NFS
  # accessible to the driver.
  # TODO: Stream batches to the driver using ArrowCollectSerializer instead of throwing an error.
  if self._cache_dir:
   UpperCamelCase_					:						Tuple =					(
       self._spark.sparkContext.parallelize(range(1			)   ,      1			).mapPartitions(snake_case			).collect()
   )
   if os.path.isfile(probe[0]			):
    return

  raise ValueError(
      'When using Dataset.from_spark on a multi-node cluster, the driver and all workers should be able to access cache_dir'			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							int			)   ->						Union[str, Any]:

  """simple docstring"""





  return datasets.DatasetInfo(features=self.config.features			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Optional[Any]   ,      snake_case	:							datasets.download.download_manager.DownloadManager			)   ->						Optional[int]:

  """simple docstring"""





  return [datasets.SplitGenerator(name=datasets.Split.TRAIN			)]

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Optional[Any]   ,      snake_case	:							Optional[int]			)   ->						List[Any]:

  """simple docstring"""





  import pyspark

  def get_arrow_batch_size(snake_case	:							Dict			):
   for batch in it:
    yield pa.RecordBatch.from_pydict({'batch_bytes': [batch.nbytes]}			)

  UpperCamelCase_					:						List[str] =					self.df.count()
  UpperCamelCase_					:						Union[str, Any] =					df_num_rows if df_num_rows <= 1_0_0 else 1_0_0
  # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
  UpperCamelCase_					:						str =					(
      self.df.limit(snake_case			)
      .repartition(1			)
      .mapInArrow(snake_case   ,      'batch_bytes: long'			)
      .agg(pyspark.sql.functions.sum('batch_bytes'			).alias('sample_bytes'			)			)
      .collect()[0]
      .sample_bytes
      / sample_num_rows
  )
  UpperCamelCase_					:						Optional[int] =					approx_bytes_per_row * df_num_rows
  if approx_total_size > max_shard_size:
   # Make sure there is at least one row per partition.
   UpperCamelCase_					:						Optional[Any] =					min(snake_case   ,      int(approx_total_size / max_shard_size			)			)
   UpperCamelCase_					:						int =					self.df.repartition(snake_case			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Union[str, Any]   ,      snake_case	:							str   ,      snake_case	:							str   ,      snake_case	:							int   ,      )   ->						Iterable[Tuple[int, bool, Union[int, tuple]]]:

  """simple docstring"""





  import pyspark

  UpperCamelCase_					:						List[Any] =					ParquetWriter if file_format == 'parquet' else ArrowWriter
  UpperCamelCase_					:						List[str] =					os.path.join(self._working_dir   ,      os.path.basename(snake_case			)			) if self._working_dir else fpath
  UpperCamelCase_					:						Union[str, Any] =					file_format == 'parquet'

  # Define these so that we don't reference self in write_arrow, which will result in a pickling error due to
  # pickling the SparkContext.
  UpperCamelCase_					:						Union[str, Any] =					self.config.features
  UpperCamelCase_					:						Any =					self._writer_batch_size
  UpperCamelCase_					:						Dict =					self._fs.storage_options

  def write_arrow(snake_case	:							List[str]			):
   # Within the same SparkContext, no two task attempts will share the same attempt ID.
   UpperCamelCase_					:						Any =					pyspark.TaskContext().taskAttemptId()
   UpperCamelCase_					:						str =					next(snake_case   ,      snake_case			)
   if first_batch is None:
    # Some partitions might not receive any data.
    return pa.RecordBatch.from_arrays(
        [[task_id], [0], [0]]   ,      names=['task_id', 'num_examples', 'num_bytes']   ,      )
   UpperCamelCase_					:						Any =					0
   UpperCamelCase_					:						Optional[Any] =					writer_class(
       features=snake_case   ,      path=working_fpath.replace('SSSSS'   ,      f"{shard_id:05d}"			).replace('TTTTT'   ,      f"{task_id:05d}"			)   ,      writer_batch_size=snake_case   ,      storage_options=snake_case   ,      embed_local_files=snake_case   ,      )
   UpperCamelCase_					:						str =					pa.Table.from_batches([first_batch]			)
   writer.write_table(snake_case			)
   for batch in it:
    if max_shard_size is not None and writer._num_bytes >= max_shard_size:
     UpperCamelCase_, UpperCamelCase_					:						str =					writer.finalize()
     writer.close()
     yield pa.RecordBatch.from_arrays(
         [[task_id], [num_examples], [num_bytes]]   ,      names=['task_id', 'num_examples', 'num_bytes']   ,      )
     shard_id += 1
     UpperCamelCase_					:						Union[str, Any] =					writer_class(
         features=writer._features   ,      path=working_fpath.replace('SSSSS'   ,      f"{shard_id:05d}"			).replace('TTTTT'   ,      f"{task_id:05d}"			)   ,      writer_batch_size=snake_case   ,      storage_options=snake_case   ,      embed_local_files=snake_case   ,      )
    UpperCamelCase_					:						Optional[Any] =					pa.Table.from_batches([batch]			)
    writer.write_table(snake_case			)

   if writer._num_bytes > 0:
    UpperCamelCase_, UpperCamelCase_					:						str =					writer.finalize()
    writer.close()
    yield pa.RecordBatch.from_arrays(
        [[task_id], [num_examples], [num_bytes]]   ,      names=['task_id', 'num_examples', 'num_bytes']   ,      )

   if working_fpath != fpath:
    for file in os.listdir(os.path.dirname(snake_case			)			):
     UpperCamelCase_					:						Dict =					os.path.join(os.path.dirname(snake_case			)   ,      os.path.basename(snake_case			)			)
     shutil.move(snake_case   ,      snake_case			)

  UpperCamelCase_					:						int =					(
      self.df.mapInArrow(snake_case   ,      'task_id: long, num_examples: long, num_bytes: long'			)
      .groupBy('task_id'			)
      .agg(
          pyspark.sql.functions.sum('num_examples'			).alias('total_num_examples'			)   ,      pyspark.sql.functions.sum('num_bytes'			).alias('total_num_bytes'			)   ,      pyspark.sql.functions.count('num_bytes'			).alias('num_shards'			)   ,      pyspark.sql.functions.collect_list('num_examples'			).alias('shard_lengths'			)   ,      )
      .collect()
  )
  for row in stats:
   yield row.task_id, (row.total_num_examples, row.total_num_bytes, row.num_shards, row.shard_lengths)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							int   ,      snake_case	:							"datasets.SplitGenerator"   ,      snake_case	:							str = "arrow"   ,      snake_case	:							Optional[Union[str, int]] = None   ,      snake_case	:							Optional[int] = None   ,      **snake_case	:							Any   ,      )   ->						int:

  """simple docstring"""





  self._validate_cache_dir()

  UpperCamelCase_					:						Optional[int] =					convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE			)
  self._repartition_df_if_needed(snake_case			)
  UpperCamelCase_					:						List[str] =					not is_remote_filesystem(self._fs			)
  UpperCamelCase_					:						List[Any] =					os.path.join if is_local else posixpath.join

  UpperCamelCase_					:						Optional[int] =					'-TTTTT-SSSSS-of-NNNNN'
  UpperCamelCase_					:						Dict =					f"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
  UpperCamelCase_					:						int =					path_join(self._output_dir   ,      snake_case			)

  UpperCamelCase_					:						int =					0
  UpperCamelCase_					:						Optional[int] =					0
  UpperCamelCase_					:						Union[str, Any] =					0
  UpperCamelCase_					:						Optional[Any] =					[]
  UpperCamelCase_					:						Any =					[]

  for task_id, content in self._prepare_split_single(snake_case   ,      snake_case   ,      snake_case			):
   (
       (
       UpperCamelCase_
   ), (
       UpperCamelCase_
   ), (
       UpperCamelCase_
   ), (
       UpperCamelCase_
   ), 
   )					:						Optional[Any] =					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(snake_case			)

  UpperCamelCase_					:						Optional[Any] =					total_num_examples
  UpperCamelCase_					:						Any =					total_num_bytes

  # should rename everything at the end
  logger.debug(f"Renaming {total_shards} shards."			)
  if total_shards > 1:
   UpperCamelCase_					:						List[Any] =					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.
   UpperCamelCase_					:						int =					self._fs

   # use the -SSSSS-of-NNNNN pattern
   def _rename_shard(
       snake_case	:							int   ,      snake_case	:							int   ,      snake_case	:							int   ,      ):
    rename(
        snake_case   ,      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}"			)   ,      )

   UpperCamelCase_					:						Any =					[]
   UpperCamelCase_					:						Optional[int] =					0
   for i in range(len(snake_case			)			):
    UpperCamelCase_, UpperCamelCase_					:						Union[str, Any] =					task_id_and_num_shards[i]
    for shard_id in range(snake_case			):
     args.append([task_id, shard_id, global_shard_id]			)
     global_shard_id += 1
   self._spark.sparkContext.parallelize(snake_case   ,      len(snake_case			)			).map(lambda snake_case			: _rename_shard(*snake_case			)			).collect()
  else:
   # don't use any pattern
   UpperCamelCase_					:						Tuple =					0
   UpperCamelCase_					:						Optional[Any] =					task_id_and_num_shards[0][0]
   self._rename(
       fpath.replace('SSSSS'   ,      f"{shard_id:05d}"			).replace('TTTTT'   ,      f"{task_id:05d}"			)   ,      fpath.replace(snake_case   ,      ''			)   ,      )


 def 						SCREAMING_SNAKE_CASE__   ( self	:							Tuple   ,      snake_case	:							"datasets.SplitGenerator"   ,      )   ->						SparkExamplesIterable:

  """simple docstring"""





  return SparkExamplesIterable(self.df			)



