






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


SCREAMING_SNAKE_CASE__		:							Union[str, Any] 				=				datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
      import pyspark







@dataclass
class 				UpperCAmelCase_  (		datasets.BuilderConfig		):
     __lowerCamelCase      				=  None





def  _lowerCamelCase						(						__lowerCamelCase ,							__lowerCamelCase ,							) ->			str:

     '''simple docstring'''
     import pyspark

     def generate_fn():
          UpperCAmelCase__      :  Any				      =			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__      :  Dict				      =			partition_df.collect()
               UpperCAmelCase__      :  int				      =			0
               for row in rows:
                    yield F"{partition_id}_{row_id}", row.asDict()
                    row_id += 1

     return generate_fn







class 				UpperCAmelCase_  (		_BaseExamplesIterable		):




     def __init__(				self	,						_lowerCAmelCase	,						_lowerCAmelCase=None	,						):
          UpperCAmelCase__      :  str				      =			df
          UpperCAmelCase__      :  Any				      =			partition_order or range(self.df.rdd.getNumPartitions()		)
          UpperCAmelCase__      :  Dict				      =			_generate_iterable_examples(self.df	,						self.partition_order		)




     def __iter__(				self		):
          yield from self.generate_examples_fn()




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase		):
          UpperCAmelCase__      :  Union[str, Any]				      =			list(range(self.df.rdd.getNumPartitions()		)		)
          generator.shuffle(_lowerCAmelCase		)
          return SparkExamplesIterable(self.df	,						partition_order=_lowerCAmelCase		)




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase	,						_lowerCAmelCase		):
          UpperCAmelCase__      :  Any				      =			self.split_shard_indices_by_worker(_lowerCAmelCase	,						_lowerCAmelCase		)
          return SparkExamplesIterable(self.df	,						partition_order=_lowerCAmelCase		)




     @property
     def   __UpperCAmelCase  (				self		):
          return len(self.partition_order		)







class 				UpperCAmelCase_  (		datasets.DatasetBuilder		):
     __lowerCamelCase      				=  SparkConfig




     def __init__(				self	,						_lowerCAmelCase	,						_lowerCAmelCase = None	,						_lowerCAmelCase = None	,						**_lowerCAmelCase	,						):
          import pyspark

          UpperCAmelCase__      :  List[str]				      =			pyspark.sql.SparkSession.builder.getOrCreate()
          UpperCAmelCase__      :  Optional[int]				      =			df
          UpperCAmelCase__      :  Tuple				      =			working_dir

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




     def   __UpperCAmelCase  (				self		):
          # Returns the path of the created file.
          def create_cache_and_write_probe(_lowerCAmelCase		):
               # 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		)
               UpperCAmelCase__      :  Dict				      =			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:
               UpperCAmelCase__      :  Optional[Any]				      =			(
                   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		):
          return datasets.DatasetInfo(features=self.config.features		)




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase		):
          return [datasets.SplitGenerator(name=datasets.Split.TRAIN		)]




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase		):
          import pyspark

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

          UpperCAmelCase__      :  List[Any]				      =			self.df.count()
          UpperCAmelCase__      :  Dict				      =			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__      :  Optional[Any]				      =			(
              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
          )
          UpperCAmelCase__      :  str				      =			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__      :  Dict				      =			min(_lowerCAmelCase	,						int(approx_total_size / max_shard_size		)		)
               UpperCAmelCase__      :  Optional[Any]				      =			self.df.repartition(_lowerCAmelCase		)




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase	,						_lowerCAmelCase	,						_lowerCAmelCase	,						):
          import pyspark

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

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

               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(_lowerCAmelCase		)		):
                         UpperCAmelCase__      :  Any				      =			os.path.join(os.path.dirname(_lowerCAmelCase		)	,						os.path.basename(_lowerCAmelCase		)		)
                         shutil.move(_lowerCAmelCase	,						_lowerCAmelCase		)

          UpperCAmelCase__      :  int				      =			(
              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	,						_lowerCAmelCase	,						_lowerCAmelCase = "arrow"	,						_lowerCAmelCase = None	,						_lowerCAmelCase = None	,						**_lowerCAmelCase	,						):
          self._validate_cache_dir()

          UpperCAmelCase__      :  Optional[Any]				      =			convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE		)
          self._repartition_df_if_needed(_lowerCAmelCase		)
          UpperCAmelCase__      :  Optional[int]				      =			not is_remote_filesystem(self._fs		)
          UpperCAmelCase__      :  Union[str, Any]				      =			os.path.join if is_local else posixpath.join

          UpperCAmelCase__      :  List[Any]				      =			"""-TTTTT-SSSSS-of-NNNNN"""
          UpperCAmelCase__      :  Optional[int]				      =			f"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
          UpperCAmelCase__      :  Optional[int]				      =			path_join(self._output_dir	,						_lowerCAmelCase		)

          UpperCAmelCase__      :  str				      =			0
          UpperCAmelCase__      :  Union[str, Any]				      =			0
          UpperCAmelCase__      :  Union[str, Any]				      =			0
          UpperCAmelCase__      :  Tuple				      =			[]
          UpperCAmelCase__      :  Tuple				      =			[]

          for task_id, content in self._prepare_split_single(_lowerCAmelCase	,						_lowerCAmelCase	,						_lowerCAmelCase		):
               (
                   (
                   UpperCAmelCase__
               )      ,		(
                   UpperCAmelCase__
               )      ,		(
                   UpperCAmelCase__
               )      ,		(
                   UpperCAmelCase__
               )      ,		
               )      :  Union[str, 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(_lowerCAmelCase		)

          UpperCAmelCase__      :  Any				      =			total_num_examples
          UpperCAmelCase__      :  Optional[int]				      =			total_num_bytes

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

               UpperCAmelCase__      :  Optional[int]				      =			[]
               UpperCAmelCase__      :  List[str]				      =			0
               for i in range(len(_lowerCAmelCase		)		):
                    UpperCAmelCase__      ,		UpperCAmelCase__      :  Tuple				      =			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 _lowerCAmelCase		: _rename_shard(*_lowerCAmelCase		)		).collect()
          else:
               # don't use any pattern
               UpperCAmelCase__      :  Union[str, Any]				      =			0
               UpperCAmelCase__      :  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(_lowerCAmelCase	,						""""""		)	,						)




     def   __UpperCAmelCase  (				self	,						_lowerCAmelCase	,						):
          return SparkExamplesIterable(self.df		)














'''simple docstring'''



from math import isqrt



def  __UpperCAmelCase						(					SCREAMING_SNAKE_CASE__:						int     )	->      bool:

   """simple docstring"""
   return all(number % divisor != 0 for divisor in range(2,       isqrt(SCREAMING_SNAKE_CASE__     ) + 1     )     )



def  __UpperCAmelCase						(					SCREAMING_SNAKE_CASE__:						int = 10**6     )	->      int:

   """simple docstring"""
   __a					  =     0
   __a					  =     1
   __a					  =     7
   while prime_candidate < max_prime:
      primes_count += is_prime(SCREAMING_SNAKE_CASE__     )

      cube_index += 1
      prime_candidate += 6 * cube_index

   return primes_count


if __name__ == "__main__":
 print(f"""{solution() = }""")