

lowerCamelCase      				=     [
    '''Audio''',
    '''Array2D''',
    '''Array3D''',
    '''Array4D''',
    '''Array5D''',
    '''ClassLabel''',
    '''Features''',
    '''Sequence''',
    '''Value''',
    '''Image''',
    '''Translation''',
    '''TranslationVariableLanguages''',
]
from .audio import Audio
from .features import ArrayaD, ArrayaD, ArrayaD, ArrayaD, ClassLabel, Features, Sequence, Value
from .image import Image
from .translation import Translation, TranslationVariableLanguages









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


							"""simple docstring"""





							__UpperCAmelCase       =None


def        __lowercase    (  UpperCAmelCase__		,					UpperCAmelCase__		,					):

					"""simple docstring"""






					import pyspark

					def generate_fn():
										__lowerCAmelCase					=							df.select('*'		,					pyspark.sql.functions.spark_partition_id().alias('part_id'      )      )
										for partition_id in partition_order:
															__lowerCAmelCase					=							df_with_partition_id.select('*'      ).where(F"""part_id = {partition_id}"""      ).drop('part_id'      )
															__lowerCAmelCase					=							partition_df.collect()
															__lowerCAmelCase					=							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					,						_A					,						_A=None					,						):
												__lowerCAmelCase					=							df
												__lowerCAmelCase					=							partition_order or range(self.df.rdd.getNumPartitions()							)
												__lowerCAmelCase					=							_generate_iterable_examples(self.df					,						self.partition_order							)




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




							def        A__  ( self					,						_A							):
												__lowerCAmelCase					=							list(range(self.df.rdd.getNumPartitions()							)							)
												generator.shuffle(_A							)
												return SparkExamplesIterable(self.df					,						partition_order=_A							)




							def        A__  ( self					,						_A					,						_A							):
												__lowerCAmelCase					=							self.split_shard_indices_by_worker(_A					,						_A							)
												return SparkExamplesIterable(self.df					,						partition_order=_A							)




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





class     snake_case_			(				datasets.DatasetBuilder							):


							"""simple docstring"""





							__UpperCAmelCase       =SparkConfig




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

												__lowerCAmelCase					=							pyspark.sql.SparkSession.builder.getOrCreate()
												__lowerCAmelCase					=							df
												__lowerCAmelCase					=							working_dir

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




							def        A__  ( 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							)
																	__lowerCAmelCase					=							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:
																	__lowerCAmelCase					=							(
																	    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        A__  ( self							):
												return datasets.DatasetInfo(features=self.config.features							)




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




							def        A__  ( self					,						_A							):
												import pyspark

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

												__lowerCAmelCase					=							self.df.count()
												__lowerCAmelCase					=							df_num_rows if df_num_rows <= 1_0_0 else 1_0_0
												# Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
												__lowerCAmelCase					=							(
												    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
												)
												__lowerCAmelCase					=							approx_bytes_per_row * df_num_rows
												if approx_total_size > max_shard_size:
																	# Make sure there is at least one row per partition.
																	__lowerCAmelCase					=							min(_A					,						int(approx_total_size / max_shard_size							)							)
																	__lowerCAmelCase					=							self.df.repartition(_A							)




							def        A__  ( self					,						_A					,						_A					,						_A					,						):
												import pyspark

												__lowerCAmelCase					=							ParquetWriter if file_format == 'parquet' else ArrowWriter
												__lowerCAmelCase					=							os.path.join(self._working_dir					,						os.path.basename(_A							)							) if self._working_dir else fpath
												__lowerCAmelCase					=							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.
												__lowerCAmelCase					=							self.config.features
												__lowerCAmelCase					=							self._writer_batch_size
												__lowerCAmelCase					=							self._fs.storage_options

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

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

												__lowerCAmelCase					=							(
												    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        A__  ( self					,						_A					,						_A = "arrow"					,						_A = None					,						_A = None					,						**_A					,						):
												self._validate_cache_dir()

												__lowerCAmelCase					=							convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE							)
												self._repartition_df_if_needed(_A							)
												__lowerCAmelCase					=							not is_remote_filesystem(self._fs							)
												__lowerCAmelCase					=							os.path.join if is_local else posixpath.join

												__lowerCAmelCase					=							'-TTTTT-SSSSS-of-NNNNN'
												__lowerCAmelCase					=							F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
												__lowerCAmelCase					=							path_join(self._output_dir					,						_A							)

												__lowerCAmelCase					=							0
												__lowerCAmelCase					=							0
												__lowerCAmelCase					=							0
												__lowerCAmelCase					=							[]
												__lowerCAmelCase					=							[]

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

												__lowerCAmelCase					=							total_num_examples
												__lowerCAmelCase					=							total_num_bytes

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

																	__lowerCAmelCase					=							[]
																	__lowerCAmelCase					=							0
																	for i in range(len(_A							)							):
																						__lowerCAmelCase,	__lowerCAmelCase					=							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
																	__lowerCAmelCase					=							0
																	__lowerCAmelCase					=							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        A__  ( self					,						_A					,						):
												return SparkExamplesIterable(self.df							)






