




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     __a(	datasets.BuilderConfig  ):






    """simple docstring"""





    lowerCAmelCase         = None







def lowerCamelCase__  (     _lowercase					,		_lowercase					,		):

     '''simple docstring'''


     import pyspark

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

     return generate_fn

class     __a(	_BaseExamplesIterable  ):






    """simple docstring"""





    def __init__( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE=None  ,)					->						Dict:
         UpperCAmelCase_     :				str  =     df
         UpperCAmelCase_     :				Optional[Any]  =     partition_order or range(self.df.rdd.getNumPartitions()    )
         UpperCAmelCase_     :				str  =     _generate_iterable_examples(self.df  ,self.partition_order    )





    def __iter__( self    )					->						Dict:
         yield from self.generate_examples_fn()





    def 				a__ ( self  ,_SCREAMING_SNAKE_CASE    )					->						"SparkExamplesIterable":
         UpperCAmelCase_     :				Union[str, Any]  =     list(range(self.df.rdd.getNumPartitions()    )    )
         generator.shuffle(_SCREAMING_SNAKE_CASE    )
         return SparkExamplesIterable(self.df  ,partition_order=_SCREAMING_SNAKE_CASE    )





    def 				a__ ( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE    )					->						"SparkExamplesIterable":
         UpperCAmelCase_     :				Dict  =     self.split_shard_indices_by_worker(_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE    )
         return SparkExamplesIterable(self.df  ,partition_order=_SCREAMING_SNAKE_CASE    )





    @property
    def 				a__ ( self    )					->						int:
         return len(self.partition_order    )


class     __a(	datasets.DatasetBuilder  ):






    """simple docstring"""





    lowerCAmelCase         = SparkConfig





    def __init__( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE = None  ,_SCREAMING_SNAKE_CASE = None  ,**_SCREAMING_SNAKE_CASE  ,)					->						Optional[Any]:
         import pyspark

         UpperCAmelCase_     :				List[Any]  =     pyspark.sql.SparkSession.builder.getOrCreate()
         UpperCAmelCase_     :				Union[str, Any]  =     df
         UpperCAmelCase_     :				Optional[Any]  =     working_dir

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





    def 				a__ ( self    )					->						int:
         # Returns the path of the created file.
         def create_cache_and_write_probe(_SCREAMING_SNAKE_CASE    ):
              # 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=_SCREAMING_SNAKE_CASE    )
              UpperCAmelCase_     :				List[str]  =     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(_SCREAMING_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(_SCREAMING_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 				a__ ( self    )					->						Tuple:
         return datasets.DatasetInfo(features=self.config.features    )





    def 				a__ ( self  ,_SCREAMING_SNAKE_CASE    )					->						Any:
         return [datasets.SplitGenerator(name=datasets.Split.TRAIN    )]





    def 				a__ ( self  ,_SCREAMING_SNAKE_CASE    )					->						Union[str, Any]:
         import pyspark

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

         UpperCAmelCase_     :				Union[str, Any]  =     self.df.count()
         UpperCAmelCase_     :				Tuple  =     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.
         UpperCAmelCase_     :				Tuple  =     (
             self.df.limit(_SCREAMING_SNAKE_CASE    )
             .repartition(1    )
             .mapInArrow(_SCREAMING_SNAKE_CASE  ,'''batch_bytes: long'''    )
             .agg(pyspark.sql.functions.sum('''batch_bytes'''    ).alias('''sample_bytes'''    )    )
             .collect()[0]
             .sample_bytes
             / sample_num_rows
         )
         UpperCAmelCase_     :				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.
              UpperCAmelCase_     :				List[Any]  =     min(_SCREAMING_SNAKE_CASE  ,int(approx_total_size / max_shard_size    )    )
              UpperCAmelCase_     :				List[str]  =     self.df.repartition(_SCREAMING_SNAKE_CASE    )





    def 				a__ ( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,)					->						Iterable[Tuple[int, bool, Union[int, tuple]]]:
         import pyspark

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

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

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

         UpperCAmelCase_     :				int  =     (
             self.df.mapInArrow(_SCREAMING_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 				a__ ( self  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE = "arrow"  ,_SCREAMING_SNAKE_CASE = None  ,_SCREAMING_SNAKE_CASE = None  ,**_SCREAMING_SNAKE_CASE  ,)					->						int:
         self._validate_cache_dir()

         UpperCAmelCase_     :				str  =     convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE    )
         self._repartition_df_if_needed(_SCREAMING_SNAKE_CASE    )
         UpperCAmelCase_     :				str  =     not is_remote_filesystem(self._fs    )
         UpperCAmelCase_     :				Dict  =     os.path.join if is_local else posixpath.join

         UpperCAmelCase_     :				str  =     '''-TTTTT-SSSSS-of-NNNNN'''
         UpperCAmelCase_     :				List[Any]  =     f'''{self.name}-{split_generator.name}{SUFFIX}.{file_format}'''
         UpperCAmelCase_     :				Optional[Any]  =     path_join(self._output_dir  ,_SCREAMING_SNAKE_CASE    )

         UpperCAmelCase_     :				Optional[int]  =     0
         UpperCAmelCase_     :				str  =     0
         UpperCAmelCase_     :				Union[str, Any]  =     0
         UpperCAmelCase_     :				List[Any]  =     []
         UpperCAmelCase_     :				Dict  =     []

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

         UpperCAmelCase_     :				Tuple  =     total_num_examples
         UpperCAmelCase_     :				str  =     total_num_bytes

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

              # use the -SSSSS-of-NNNNN pattern
              def _rename_shard(
                  _SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,_SCREAMING_SNAKE_CASE  ,):
                   rename(
                       _SCREAMING_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_     :				str  =     []
              UpperCAmelCase_     :				Any  =     0
              for i in range(len(_SCREAMING_SNAKE_CASE    )    ):
                   UpperCAmelCase_,       UpperCAmelCase_     :				Tuple  =     task_id_and_num_shards[i]
                   for shard_id in range(_SCREAMING_SNAKE_CASE    ):
                        args.append([task_id, shard_id, global_shard_id]    )
                        global_shard_id += 1
              self._spark.sparkContext.parallelize(_SCREAMING_SNAKE_CASE  ,len(_SCREAMING_SNAKE_CASE    )    ).map(lambda _SCREAMING_SNAKE_CASE    : _rename_shard(*_SCREAMING_SNAKE_CASE    )    ).collect()
         else:
              # don't use any pattern
              UpperCAmelCase_     :				Tuple  =     0
              UpperCAmelCase_     :				List[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(_SCREAMING_SNAKE_CASE  ,''''''    )  ,)





    def 				a__ ( self  ,_SCREAMING_SNAKE_CASE  ,)					->						SparkExamplesIterable:
         return SparkExamplesIterable(self.df    )





def lowerCamelCase__  (     _lowercase					,		_lowercase ):

     '''simple docstring'''


     while a != 0:
          UpperCAmelCase_,       UpperCAmelCase_     :				Optional[int]  =     b % a, a
     return b







def lowerCamelCase__  (     _lowercase					,		_lowercase ):

     '''simple docstring'''


     if gcd(_lowercase					,		_lowercase ) != 1:
          UpperCAmelCase_     :				int  =     f'''mod inverse of {a!r} and {m!r} does not exist'''
          raise ValueError(_lowercase )
     UpperCAmelCase_,       UpperCAmelCase_,       UpperCAmelCase_     :				Union[str, Any]  =     1, 0, a
     UpperCAmelCase_,       UpperCAmelCase_,       UpperCAmelCase_     :				Dict  =     0, 1, m
     while va != 0:
          UpperCAmelCase_     :				List[Any]  =     ua // va
          UpperCAmelCase_,       UpperCAmelCase_,       UpperCAmelCase_,       UpperCAmelCase_,       UpperCAmelCase_,       UpperCAmelCase_     :				Any  =     (ua - q * va), (ua - q * va), (ua - q * va), va, va, va
     return ua % m