

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

if TYPE_CHECKING:
						import pyspark



@dataclass
class   __lowercase				( datasets.BuilderConfig	):




			"""simple docstring"""







			_UpperCAmelCase   :							Optional[datasets.Features]										=     None







def 			A_      (    _UpperCAmelCase				,	_UpperCAmelCase				,	):
						import pyspark

						def generate_fn():
												SCREAMING_SNAKE_CASE_: List[Any]  		=       df.select("*"				,	pyspark.sql.functions.spark_partition_id().alias("part_id"    )    )
												for partition_id in partition_order:
																		SCREAMING_SNAKE_CASE_: Optional[Any]  		=       df_with_partition_id.select("*"    ).where(f"part_id = {partition_id}"    ).drop("part_id"    )
																		SCREAMING_SNAKE_CASE_: int  		=       partition_df.collect()
																		SCREAMING_SNAKE_CASE_: Optional[int]  		=       0
																		for row in rows:
																								yield f"{partition_id}_{row_id}", row.asDict()
																								row_id += 1

						return generate_fn



class   __lowercase				( _BaseExamplesIterable	):




			"""simple docstring"""







			def __init__( self  :     Optional[Any]   ,						lowerCAmelCase__  :     "pyspark.sql.DataFrame"   ,						lowerCAmelCase__  :     Optional[Any]=None   ,						):
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       df
									SCREAMING_SNAKE_CASE_: Tuple  		=       partition_order or range(self.df.rdd.getNumPartitions())
									SCREAMING_SNAKE_CASE_: Tuple  		=       _generate_iterable_examples(self.df   ,						self.partition_order)



			def __iter__( self  :     Optional[Any]):
									yield from self.generate_examples_fn()



			def        _SCREAMING_SNAKE_CASE			( self  :     str   ,						lowerCAmelCase__  :     np.random.Generator):
									SCREAMING_SNAKE_CASE_: List[str]  		=       list(range(self.df.rdd.getNumPartitions()))
									generator.shuffle(lowerCAmelCase__)
									return SparkExamplesIterable(self.df   ,						partition_order=lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[int]   ,						lowerCAmelCase__  :     int   ,						lowerCAmelCase__  :     int):
									SCREAMING_SNAKE_CASE_: str  		=       self.split_shard_indices_by_worker(lowerCAmelCase__   ,						lowerCAmelCase__)
									return SparkExamplesIterable(self.df   ,						partition_order=lowerCAmelCase__)




			@property
			def        _SCREAMING_SNAKE_CASE			( self  :     Tuple):
									return len(self.partition_order)







class   __lowercase				( datasets.DatasetBuilder	):




			"""simple docstring"""







			_UpperCAmelCase   :							Union[str, Any]										=     SparkConfig



			def __init__( self  :     Union[str, Any]   ,						lowerCAmelCase__  :     "pyspark.sql.DataFrame"   ,						lowerCAmelCase__  :     str = None   ,						lowerCAmelCase__  :     str = None   ,						**lowerCAmelCase__  :     Any   ,						):
									import pyspark

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

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



			def        _SCREAMING_SNAKE_CASE			( self  :     Union[str, Any]):
									# Returns the path of the created file.
									def create_cache_and_write_probe(lowerCAmelCase__  :     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__)
															SCREAMING_SNAKE_CASE_: Union[str, 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(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:
															SCREAMING_SNAKE_CASE_: Dict  		=       (
															    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        _SCREAMING_SNAKE_CASE			( self  :     Any):
									return datasets.DatasetInfo(features=self.config.features)



			def        _SCREAMING_SNAKE_CASE			( self  :     Dict   ,						lowerCAmelCase__  :     datasets.download.download_manager.DownloadManager):
									return [datasets.SplitGenerator(name=datasets.Split.TRAIN)]



			def        _SCREAMING_SNAKE_CASE			( self  :     Any   ,						lowerCAmelCase__  :     Optional[Any]):
									import pyspark

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

									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       self.df.count()
									SCREAMING_SNAKE_CASE_: Tuple  		=       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.
									SCREAMING_SNAKE_CASE_: Optional[int]  		=       (
									    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
									)
									SCREAMING_SNAKE_CASE_: Union[str, 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.
															SCREAMING_SNAKE_CASE_: Any  		=       min(lowerCAmelCase__   ,						int(approx_total_size / max_shard_size))
															SCREAMING_SNAKE_CASE_: Any  		=       self.df.repartition(lowerCAmelCase__)



			def        _SCREAMING_SNAKE_CASE			( self  :     Optional[Any]   ,						lowerCAmelCase__  :     str   ,						lowerCAmelCase__  :     str   ,						lowerCAmelCase__  :     int   ,						):
									import pyspark

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

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

															if writer._num_bytes > 0:
																					SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: List[Any]  		=       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__)):
																											SCREAMING_SNAKE_CASE_: List[Any]  		=       os.path.join(os.path.dirname(lowerCAmelCase__)   ,						os.path.basename(lowerCAmelCase__))
																											shutil.move(lowerCAmelCase__   ,						lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       (
									    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        _SCREAMING_SNAKE_CASE			( self  :     Optional[Any]   ,						lowerCAmelCase__  :     "datasets.SplitGenerator"   ,						lowerCAmelCase__  :     str = "arrow"   ,						lowerCAmelCase__  :     Optional[Union[str, int]] = None   ,						lowerCAmelCase__  :     Optional[int] = None   ,						**lowerCAmelCase__  :     Union[str, Any]   ,						):
									self._validate_cache_dir()

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

									SCREAMING_SNAKE_CASE_: List[Any]  		=       "-TTTTT-SSSSS-of-NNNNN"
									SCREAMING_SNAKE_CASE_: int  		=       F"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
									SCREAMING_SNAKE_CASE_: Any  		=       path_join(self._output_dir   ,						lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_: List[Any]  		=       0
									SCREAMING_SNAKE_CASE_: Any  		=       0
									SCREAMING_SNAKE_CASE_: Union[str, Any]  		=       0
									SCREAMING_SNAKE_CASE_: Optional[Any]  		=       []
									SCREAMING_SNAKE_CASE_: Tuple  		=       []

									for task_id, content in self._prepare_split_single(lowerCAmelCase__   ,						lowerCAmelCase__   ,						lowerCAmelCase__):
															(
															    (
															    SCREAMING_SNAKE_CASE_
															)		,    (
															    SCREAMING_SNAKE_CASE_
															)		,    (
															    SCREAMING_SNAKE_CASE_
															)		,    (
															    SCREAMING_SNAKE_CASE_
															)		,    
															): List[str]  		=       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__)

									SCREAMING_SNAKE_CASE_: Any  		=       total_num_examples
									SCREAMING_SNAKE_CASE_: Any  		=       total_num_bytes

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

															SCREAMING_SNAKE_CASE_: Tuple  		=       []
															SCREAMING_SNAKE_CASE_: Tuple  		=       0
															for i in range(len(lowerCAmelCase__)):
																					SCREAMING_SNAKE_CASE_		,    SCREAMING_SNAKE_CASE_: int  		=       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
															SCREAMING_SNAKE_CASE_: Optional[int]  		=       0
															SCREAMING_SNAKE_CASE_: Optional[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(lowerCAmelCase__   ,						"")   ,						)




			def        _SCREAMING_SNAKE_CASE			( self  :     List[Any]   ,						lowerCAmelCase__  :     "datasets.SplitGenerator"   ,						):
									return SparkExamplesIterable(self.df)









def 			A_      (    _UpperCAmelCase				,	_UpperCAmelCase = False    ):

						if not isinstance(_UpperCAmelCase				,	_UpperCAmelCase    ):
												SCREAMING_SNAKE_CASE_: str  		=       f"Expected string as input, found {type(_UpperCAmelCase    )}"
												raise ValueError(_UpperCAmelCase    )
						if not isinstance(_UpperCAmelCase				,	_UpperCAmelCase    ):
												SCREAMING_SNAKE_CASE_: Optional[Any]  		=       f"Expected boolean as use_pascal parameter, found {type(_UpperCAmelCase    )}"
												raise ValueError(_UpperCAmelCase    )

						SCREAMING_SNAKE_CASE_: Tuple  		=       input_str.split("_"    )

						SCREAMING_SNAKE_CASE_: str  		=       0 if use_pascal else 1

						SCREAMING_SNAKE_CASE_: int  		=       words[start_index:]

						SCREAMING_SNAKE_CASE_: List[str]  		=       [word[0].upper() + word[1:] for word in words_to_capitalize]

						SCREAMING_SNAKE_CASE_: List[Any]  		=       "" if use_pascal else words[0]

						return "".join([initial_word, *capitalized_words]    )


if __name__ == "__main__":
						from doctest import testmod

						testmod()






