


'''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


a_					:    Optional[int]          =    datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
 import pyspark




@dataclass
class  a     (     datasets.BuilderConfig       ):
 _lowerCAmelCase 					=	None






def 	_A				(lowerCAmelCase__  :Tuple				,						lowerCAmelCase__  :List[str]				,						)   ->  Any:




     '''simple docstring'''



     import pyspark

     def generate_fn():
          _a			      =							df.select('*'				,						pyspark.sql.functions.spark_partition_id().alias('part_id'		)		)
          for partition_id in partition_order:
               _a			      =							df_with_partition_id.select('*'		).where(f'part_id = {partition_id}'		).drop('part_id'		)
               _a			      =							partition_df.collect()
               _a			      =							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      ,				__magic_name__      ,				__magic_name__=None      ,				)					->							Dict:
      _a			      =							df
      _a			      =							partition_order or range(self.df.rdd.getNumPartitions()						)
      _a			      =							_generate_iterable_examples(self.df      ,				self.partition_order						)
 def __iter__( self						)					->							List[str]:
      yield from self.generate_examples_fn()
 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							Optional[int]:
      _a			      =							list(range(self.df.rdd.getNumPartitions()						)						)
      generator.shuffle(__lowerCAmelCase						)
      return SparkExamplesIterable(self.df      ,				partition_order=__lowerCAmelCase						)
 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				__magic_name__						)					->							Union[str, Any]:
      _a			      =							self.split_shard_indices_by_worker(__lowerCAmelCase      ,				__lowerCAmelCase						)
      return SparkExamplesIterable(self.df      ,				partition_order=__lowerCAmelCase						)




 @property
 def 							__UpperCAmelCase		( self						)					->							List[str]:
      return len(self.partition_order						)





class  a     (     datasets.DatasetBuilder       ):
 _lowerCAmelCase 					=	SparkConfig
 def __init__( self      ,				__magic_name__      ,				__magic_name__ = None      ,				__magic_name__ = None      ,				**__magic_name__      ,				)					->							int:
      import pyspark

      _a			      =							pyspark.sql.SparkSession.builder.getOrCreate()
      _a			      =							df
      _a			      =							working_dir

      super().__init__(
          cache_dir=__lowerCAmelCase      ,				config_name=str(self.df.semanticHash()						)      ,				**__lowerCAmelCase      ,				)
 def 							__UpperCAmelCase		( self						)					->							str:







      def create_cache_and_write_probe(__magic_name__						):
           # 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=__lowerCAmelCase						)
           _a			      =							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(__lowerCAmelCase      ,				'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			      =							(
               self._spark.sparkContext.parallelize(range(1						)      ,				1						).mapPartitions(__lowerCAmelCase						).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 							__UpperCAmelCase		( self						)					->							str:
      return datasets.DatasetInfo(features=self.config.features						)
 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							Any:
      return [datasets.SplitGenerator(name=datasets.Split.TRAIN						)]
 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							List[Any]:
      import pyspark

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

      _a			      =							self.df.count()
      _a			      =							df_num_rows if df_num_rows <= 1_00 else 1_00
      # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
      _a			      =							(
          self.df.limit(__lowerCAmelCase						)
          .repartition(1						)
          .mapInArrow(__lowerCAmelCase      ,				'batch_bytes: long'						)
          .agg(pyspark.sql.functions.sum('batch_bytes'						).alias('sample_bytes'						)						)
          .collect()[0]
          .sample_bytes
          / sample_num_rows
      )
      _a			      =							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			      =							min(__lowerCAmelCase      ,				int(approx_total_size / max_shard_size						)						)
           _a			      =							self.df.repartition(__lowerCAmelCase						)
 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				__magic_name__      ,				__magic_name__      ,				)					->							str:
      import pyspark

      _a			      =							ParquetWriter if file_format == 'parquet' else ArrowWriter
      _a			      =							os.path.join(self._working_dir      ,				os.path.basename(__lowerCAmelCase						)						) if self._working_dir else fpath
      _a			      =							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			      =							self.config.features
      _a			      =							self._writer_batch_size
      _a			      =							self._fs.storage_options

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

           if writer._num_bytes > 0:
                _a      ,   _a			      =							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(__lowerCAmelCase						)						):
                     _a			      =							os.path.join(os.path.dirname(__lowerCAmelCase						)      ,				os.path.basename(__lowerCAmelCase						)						)
                     shutil.move(__lowerCAmelCase      ,				__lowerCAmelCase						)

      _a			      =							(
          self.df.mapInArrow(__lowerCAmelCase      ,				'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 							__UpperCAmelCase		( self      ,				__magic_name__      ,				__magic_name__ = "arrow"      ,				__magic_name__ = None      ,				__magic_name__ = None      ,				**__magic_name__      ,				)					->							Dict:
      self._validate_cache_dir()

      _a			      =							convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE						)
      self._repartition_df_if_needed(__lowerCAmelCase						)
      _a			      =							not is_remote_filesystem(self._fs						)
      _a			      =							os.path.join if is_local else posixpath.join

      _a			      =							'-TTTTT-SSSSS-of-NNNNN'
      _a			      =							f'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
      _a			      =							path_join(self._output_dir      ,				__lowerCAmelCase						)

      _a			      =							0
      _a			      =							0
      _a			      =							0
      _a			      =							[]
      _a			      =							[]

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

      _a			      =							total_num_examples
      _a			      =							total_num_bytes

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

           # use the -SSSSS-of-NNNNN pattern
           def _rename_shard(
               __magic_name__      ,				__magic_name__      ,				__magic_name__      ,				):
                rename(
                    __lowerCAmelCase      ,				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			      =							[]
           _a			      =							0
           for i in range(len(__lowerCAmelCase						)						):
                _a      ,   _a			      =							task_id_and_num_shards[i]
                for shard_id in range(__lowerCAmelCase						):
                     args.append([task_id, shard_id, global_shard_id]						)
                     global_shard_id += 1
           self._spark.sparkContext.parallelize(__lowerCAmelCase      ,				len(__lowerCAmelCase						)						).map(lambda __magic_name__						: _rename_shard(*__lowerCAmelCase						)						).collect()
      else:
           # don't use any pattern
           _a			      =							0
           _a			      =							task_id_and_num_shards[0][0]
           self._rename(
               fpath.replace('SSSSS'      ,				f'{shard_id:05d}'						).replace('TTTTT'      ,				f'{task_id:05d}'						)      ,				fpath.replace(__lowerCAmelCase      ,				''						)      ,				)




 def 							__UpperCAmelCase		( self      ,				__magic_name__      ,				)					->							int:
      return SparkExamplesIterable(self.df						)





'''simple docstring'''






class  a     :
 def __init__( self      ,				__magic_name__						)					->							Optional[int]:
      _a			      =							n
      _a			      =							[None] * self.n
      _a			      =							0  # index of the first element
      _a			      =							0
      _a			      =							0
 def __len__( self						)					->							int:
      return self.size
 def 							__UpperCAmelCase		( self						)					->							bool:
      return self.size == 0
 def 							__UpperCAmelCase		( self						)					->							Optional[Any]:
      return False if self.is_empty() else self.array[self.front]
 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							Optional[Any]:
      if self.size >= self.n:
           raise Exception('QUEUE IS FULL'						)

      _a			      =							data
      _a			      =							(self.rear + 1) % self.n
      self.size += 1
      return self




 def 							__UpperCAmelCase		( self						)					->							Any:
      if self.size == 0:
           raise Exception('UNDERFLOW'						)

      _a			      =							self.array[self.front]
      _a			      =							None
      _a			      =							(self.front + 1) % self.n
      self.size -= 1
      return temp

