




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


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

if TYPE_CHECKING:
 import pyspark
@dataclass
class 		_A   (    datasets.BuilderConfig ):
  _UpperCamelCase  :     Optional[Any]					=     None






def    snake_case(							__magic_name__					,						__magic_name__					,						)					->				Optional[int]:





     '''simple docstring'''


     import pyspark

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

     return generate_fn
class 		_A   (    _BaseExamplesIterable ):






  def __init__(    self						:      Dict ,		_A						:      "pyspark.sql.DataFrame" ,		_A						:      str=None ,		)						->       List[str]:






       """simple docstring"""







       lowercase   :  List[str]        =					df
       lowercase   :  Any        =					partition_order or range(self.df.rdd.getNumPartitions() )
       lowercase   :  List[str]        =					_generate_iterable_examples(self.df ,		self.partition_order )






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






       """simple docstring"""







       yield from self.generate_examples_fn()






  def    __a     (    self						:      Optional[int] ,		_A						:      np.random.Generator )						->       "SparkExamplesIterable":






       """simple docstring"""







       lowercase   :  str        =					list(range(self.df.rdd.getNumPartitions() ) )
       generator.shuffle(_A )
       return SparkExamplesIterable(self.df ,		partition_order=_A )






  def    __a     (    self						:      List[Any] ,		_A						:      int ,		_A						:      int )						->       "SparkExamplesIterable":






       """simple docstring"""







       lowercase   :  Tuple        =					self.split_shard_indices_by_worker(_A ,		_A )
       return SparkExamplesIterable(self.df ,		partition_order=_A )






  @property
  def    __a     (    self						:      Optional[int] )						->       int:






       """simple docstring"""







       return len(self.partition_order )


class 		_A   (    datasets.DatasetBuilder ):
  _UpperCamelCase  :     List[str]					=     SparkConfig






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






       """simple docstring"""







       import pyspark

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

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






  def    __a     (    self						:      str )						->       Tuple:






       """simple docstring"""







       def create_cache_and_write_probe(_A						:      Optional[int] ):
            # 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 )
            lowercase   :  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(_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:
            lowercase   :  Optional[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						:      Any )						->       Dict:






       """simple docstring"""







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






  def    __a     (    self						:      Optional[Any] ,		_A						:      datasets.download.download_manager.DownloadManager )						->       Any:






       """simple docstring"""







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






  def    __a     (    self						:      int ,		_A						:      Union[str, Any] )						->       Optional[Any]:






       """simple docstring"""







       import pyspark

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

       lowercase   :  Tuple        =					self.df.count()
       lowercase   :  int        =					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.
       lowercase   :  str        =					(
           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
       )
       lowercase   :  Dict        =					approx_bytes_per_row * df_num_rows
       if approx_total_size > max_shard_size:
            # Make sure there is at least one row per partition.
            lowercase   :  Tuple        =					min(_A ,		int(approx_total_size / max_shard_size ) )
            lowercase   :  Tuple        =					self.df.repartition(_A )






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






       """simple docstring"""







       import pyspark

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

       def write_arrow(_A						:      List[Any] ):
            # Within the same SparkContext, no two task attempts will share the same attempt ID.
            lowercase   :  List[str]        =					pyspark.TaskContext().taskAttemptId()
            lowercase   :  List[str]        =					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'''] ,		)
            lowercase   :  Tuple        =					0
            lowercase   :  Union[str, 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 ,		)
            lowercase   :  Dict        =					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:
                      lowercase   :  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
                      lowercase   :  Any        =					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 ,		)
                 lowercase   :  str        =					pa.Table.from_batches([batch] )
                 writer.write_table(_A )

            if writer._num_bytes > 0:
                 lowercase   :  List[Any]        =					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 ) ):
                      lowercase   :  Union[str, Any]        =					os.path.join(os.path.dirname(_A ) ,		os.path.basename(_A ) )
                      shutil.move(_A ,		_A )

       lowercase   :  Optional[int]        =					(
           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						:      Optional[Any] ,		_A						:      "datasets.SplitGenerator" ,		_A						:      str = "arrow" ,		_A						:      Optional[Union[str, int]] = None ,		_A						:      Optional[int] = None ,		**_A						:      str ,		)						->       str:






       """simple docstring"""







       self._validate_cache_dir()

       lowercase   :  int        =					convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE )
       self._repartition_df_if_needed(_A )
       lowercase   :  Dict        =					not is_remote_filesystem(self._fs )
       lowercase   :  Optional[Any]        =					os.path.join if is_local else posixpath.join

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

       lowercase   :  int        =					0
       lowercase   :  str        =					0
       lowercase   :  List[Any]        =					0
       lowercase   :  Union[str, Any]        =					[]
       lowercase   :  List[Any]        =					[]

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

       lowercase   :  List[str]        =					total_num_examples
       lowercase   :  str        =					total_num_bytes

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

            lowercase   :  Tuple        =					[]
            lowercase   :  str        =					0
            for i in range(len(_A ) ):
                 lowercase   :  Optional[int]        =					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
            lowercase   :  Optional[int]        =					0
            lowercase   :  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(_A ,		'''''' ) ,		)






  def    __a     (    self						:      int ,		_A						:      "datasets.SplitGenerator" ,		)						->       SparkExamplesIterable:






       """simple docstring"""







       return SparkExamplesIterable(self.df )


from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


a_      =     {'configuration_van': ['VAN_PRETRAINED_CONFIG_ARCHIVE_MAP', 'VanConfig']}


try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			a_      =     [
			    'VAN_PRETRAINED_MODEL_ARCHIVE_LIST',
			    'VanForImageClassification',
			    'VanModel',
			    'VanPreTrainedModel',
			]

if TYPE_CHECKING:
			from .configuration_van import VAN_PRETRAINED_CONFIG_ARCHIVE_MAP, VanConfig

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_van import (
						    VAN_PRETRAINED_MODEL_ARCHIVE_LIST,
						    VanForImageClassification,
						    VanModel,
						    VanPreTrainedModel,
						)

else:
			import sys

			a_      =     _LazyModule(__name__, globals()['__file__'], _import_structure)