






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 							=						datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
  import pyspark
@dataclass
class 							A_    (						datasets.BuilderConfig			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			Optional[datasets.Features] =					None



def      _lowerCamelCase	(						__A    :							"pyspark.sql.DataFrame"		,		__A    :							List[int]		,		)      ->					int:
    import pyspark

    def generate_fn():
        _UpperCAmelCase				:      Optional[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				:      str			  =				partition_df.collect()
            _UpperCAmelCase				:      Optional[int]			  =				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							,  _A							,  _A=None							,  ) ->		Tuple:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				df
     _UpperCAmelCase				:      Dict			  =				partition_order or range(self.df.rdd.getNumPartitions())
     _UpperCAmelCase				:      int			  =				_generate_iterable_examples(self.df							,  self.partition_order)


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

     """simple docstring"""



     yield from self.generate_examples_fn()


 def  snake_case__				(	self							,  _A) ->		"SparkExamplesIterable":

     """simple docstring"""



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


 def  snake_case__				(	self							,  _A							,  _A) ->		"SparkExamplesIterable":

     """simple docstring"""



     _UpperCAmelCase				:      Union[str, Any]			  =				self.split_shard_indices_by_worker(_A							,  _A)
     return SparkExamplesIterable(self.df							,  partition_order=_A)


 @property
 def  snake_case__				(	self) ->		int:

     """simple docstring"""



     return len(self.partition_order)



class 							A_    (						datasets.DatasetBuilder			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			Optional[Any] =					SparkConfig


 def __init__(	self							,  _A							,  _A = None							,  _A = None							,  **_A							,  ) ->		Tuple:

     """simple docstring"""



     import pyspark

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

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


 def  snake_case__				(	self) ->		Tuple:

     """simple docstring"""



     def create_cache_and_write_probe(_A):
         # 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)
         _UpperCAmelCase				:      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:
         _UpperCAmelCase				:      List[str]			  =				(
             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  snake_case__				(	self) ->		str:

     """simple docstring"""



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


 def  snake_case__				(	self							,  _A) ->		Any:

     """simple docstring"""



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


 def  snake_case__				(	self							,  _A) ->		Optional[int]:

     """simple docstring"""



     import pyspark

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

     _UpperCAmelCase				:      List[Any]			  =				self.df.count()
     _UpperCAmelCase				:      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.
     _UpperCAmelCase				:      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
     )
     _UpperCAmelCase				:      List[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.
         _UpperCAmelCase				:      Tuple			  =				min(_A							,  int(approx_total_size / max_shard_size))
         _UpperCAmelCase				:      Dict			  =				self.df.repartition(_A)


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

     """simple docstring"""



     import pyspark

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

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

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

     _UpperCAmelCase				:      List[Any]			  =				(
         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  snake_case__				(	self							,  _A							,  _A = "arrow"							,  _A = None							,  _A = None							,  **_A							,  ) ->		Tuple:

     """simple docstring"""



     self._validate_cache_dir()

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

     _UpperCAmelCase				:      int			  =				'''-TTTTT-SSSSS-of-NNNNN'''
     _UpperCAmelCase				:      Tuple			  =				f'''{self.name}-{split_generator.name}{SUFFIX}.{file_format}'''
     _UpperCAmelCase				:      Optional[Any]			  =				path_join(self._output_dir							,  _A)

     _UpperCAmelCase				:      List[str]			  =				0
     _UpperCAmelCase				:      List[Any]			  =				0
     _UpperCAmelCase				:      int			  =				0
     _UpperCAmelCase				:      int			  =				[]
     _UpperCAmelCase				:      Optional[Any]			  =				[]

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

     _UpperCAmelCase				:      Any			  =				total_num_examples
     _UpperCAmelCase				:      Dict			  =				total_num_bytes

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

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

         _UpperCAmelCase				:      Union[str, Any]			  =				[]
         _UpperCAmelCase				:      int			  =				0
         for i in range(len(_A)):
             _UpperCAmelCase  , _UpperCAmelCase				:      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
         _UpperCAmelCase				:      Any			  =				0
         _UpperCAmelCase				:      int			  =				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  snake_case__				(	self							,  _A							,  ) ->		SparkExamplesIterable:

     """simple docstring"""



     return SparkExamplesIterable(self.df)












import math_equivalence  # From: git+https://github.com/hendrycks/math.git

import datasets


SCREAMING_SNAKE_CASE 							=						'\\n@article{hendrycksmath2021,\n  title={Measuring Mathematical Problem Solving With the MATH Dataset},\n  author={Dan Hendrycks\n    and Collin Burns\n    and Saurav Kadavath\n    and Akul Arora\n    and Steven Basart\n    and Eric Tang\n    and Dawn Song\n    and Jacob Steinhardt},\n  journal={arXiv preprint arXiv:2103.03874},\n  year={2021}\n}\n'


SCREAMING_SNAKE_CASE 							=						'\\nThis metric is used to assess performance on the Mathematics Aptitude Test of Heuristics (MATH) dataset.\nIt first canonicalizes the inputs (e.g., converting "1/2" to "\\frac{1}{2}") and then computes accuracy.\n'


SCREAMING_SNAKE_CASE 							=						R'\nCalculates accuracy after canonicalizing inputs.\n\nArgs:\n    predictions: list of predictions to score. Each prediction\n        is a string that contains natural language and LaTex.\n    references: list of reference for each prediction. Each\n        reference is a string that contains natural language\n        and LaTex.\nReturns:\n    accuracy: accuracy after canonicalizing inputs\n        (e.g., converting "1/2" to "\\frac{1}{2}")\n\nExamples:\n    >>> metric = datasets.load_metric("competition_math")\n    >>> results = metric.compute(references=["\\frac{1}{2}"], predictions=["1/2"])\n    >>> print(results)\n    {\'accuracy\': 1.0}\n'
@datasets.utils.file_utils.add_end_docstrings(_DESCRIPTION ,   _KWARGS_DESCRIPTION			)
class 							A_    (						datasets.Metric			):






 '''simple docstring'''




 def  snake_case__				(	self) ->		Optional[Any]:

     """simple docstring"""



     return datasets.MetricInfo(
         description=_DESCRIPTION							,  citation=_CITATION							,  inputs_description=_KWARGS_DESCRIPTION							,  features=datasets.Features(
             {
                 '''predictions''': datasets.Value('''string'''),
                 '''references''': datasets.Value('''string'''),
             })							,  homepage='''https://github.com/hendrycks/math'''							,  codebase_urls=['''https://github.com/hendrycks/math''']							,  )


 def  snake_case__				(	self							,  _A							,  _A) ->		Tuple:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				0.0
     for i, j in zip(_A							,  _A):
         n_correct += 1.0 if math_equivalence.is_equiv(_A							,  _A) else 0.0
     _UpperCAmelCase				:      Tuple			  =				n_correct / len(_A)
     return {
         "accuracy": accuracy,
     }




