


'''simple docstring'''


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


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

if TYPE_CHECKING:
     import pyspark



@dataclass
class 	__lowerCAmelCase		(							datasets.BuilderConfig    ):


  '''simple docstring'''







  a_  					=							None







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

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

      return generate_fn



class 	__lowerCAmelCase		(							_BaseExamplesIterable    ):


  '''simple docstring'''







  def __init__(							self :       Optional[Any]	,_a :       "pyspark.sql.DataFrame"	,_a :       Union[str, Any]=None	,):

        '''simple docstring'''




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






  def __iter__(							self :       Optional[Any] ):

        '''simple docstring'''




        yield from self.generate_examples_fn()






  def        _a							(							self :       Optional[int]	,_a :       np.random.Generator ):

        '''simple docstring'''




        A_						:  Union[str, Any]          =				list(range(self.df.rdd.getNumPartitions() ) )
        generator.shuffle(_a )
        return SparkExamplesIterable(self.df	,partition_order=_a )






  def        _a							(							self :       Optional[Any]	,_a :       int	,_a :       int ):

        '''simple docstring'''




        A_						:  Optional[Any]          =				self.split_shard_indices_by_worker(_a	,_a )
        return SparkExamplesIterable(self.df	,partition_order=_a )






  @property
  def        _a							(							self :       Tuple ):

        '''simple docstring'''




        return len(self.partition_order )







class 	__lowerCAmelCase		(							datasets.DatasetBuilder    ):


  '''simple docstring'''







  a_  					=							SparkConfig






  def __init__(							self :       Any	,_a :       "pyspark.sql.DataFrame"	,_a :       str = None	,_a :       str = None	,**_a :       Optional[int]	,):

        '''simple docstring'''




        import pyspark

        A_						:  int          =				pyspark.sql.SparkSession.builder.getOrCreate()
        A_						:  Optional[int]          =				df
        A_						:  List[Any]          =				working_dir

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






  def        _a							(							self :       int ):

        '''simple docstring'''




        def create_cache_and_write_probe(_a :       Any ):
              # 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=_a )
              A_						:  Optional[Any]          =				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(_a	,"""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:
              A_						:  Any          =				(
                  self._spark.sparkContext.parallelize(range(1 )	,1 ).mapPartitions(_a ).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        _a							(							self :       Dict ):

        '''simple docstring'''




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






  def        _a							(							self :       List[Any]	,_a :       datasets.download.download_manager.DownloadManager ):

        '''simple docstring'''




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






  def        _a							(							self :       List[Any]	,_a :       List[Any] ):

        '''simple docstring'''




        import pyspark

        def get_arrow_batch_size(_a :       int ):
              for batch in it:
                    yield pa.RecordBatch.from_pydict({"""batch_bytes""": [batch.nbytes]} )

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






  def        _a							(							self :       int	,_a :       str	,_a :       str	,_a :       int	,):

        '''simple docstring'''




        import pyspark

        A_						:  List[Any]          =				ParquetWriter if file_format == """parquet""" else ArrowWriter
        A_						:  Optional[Any]          =				os.path.join(self._working_dir	,os.path.basename(_a ) ) if self._working_dir else fpath
        A_						:  int          =				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.
        A_						:  Dict          =				self.config.features
        A_						:  Optional[Any]          =				self._writer_batch_size
        A_						:  Any          =				self._fs.storage_options

        def write_arrow(_a :       Union[str, Any] ):
              # Within the same SparkContext, no two task attempts will share the same attempt ID.
              A_						:  str          =				pyspark.TaskContext().taskAttemptId()
              A_						:  Tuple          =				next(_a	,_a )
              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"""]	,)
              A_						:  Optional[int]          =				0
              A_						:  Any          =				writer_class(
                  features=_a	,path=working_fpath.replace("""SSSSS"""	,f'{shard_id:05d}' ).replace("""TTTTT"""	,f'{task_id:05d}' )	,writer_batch_size=_a	,storage_options=_a	,embed_local_files=_a	,)
              A_						:  Union[str, Any]          =				pa.Table.from_batches([first_batch] )
              writer.write_table(_a )
              for batch in it:
                    if max_shard_size is not None and writer._num_bytes >= max_shard_size:
                          A_   ,					A_						:  Optional[Any]          =				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
                          A_						:  Dict          =				writer_class(
                              features=writer._features	,path=working_fpath.replace("""SSSSS"""	,f'{shard_id:05d}' ).replace("""TTTTT"""	,f'{task_id:05d}' )	,writer_batch_size=_a	,storage_options=_a	,embed_local_files=_a	,)
                    A_						:  List[str]          =				pa.Table.from_batches([batch] )
                    writer.write_table(_a )

              if writer._num_bytes > 0:
                    A_   ,					A_						:  Optional[int]          =				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(_a ) ):
                          A_						:  List[str]          =				os.path.join(os.path.dirname(_a )	,os.path.basename(_a ) )
                          shutil.move(_a	,_a )

        A_						:  Dict          =				(
            self.df.mapInArrow(_a	,"""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        _a							(							self :       Union[str, Any]	,_a :       "datasets.SplitGenerator"	,_a :       str = "arrow"	,_a :       Optional[Union[str, int]] = None	,_a :       Optional[int] = None	,**_a :       int	,):

        '''simple docstring'''




        self._validate_cache_dir()

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

        A_						:  List[str]          =				"""-TTTTT-SSSSS-of-NNNNN"""
        A_						:  List[str]          =				f'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
        A_						:  Tuple          =				path_join(self._output_dir	,_a )

        A_						:  Dict          =				0
        A_						:  Optional[Any]          =				0
        A_						:  Optional[Any]          =				0
        A_						:  Any          =				[]
        A_						:  str          =				[]

        for task_id, content in self._prepare_split_single(_a	,_a	,_a ):
              (
                  (
                  A_
              )   ,					(
                  A_
              )   ,					(
                  A_
              )   ,					(
                  A_
              )   ,					
              )						:  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(_a )

        A_						:  Any          =				total_num_examples
        A_						:  Any          =				total_num_bytes

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

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

              A_						:  int          =				[]
              A_						:  Union[str, Any]          =				0
              for i in range(len(_a ) ):
                    A_   ,					A_						:  Any          =				task_id_and_num_shards[i]
                    for shard_id in range(_a ):
                          args.append([task_id, shard_id, global_shard_id] )
                          global_shard_id += 1
              self._spark.sparkContext.parallelize(_a	,len(_a ) ).map(lambda _a : _rename_shard(*_a ) ).collect()
        else:
              # don't use any pattern
              A_						:  str          =				0
              A_						:  Union[str, 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(_a	,"""""" )	,)






  def        _a							(							self :       Optional[int]	,_a :       "datasets.SplitGenerator"	,):

        '''simple docstring'''




        return SparkExamplesIterable(self.df )




'''simple docstring'''


from ..utils import DummyObject, requires_backends



class 	__lowerCAmelCase		(							metaclass=__SCREAMING_SNAKE_CASE    ):


  '''simple docstring'''







  a_  					=							["""torch""", """torchsde"""]






  def __init__(							self :       Any	,*_a :       Union[str, Any]	,**_a :       Optional[int] ):

        '''simple docstring'''




        requires_backends(self	,["""torch""", """torchsde"""] )






  @classmethod
  def        _a							(							cls :       Optional[int]	,*_a :       List[Any]	,**_a :       Any ):

        '''simple docstring'''




        requires_backends(cls	,["""torch""", """torchsde"""] )






  @classmethod
  def        _a							(							cls :       List[Any]	,*_a :       Tuple	,**_a :       Union[str, Any] ):

        '''simple docstring'''




        requires_backends(cls	,["""torch""", """torchsde"""] )
