





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    _UpperCAmelCase							(		datasets.BuilderConfig     ):
      a						       =							None






def     _lowerCAmelCase		(  _lowerCAmelCase  ,_lowerCAmelCase  ,):
   '''simple docstring'''


   import pyspark

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






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






class    _UpperCAmelCase							(		datasets.DatasetBuilder     ):
      a						       =							SparkConfig
      def __init__(		self      ,       a__      ,       a__ = None      ,       a__ = None      ,       **a__      ,       ):
         import pyspark

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

         super().__init__(
             cache_dir=_A      ,       config_name=str(self.df.semanticHash()       )      ,       **_A      ,       )
      def 		_lowerCamelCase      (		self       ):
         # Returns the path of the created file.
         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       )
            A_     :				int			      =	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:
            A_     :				Dict			      =	(
                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 		_lowerCamelCase      (		self       ):
         return datasets.DatasetInfo(features=self.config.features       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         return [datasets.SplitGenerator(name=datasets.Split.TRAIN       )]
      def 		_lowerCamelCase      (		self      ,       a__       ):
         import pyspark

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

         A_     :				Tuple			      =	self.df.count()
         A_     :				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.
         A_     :				Dict			      =	(
             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
         )
         A_     :				Optional[int]			      =	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_     :				Tuple			      =	min(_A      ,       int(approx_total_size / max_shard_size       )       )
            A_     :				int			      =	self.df.repartition(_A       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__      ,       ):
         import pyspark

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

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

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

         A_     :				List[str]			      =	(
             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 		_lowerCamelCase      (		self      ,       a__      ,       a__ = "arrow"      ,       a__ = None      ,       a__ = None      ,       **a__      ,       ):
         self._validate_cache_dir()

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

         A_     :				str			      =	"""-TTTTT-SSSSS-of-NNNNN"""
         A_     :				Optional[int]			      =	F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
         A_     :				Optional[Any]			      =	path_join(self._output_dir      ,       _A       )

         A_     :				Union[str, Any]			      =	0
         A_     :				Union[str, Any]			      =	0
         A_     :				Optional[int]			      =	0
         A_     :				Union[str, Any]			      =	[]
         A_     :				int			      =	[]

         for task_id, content in self._prepare_split_single(_A      ,       _A      ,       _A       ):
            (
                (
                A_	
            )      ,				(
                A_	
            )      ,				(
                A_	
            )      ,				(
                A_	
            )      ,					
            )     :				Optional[int]			      =	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       )

         A_     :				int			      =	total_num_examples
         A_     :				Dict			      =	total_num_bytes

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

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






      def 		_lowerCamelCase      (		self      ,       a__      ,       ):
         return SparkExamplesIterable(self.df       )













import datasets

from .evaluate import evaluate


_lowerCAmelCase 				=     """\
@inproceedings{Rajpurkar2016SQuAD10,
  title={SQuAD: 100, 000+ Questions for Machine Comprehension of Text},
  author={Pranav Rajpurkar and Jian Zhang and Konstantin Lopyrev and Percy Liang},
  booktitle={EMNLP},
  year={2016}
}
"""

_lowerCAmelCase 				=     """
This metric wrap the official scoring script for version 1 of the Stanford Question Answering Dataset (SQuAD).

Stanford Question Answering Dataset (SQuAD) is a reading comprehension dataset, consisting of questions posed by
crowdworkers on a set of Wikipedia articles, where the answer to every question is a segment of text, or span,
from the corresponding reading passage, or the question might be unanswerable.
"""

_lowerCAmelCase 				=     """
Computes SQuAD scores (F1 and EM).
Args:
    predictions: List of question-answers dictionaries with the following key-values:
        - 'id': id of the question-answer pair as given in the references (see below)
        - 'prediction_text': the text of the answer
    references: List of question-answers dictionaries with the following key-values:
        - 'id': id of the question-answer pair (see above),
        - 'answers': a Dict in the SQuAD dataset format
            {
                'text': list of possible texts for the answer, as a list of strings
                'answer_start': list of start positions for the answer, as a list of ints
            }
            Note that answer_start values are not taken into account to compute the metric.
Returns:
    'exact_match': Exact match (the normalized answer exactly match the gold answer)
    'f1': The F-score of predicted tokens versus the gold answer
Examples:

    >>> predictions = [{'prediction_text': '1976', 'id': '56e10a3be3433e1400422b22'}]
    >>> references = [{'answers': {'answer_start': [97], 'text': ['1976']}, 'id': '56e10a3be3433e1400422b22'}]
    >>> squad_metric = datasets.load_metric(\"squad\")
    >>> results = squad_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'exact_match': 100.0, 'f1': 100.0}
"""

@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION			,       _KWARGS_DESCRIPTION     )
class    _UpperCAmelCase							(		datasets.Metric     ):
      def 		_lowerCamelCase      (		self       ):
         return datasets.MetricInfo(
             description=_DESCRIPTION      ,       citation=_CITATION      ,       inputs_description=_KWARGS_DESCRIPTION      ,       features=datasets.Features(
                 {
                     """predictions""": {"""id""": datasets.Value("""string"""       ), """prediction_text""": datasets.Value("""string"""       )},
                     """references""": {
                         """id""": datasets.Value("""string"""       ),
                         """answers""": datasets.features.Sequence(
                             {
                                 """text""": datasets.Value("""string"""       ),
                                 """answer_start""": datasets.Value("""int32"""       ),
                             }       ),
                     },
                 }       )      ,       codebase_urls=["""https://rajpurkar.github.io/SQuAD-explorer/"""]      ,       reference_urls=["""https://rajpurkar.github.io/SQuAD-explorer/"""]      ,       )






      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				Union[str, Any]			      =	{prediction["""id"""]: prediction["""prediction_text"""] for prediction in predictions}
         A_     :				Optional[Any]			      =	[
             {
                 """paragraphs""": [
                     {
                         """qas""": [
                             {
                                 """answers""": [{"""text""": answer_text} for answer_text in ref["""answers"""]["""text"""]],
                                 """id""": ref["""id"""],
                             }
                             for ref in references
                         ]
                     }
                 ]
             }
         ]
         A_     :				Union[str, Any]			      =	evaluate(dataset=a__      ,       predictions=a__       )
         return score






