




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		:				List[str]       			=		datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
		import pyspark




@dataclass
class     snake_case__							(datasets.BuilderConfig   ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_       :							Optional[datasets.Features]							 =					None




def 			__magic_name__				( A		:					"pyspark.sql.DataFrame",  A		:					List[int],  ):






			'''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     snake_case__							(_BaseExamplesIterable   ):
				"""simple docstring"""



				def __init__(      self				:     Optional[int]					,      __lowerCamelCase				:     "pyspark.sql.DataFrame"					,      __lowerCamelCase				:     int=None					,      )					->     Optional[Any]:
							a    =			df
							a    =			partition_order or range(self.df.rdd.getNumPartitions()							)
							a    =			_generate_iterable_examples(self.df					,      self.partition_order							)





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





				def 							__UpperCAmelCase							(      self				:     Any					,      __lowerCamelCase				:     np.random.Generator							)					->     "SparkExamplesIterable":
							a    =			list(range(self.df.rdd.getNumPartitions()							)							)
							generator.shuffle(__lowerCamelCase							)
							return SparkExamplesIterable(self.df					,      partition_order=__lowerCamelCase							)





				def 							__UpperCAmelCase							(      self				:     str					,      __lowerCamelCase				:     int					,      __lowerCamelCase				:     int							)					->     "SparkExamplesIterable":
							a    =			self.split_shard_indices_by_worker(__lowerCamelCase					,      __lowerCamelCase							)
							return SparkExamplesIterable(self.df					,      partition_order=__lowerCamelCase							)





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




class     snake_case__							(datasets.DatasetBuilder   ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_       :							Optional[int]							 =					SparkConfig





				def __init__(      self				:     Tuple					,      __lowerCamelCase				:     "pyspark.sql.DataFrame"					,      __lowerCamelCase				:     str = None					,      __lowerCamelCase				:     str = None					,      **__lowerCamelCase				:     List[str]					,      )					->     Optional[Any]:
							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				:     Union[str, Any]							)					->     List[str]:
							# Returns the path of the created file.
							def create_cache_and_write_probe(__lowerCamelCase				:     Optional[int]							):
										# 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				:     Union[str, Any]							)					->     str:
							return datasets.DatasetInfo(features=self.config.features							)





				def 							__UpperCAmelCase							(      self				:     int					,      __lowerCamelCase				:     datasets.download.download_manager.DownloadManager							)					->     Union[str, Any]:
							return [datasets.SplitGenerator(name=datasets.Split.TRAIN							)]





				def 							__UpperCAmelCase							(      self				:     str					,      __lowerCamelCase				:     str							)					->     Union[str, Any]:
							import pyspark

							def get_arrow_batch_size(__lowerCamelCase				:     int							):
										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				:     Optional[Any]					,      __lowerCamelCase				:     str					,      __lowerCamelCase				:     str					,      __lowerCamelCase				:     int					,      )					->     Iterable[Tuple[int, bool, Union[int, tuple]]]:
							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(__lowerCamelCase				:     Dict							):
										# 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				:     Dict					,      __lowerCamelCase				:     "datasets.SplitGenerator"					,      __lowerCamelCase				:     str = "arrow"					,      __lowerCamelCase				:     Optional[Union[str, int]] = None					,      __lowerCamelCase				:     Optional[int] = None					,      **__lowerCamelCase				:     Any					,      )					->     Union[str, Any]:
							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(
										    __lowerCamelCase				:     int					,      __lowerCamelCase				:     int					,      __lowerCamelCase				:     int					,      ):
													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 __lowerCamelCase							: _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				:     List[Any]					,      __lowerCamelCase				:     "datasets.SplitGenerator"					,      )					->     SparkExamplesIterable:
							return SparkExamplesIterable(self.df							)












from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig


__lowerCAmelCase		:				int       			=		{
    'albert-base-v1': 'https://huggingface.co/albert-base-v1/resolve/main/config.json',
    'albert-large-v1': 'https://huggingface.co/albert-large-v1/resolve/main/config.json',
    'albert-xlarge-v1': 'https://huggingface.co/albert-xlarge-v1/resolve/main/config.json',
    'albert-xxlarge-v1': 'https://huggingface.co/albert-xxlarge-v1/resolve/main/config.json',
    'albert-base-v2': 'https://huggingface.co/albert-base-v2/resolve/main/config.json',
    'albert-large-v2': 'https://huggingface.co/albert-large-v2/resolve/main/config.json',
    'albert-xlarge-v2': 'https://huggingface.co/albert-xlarge-v2/resolve/main/config.json',
    'albert-xxlarge-v2': 'https://huggingface.co/albert-xxlarge-v2/resolve/main/config.json',
}




class     snake_case__							(_UpperCamelCase   ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_       :							Optional[int]							 =					"""albert"""





				def __init__(      self				:     List[str]					,      __lowerCamelCase				:     Union[str, Any]=3_00_00					,      __lowerCamelCase				:     Union[str, Any]=1_28					,      __lowerCamelCase				:     Any=40_96					,      __lowerCamelCase				:     str=12					,      __lowerCamelCase				:     Union[str, Any]=1					,      __lowerCamelCase				:     Union[str, Any]=64					,      __lowerCamelCase				:     Dict=1_63_84					,      __lowerCamelCase				:     List[Any]=1					,      __lowerCamelCase				:     Optional[Any]="gelu_new"					,      __lowerCamelCase				:     List[str]=0					,      __lowerCamelCase				:     List[str]=0					,      __lowerCamelCase				:     Tuple=5_12					,      __lowerCamelCase				:     Optional[Any]=2					,      __lowerCamelCase				:     Optional[Any]=0.02					,      __lowerCamelCase				:     Optional[Any]=1e-12					,      __lowerCamelCase				:     Union[str, Any]=0.1					,      __lowerCamelCase				:     Any="absolute"					,      __lowerCamelCase				:     List[str]=0					,      __lowerCamelCase				:     List[str]=2					,      __lowerCamelCase				:     Any=3					,      **__lowerCamelCase				:     int					,      )					->     int:
							super().__init__(pad_token_id=__lowerCamelCase					,      bos_token_id=__lowerCamelCase					,      eos_token_id=__lowerCamelCase					,      **__lowerCamelCase							)

							a    =			vocab_size
							a    =			embedding_size
							a    =			hidden_size
							a    =			num_hidden_layers
							a    =			num_hidden_groups
							a    =			num_attention_heads
							a    =			inner_group_num
							a    =			hidden_act
							a    =			intermediate_size
							a    =			hidden_dropout_prob
							a    =			attention_probs_dropout_prob
							a    =			max_position_embeddings
							a    =			type_vocab_size
							a    =			initializer_range
							a    =			layer_norm_eps
							a    =			classifier_dropout_prob
							a    =			position_embedding_type




class     snake_case__							(_UpperCamelCase   ):
				"""simple docstring"""



				@property
				def 							__UpperCAmelCase							(      self				:     List[Any]							)					->     Mapping[str, Mapping[int, str]]:
							if self.task == "multiple-choice":
										a    =			{0: "batch", 1: "choice", 2: "sequence"}
							else:
										a    =			{0: "batch", 1: "sequence"}
							return OrderedDict(
							    [
							        ("input_ids", dynamic_axis),
							        ("attention_mask", dynamic_axis),
							        ("token_type_ids", dynamic_axis),
							    ]							)






