"""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


_a  :  Tuple=				datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
	import pyspark

@dataclass
class 	UpperCamelCase    (					datasets.BuilderConfig		):
				UpperCAmelCase       :   Optional[datasets.Features]					=	None







def 						__UpperCAmelCase       (					UpperCAmelCase_					:       "pyspark.sql.DataFrame"    ,							UpperCAmelCase_					:       List[int]    ,							)							->    Any:

	'''simple docstring'''





	import pyspark

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



				def __iter__(self       :   List[Any])					->			Dict:
					yield from self.generate_examples_fn()



				def       _lowercase  (self       :   Tuple			,						_A       :   np.random.Generator)					->			"SparkExamplesIterable":
					__snake_case    :							Optional[int]   	=				list(range(self.df.rdd.getNumPartitions()))
					generator.shuffle(_A)
					return SparkExamplesIterable(self.df			,						partition_order=_A)



				def       _lowercase  (self       :   Optional[int]			,						_A       :   int			,						_A       :   int)					->			"SparkExamplesIterable":
					__snake_case    :							Any   	=				self.split_shard_indices_by_worker(_A			,						_A)
					return SparkExamplesIterable(self.df			,						partition_order=_A)






				@property
				def       _lowercase  (self       :   Tuple)					->			int:
					return len(self.partition_order)







class 	UpperCamelCase    (					datasets.DatasetBuilder		):
				UpperCAmelCase       :   str					=	SparkConfig



				def __init__(self       :   Optional[int]			,						_A       :   "pyspark.sql.DataFrame"			,						_A       :   str = None			,						_A       :   str = None			,						**_A       :   str			,						)					->			Dict:
					import pyspark

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

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



				def       _lowercase  (self       :   str)					->			Tuple:
					# Returns the path of the created file.
					def create_cache_and_write_probe(_A       :   Union[str, Any]):
						# 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)
						__snake_case    :							Dict   	=				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:
						__snake_case    :							Optional[int]   	=				(
						    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       _lowercase  (self       :   Tuple)					->			Any:
					return datasets.DatasetInfo(features=self.config.features)



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



				def       _lowercase  (self       :   int			,						_A       :   Tuple)					->			List[str]:
					import pyspark

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

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



				def       _lowercase  (self       :   Optional[int]			,						_A       :   str			,						_A       :   str			,						_A       :   int			,						)					->			Iterable[Tuple[int, bool, Union[int, tuple]]]:
					import pyspark

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

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

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

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

					__snake_case    :							List[Any]   	=				convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE)
					self._repartition_df_if_needed(_A)
					__snake_case    :							Optional[int]   	=				not is_remote_filesystem(self._fs)
					__snake_case    :							Optional[Any]   	=				os.path.join if is_local else posixpath.join

					__snake_case    :							Any   	=				'-TTTTT-SSSSS-of-NNNNN'
					__snake_case    :							Union[str, Any]   	=				f"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
					__snake_case    :							Any   	=				path_join(self._output_dir			,						_A)

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

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

					__snake_case    :							Tuple   	=				total_num_examples
					__snake_case    :							List[str]   	=				total_num_bytes

					# should rename everything at the end
					logger.debug(f"Renaming {total_shards} shards.")
					if total_shards > 1:
						__snake_case    :							Tuple   	=				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    :							Any   	=				self._fs

						# use the -SSSSS-of-NNNNN pattern
						def _rename_shard(
						    _A       :   int			,						_A       :   int			,						_A       :   int			,						):
							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}")			,						)

						__snake_case    :							List[str]   	=				[]
						__snake_case    :							Optional[int]   	=				0
						for i in range(len(_A)):
							__snake_case	,       __snake_case    :							Tuple   	=				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
						__snake_case    :							Any   	=				0
						__snake_case    :							str   	=				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       _lowercase  (self       :   List[Any]			,						_A       :   "datasets.SplitGenerator"			,						)					->			SparkExamplesIterable:
					return SparkExamplesIterable(self.df)






"""simple docstring"""







import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
	import torch

	from transformers import AutoModelForCausalLM

@require_torch
class 	UpperCamelCase    (					unittest.TestCase		):



				def       _lowercase  (self       :   List[Any])					->			Dict:
					__snake_case    :							int   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Optional[Any]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

					__snake_case    :							Optional[Any]   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							Any   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							str   	=				tokenizer.decode(greedy_ids[0])

					with CaptureStdout() as cs:
						__snake_case    :							List[str]   	=				TextStreamer(_A)
						model.generate(_A			,						max_new_tokens=10			,						do_sample=_A			,						streamer=_A)
					# The greedy text should be printed to stdout, except for the final "\n" in the streamer
					__snake_case    :							Optional[Any]   	=				cs.out[:-1]

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   int)					->			Optional[Any]:
					__snake_case    :							Any   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							List[Any]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Optional[int]   	=				-1

					__snake_case    :							str   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							int   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							int   	=				tokenizer.decode(greedy_ids[0])

					__snake_case    :							int   	=				TextIteratorStreamer(_A)
					__snake_case    :							List[Any]   	=				{'input_ids': input_ids, 'max_new_tokens': 10, 'do_sample': False, 'streamer': streamer}
					__snake_case    :							Any   	=				Thread(target=model.generate			,						kwargs=_A)
					thread.start()
					__snake_case    :							Dict   	=				''
					for new_text in streamer:
						streamer_text += new_text

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   Any)					->			List[str]:
					__snake_case    :							List[str]   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Any   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

					__snake_case    :							List[Any]   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							Optional[int]   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							Union[str, Any]   	=				greedy_ids[:, input_ids.shape[1] :]
					__snake_case    :							Optional[Any]   	=				tokenizer.decode(new_greedy_ids[0])

					with CaptureStdout() as cs:
						__snake_case    :							str   	=				TextStreamer(_A			,						skip_prompt=_A)
						model.generate(_A			,						max_new_tokens=10			,						do_sample=_A			,						streamer=_A)
					# The greedy text should be printed to stdout, except for the final "\n" in the streamer
					__snake_case    :							List[str]   	=				cs.out[:-1]

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   List[str])					->			Dict:
					# Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
					# with actual models -- the dummy models' tokenizers are not aligned with their models, and
					# `skip_special_tokens=True` has no effect on them
					__snake_case    :							int   	=				AutoTokenizer.from_pretrained('distilgpt2')
					__snake_case    :							Dict   	=				AutoModelForCausalLM.from_pretrained('distilgpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

					__snake_case    :							List[str]   	=				torch.ones((1, 5)			,						device=_A).long() * model.config.bos_token_id
					with CaptureStdout() as cs:
						__snake_case    :							Tuple   	=				TextStreamer(_A			,						skip_special_tokens=_A)
						model.generate(_A			,						max_new_tokens=1			,						do_sample=_A			,						streamer=_A)

					# The prompt contains a special token, so the streamer should not print it. As such, the output text, when
					# re-tokenized, must only contain one token
					__snake_case    :							Tuple   	=				cs.out[:-1]  # Remove the final "\n"
					__snake_case    :							int   	=				tokenizer(_A			,						return_tensors='pt')
					self.assertEqual(streamer_text_tokenized.input_ids.shape			,						(1, 1))






				def       _lowercase  (self       :   Optional[int])					->			List[str]:
					__snake_case    :							Optional[int]   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Optional[int]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							int   	=				-1

					__snake_case    :							List[Any]   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							List[Any]   	=				TextIteratorStreamer(_A			,						timeout=0.001)
					__snake_case    :							str   	=				{'input_ids': input_ids, 'max_new_tokens': 10, 'do_sample': False, 'streamer': streamer}
					__snake_case    :							str   	=				Thread(target=model.generate			,						kwargs=_A)
					thread.start()

					# The streamer will timeout after 0.001 seconds, so an exception will be raised
					with self.assertRaises(_A):
						__snake_case    :							Any   	=				''
						for new_text in streamer:
							streamer_text += new_text





