
'''simple docstring'''







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


lowercase					:							Dict   =      datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
					import pyspark


@dataclass
class 		__UpperCAmelCase   (      datasets.BuilderConfig		):
							__lowercase   	= None







def        SCREAMING_SNAKE_CASE__			(     __A    ,       __A    ,       ) ->			Optional[int]:
	import pyspark

	def generate_fn():
		_snake_case     		=	df.select('*'    ,       pyspark.sql.functions.spark_partition_id().alias('part_id'			)			)
		for partition_id in partition_order:
			_snake_case     		=	df_with_partition_id.select('*'			).where(F'part_id = {partition_id}'			).drop('part_id'			)
			_snake_case     		=	partition_df.collect()
			_snake_case     		=	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					,     ):





								"""simple docstring"""






								_snake_case     		=	df
								_snake_case     		=	partition_order or range(self.df.rdd.getNumPartitions()						)
								_snake_case     		=	_generate_iterable_examples(self.df					,     self.partition_order						)


							def __iter__( self						):





								"""simple docstring"""






								yield from self.generate_examples_fn()


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_						):





								"""simple docstring"""






								_snake_case     		=	list(range(self.df.rdd.getNumPartitions()						)						)
								generator.shuffle(lowerCAmelCase_						)
								return SparkExamplesIterable(self.df					,     partition_order=lowerCAmelCase_						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_						):





								"""simple docstring"""






								_snake_case     		=	self.split_shard_indices_by_worker(lowerCAmelCase_					,     lowerCAmelCase_						)
								return SparkExamplesIterable(self.df					,     partition_order=lowerCAmelCase_						)




							@property
							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								return len(self.partition_order						)


class 		__UpperCAmelCase   (      datasets.DatasetBuilder		):
							__lowercase   	= SparkConfig


							def __init__( self					,     lowerCAmelCase_					,     lowerCAmelCase_ = None					,     lowerCAmelCase_ = None					,     **lowerCAmelCase_					,     ):





								"""simple docstring"""






								import pyspark

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

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


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								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_						)
									_snake_case     		=	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:
									_snake_case     		=	(
									    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 				lowerCamelCase       ( self						):





								"""simple docstring"""






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


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_						):





								"""simple docstring"""






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


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_						):





								"""simple docstring"""






								import pyspark

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

								_snake_case     		=	self.df.count()
								_snake_case     		=	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.
								_snake_case     		=	(
								    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
								)
								_snake_case     		=	approx_bytes_per_row * df_num_rows
								if approx_total_size > max_shard_size:
									# Make sure there is at least one row per partition.
									_snake_case     		=	min(lowerCAmelCase_					,     int(approx_total_size / max_shard_size						)						)
									_snake_case     		=	self.df.repartition(lowerCAmelCase_						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								import pyspark

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

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

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

								_snake_case     		=	(
								    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 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_ = "arrow"					,     lowerCAmelCase_ = None					,     lowerCAmelCase_ = None					,     **lowerCAmelCase_					,     ):





								"""simple docstring"""






								self._validate_cache_dir()

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

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

								_snake_case     		=	0
								_snake_case     		=	0
								_snake_case     		=	0
								_snake_case     		=	[]
								_snake_case     		=	[]

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

								_snake_case     		=	total_num_examples
								_snake_case     		=	total_num_bytes

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

									_snake_case     		=	[]
									_snake_case     		=	0
									for i in range(len(lowerCAmelCase_						)						):
										_snake_case		,				_snake_case     		=	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
									_snake_case     		=	0
									_snake_case     		=	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 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								return SparkExamplesIterable(self.df						)



'''simple docstring'''







from ..utils import DummyObject, requires_backends


class 		__UpperCAmelCase   (      metaclass=_lowerCamelCase		):
							__lowercase   	= ["""torch""", """torchsde"""]


							def __init__( self					,     *lowerCAmelCase_					,     **lowerCAmelCase_						):





								"""simple docstring"""






								requires_backends(self					,     ['torch', 'torchsde']						)


							@classmethod
							def 				lowerCamelCase       ( cls					,     *lowerCAmelCase_					,     **lowerCAmelCase_						):





								"""simple docstring"""






								requires_backends(cls					,     ['torch', 'torchsde']						)




							@classmethod
							def 				lowerCamelCase       ( cls					,     *lowerCAmelCase_					,     **lowerCAmelCase_						):





								"""simple docstring"""






								requires_backends(cls					,     ['torch', 'torchsde']						)

