






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


SCREAMING_SNAKE_CASE__				     =			datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
							import pyspark






@dataclass
class 			_UpperCamelCase(       datasets.BuilderConfig		):
		__SCREAMING_SNAKE_CASE  :       Optional[datasets.Features]							=							None




def        UpperCAmelCase__		(						lowerCamelCase_       :       "pyspark.sql.DataFrame"   ,  lowerCamelCase_       :       List[int]   ,  ):
			import pyspark

			def generate_fn():
						__a						:  List[Any]									=     df.select('*'   ,  pyspark.sql.functions.spark_partition_id().alias('part_id'				)				)
						for partition_id in partition_order:
									__a						:  Optional[int]									=     df_with_partition_id.select('*'				).where(f'''part_id = {partition_id}'''				).drop('part_id'				)
									__a						:  Optional[Any]									=     partition_df.collect()
									__a						:  Union[str, Any]									=     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		:    Any    ,       SCREAMING_SNAKE_CASE__		:    "pyspark.sql.DataFrame"    ,       SCREAMING_SNAKE_CASE__		:    Dict=None    ,       ):



					'''simple docstring'''



					__a						:  List[str]									=     df
					__a						:  Tuple									=     partition_order or range(self.df.rdd.getNumPartitions()    )
					__a						:  List[Any]									=     _generate_iterable_examples(self.df    ,       self.partition_order    )

		def __iter__(					self		:    Tuple    ):



					'''simple docstring'''



					yield from self.generate_examples_fn()

		def    __lowerCAmelCase  (					self		:    Optional[int]    ,       SCREAMING_SNAKE_CASE__		:    np.random.Generator    ):



					'''simple docstring'''



					__a						:  Union[str, Any]									=     list(range(self.df.rdd.getNumPartitions()    )    )
					generator.shuffle(SCREAMING_SNAKE_CASE__    )
					return SparkExamplesIterable(self.df    ,       partition_order=SCREAMING_SNAKE_CASE__    )

		def    __lowerCAmelCase  (					self		:    str    ,       SCREAMING_SNAKE_CASE__		:    int    ,       SCREAMING_SNAKE_CASE__		:    int    ):



					'''simple docstring'''



					__a						:  Union[str, Any]									=     self.split_shard_indices_by_worker(SCREAMING_SNAKE_CASE__    ,       SCREAMING_SNAKE_CASE__    )
					return SparkExamplesIterable(self.df    ,       partition_order=SCREAMING_SNAKE_CASE__    )



		@property
		def    __lowerCAmelCase  (					self		:    Dict    ):



					'''simple docstring'''



					return len(self.partition_order    )






class 			_UpperCamelCase(       datasets.DatasetBuilder		):
		__SCREAMING_SNAKE_CASE  :       List[str]							=							SparkConfig

		def __init__(					self		:    List[str]    ,       SCREAMING_SNAKE_CASE__		:    "pyspark.sql.DataFrame"    ,       SCREAMING_SNAKE_CASE__		:    str = None    ,       SCREAMING_SNAKE_CASE__		:    str = None    ,       **SCREAMING_SNAKE_CASE__		:    Optional[int]    ,       ):



					'''simple docstring'''



					import pyspark

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

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

		def    __lowerCAmelCase  (					self		:    Optional[int]    ):



					'''simple docstring'''



					def create_cache_and_write_probe(SCREAMING_SNAKE_CASE__		:    List[str]    ):
								# 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=SCREAMING_SNAKE_CASE__    )
								__a						:  List[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(SCREAMING_SNAKE_CASE__    ,       '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						:  List[Any]									=     (
								    self._spark.sparkContext.parallelize(range(1    )    ,       1    ).mapPartitions(SCREAMING_SNAKE_CASE__    ).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		:    List[str]    ):



					'''simple docstring'''



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

		def    __lowerCAmelCase  (					self		:    int    ,       SCREAMING_SNAKE_CASE__		:    datasets.download.download_manager.DownloadManager    ):



					'''simple docstring'''



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

		def    __lowerCAmelCase  (					self		:    List[str]    ,       SCREAMING_SNAKE_CASE__		:    Any    ):



					'''simple docstring'''



					import pyspark

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

					__a						:  List[str]									=     self.df.count()
					__a						:  Dict									=     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.
					__a						:  List[str]									=     (
					    self.df.limit(SCREAMING_SNAKE_CASE__    )
					    .repartition(1    )
					    .mapInArrow(SCREAMING_SNAKE_CASE__    ,       'batch_bytes: long'    )
					    .agg(pyspark.sql.functions.sum('batch_bytes'    ).alias('sample_bytes'    )    )
					    .collect()[0]
					    .sample_bytes
					    / sample_num_rows
					)
					__a						:  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.
								__a						:  Union[str, Any]									=     min(SCREAMING_SNAKE_CASE__    ,       int(approx_total_size / max_shard_size    )    )
								__a						:  int									=     self.df.repartition(SCREAMING_SNAKE_CASE__    )

		def    __lowerCAmelCase  (					self		:    Tuple    ,       SCREAMING_SNAKE_CASE__		:    str    ,       SCREAMING_SNAKE_CASE__		:    str    ,       SCREAMING_SNAKE_CASE__		:    int    ,       ):



					'''simple docstring'''



					import pyspark

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

					def write_arrow(SCREAMING_SNAKE_CASE__		:    Optional[int]    ):
								# Within the same SparkContext, no two task attempts will share the same attempt ID.
								__a						:  Any									=     pyspark.TaskContext().taskAttemptId()
								__a						:  str									=     next(SCREAMING_SNAKE_CASE__    ,       SCREAMING_SNAKE_CASE__    )
								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						:  Any									=     0
								__a						:  List[str]									=     writer_class(
								    features=SCREAMING_SNAKE_CASE__    ,       path=working_fpath.replace('SSSSS'    ,       f'''{shard_id:05d}'''    ).replace('TTTTT'    ,       f'''{task_id:05d}'''    )    ,       writer_batch_size=SCREAMING_SNAKE_CASE__    ,       storage_options=SCREAMING_SNAKE_CASE__    ,       embed_local_files=SCREAMING_SNAKE_CASE__    ,       )
								__a						:  Optional[Any]									=     pa.Table.from_batches([first_batch]    )
								writer.write_table(SCREAMING_SNAKE_CASE__    )
								for batch in it:
											if max_shard_size is not None and writer._num_bytes >= max_shard_size:
														__a						,					__a						:  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
														__a						:  Optional[Any]									=     writer_class(
														    features=writer._features    ,       path=working_fpath.replace('SSSSS'    ,       f'''{shard_id:05d}'''    ).replace('TTTTT'    ,       f'''{task_id:05d}'''    )    ,       writer_batch_size=SCREAMING_SNAKE_CASE__    ,       storage_options=SCREAMING_SNAKE_CASE__    ,       embed_local_files=SCREAMING_SNAKE_CASE__    ,       )
											__a						:  Union[str, Any]									=     pa.Table.from_batches([batch]    )
											writer.write_table(SCREAMING_SNAKE_CASE__    )

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

					__a						:  Dict									=     (
					    self.df.mapInArrow(SCREAMING_SNAKE_CASE__    ,       '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		:    Union[str, Any]    ,       SCREAMING_SNAKE_CASE__		:    "datasets.SplitGenerator"    ,       SCREAMING_SNAKE_CASE__		:    str = "arrow"    ,       SCREAMING_SNAKE_CASE__		:    Optional[Union[str, int]] = None    ,       SCREAMING_SNAKE_CASE__		:    Optional[int] = None    ,       **SCREAMING_SNAKE_CASE__		:    Optional[Any]    ,       ):



					'''simple docstring'''



					self._validate_cache_dir()

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

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

					__a						:  Any									=     0
					__a						:  Dict									=     0
					__a						:  int									=     0
					__a						:  List[str]									=     []
					__a						:  Optional[int]									=     []

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

					__a						:  List[str]									=     total_num_examples
					__a						:  Optional[int]									=     total_num_bytes

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

								# use the -SSSSS-of-NNNNN pattern
								def _rename_shard(
								    SCREAMING_SNAKE_CASE__		:    int    ,       SCREAMING_SNAKE_CASE__		:    int    ,       SCREAMING_SNAKE_CASE__		:    int    ,       ):
											rename(
											    SCREAMING_SNAKE_CASE__    ,       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						:  Union[str, Any]									=     []
								__a						:  List[str]									=     0
								for i in range(len(SCREAMING_SNAKE_CASE__    )    ):
											__a						,					__a						:  Union[str, Any]									=     task_id_and_num_shards[i]
											for shard_id in range(SCREAMING_SNAKE_CASE__    ):
														args.append([task_id, shard_id, global_shard_id]    )
														global_shard_id += 1
								self._spark.sparkContext.parallelize(SCREAMING_SNAKE_CASE__    ,       len(SCREAMING_SNAKE_CASE__    )    ).map(lambda SCREAMING_SNAKE_CASE__    : _rename_shard(*SCREAMING_SNAKE_CASE__    )    ).collect()
					else:
								# don't use any pattern
								__a						:  List[Any]									=     0
								__a						:  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(SCREAMING_SNAKE_CASE__    ,       ''    )    ,       )



		def    __lowerCAmelCase  (					self		:    str    ,       SCREAMING_SNAKE_CASE__		:    "datasets.SplitGenerator"    ,       ):



					'''simple docstring'''



					return SparkExamplesIterable(self.df    )











import gc
import unittest

from parameterized import parameterized

from diffusers import FlaxUNetaDConditionModel
from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import load_hf_numpy, require_flax, slow


if is_flax_available():
		import jax
		import jax.numpy as jnp

@slow
@require_flax
class 						__lowerCAmelCase      (      unittest.TestCase					):
							def  lowerCamelCase__				(		self		:Tuple						, __magic_name__		:int						, __magic_name__		:Optional[Any]							):







								'''simple docstring'''







								return F'gaussian_noise_s={seed}_shape={"_".join([str(__magic_name__							) for s in shape]							)}.npy'
							def  lowerCamelCase__				(		self		:List[str]							):







								'''simple docstring'''







								super().tearDown()
								gc.collect()
							def  lowerCamelCase__				(		self		:int						, __magic_name__		:List[str]=0						, __magic_name__		:List[Any]=(4, 4, 64, 64)						, __magic_name__		:Optional[int]=False							):







								'''simple docstring'''







								a									=				jnp.bfloataa if fpaa else jnp.floataa
								a									=				jnp.array(load_hf_numpy(self.get_file_format(__magic_name__						, __magic_name__							)							)						, dtype=__magic_name__							)
								return image
							def  lowerCamelCase__				(		self		:Any						, __magic_name__		:int=False						, __magic_name__		:List[str]="CompVis/stable-diffusion-v1-4"							):







								'''simple docstring'''







								a									=				jnp.bfloataa if fpaa else jnp.floataa
								a									=				"""bf16""" if fpaa else None

								a							,					a									=				FlaxUNetaDConditionModel.from_pretrained(
								    __magic_name__						, subfolder="""unet"""						, dtype=__magic_name__						, revision=__magic_name__							)
								return model, params
							def  lowerCamelCase__				(		self		:Optional[Any]						, __magic_name__		:Union[str, Any]=0						, __magic_name__		:Any=(4, 77, 768)						, __magic_name__		:Optional[Any]=False							):







								'''simple docstring'''







								a									=				jnp.bfloataa if fpaa else jnp.floataa
								a									=				jnp.array(load_hf_numpy(self.get_file_format(__magic_name__						, __magic_name__							)							)						, dtype=__magic_name__							)
								return hidden_states
							@parameterized.expand(
							    [
							        # fmt: off
							        [83, 4, [-0.2323, -0.1304, 0.0813, -0.3093, -0.0919, -0.1571, -0.1125, -0.5806]],
							        [17, 0.55, [-0.0831, -0.2443, 0.0901, -0.0919, 0.3396, 0.0103, -0.3743, 0.0701]],
							        [8, 0.89, [-0.4863, 0.0859, 0.0875, -0.1658, 0.9199, -0.0114, 0.4839, 0.4639]],
							        [3, 1000, [-0.5649, 0.2402, -0.5518, 0.1248, 1.1328, -0.2443, -0.0325, -1.0078]],
							        # fmt: on
							    ]							)
							def  lowerCamelCase__				(		self		:Optional[Any]						, __magic_name__		:str						, __magic_name__		:Optional[Any]						, __magic_name__		:Optional[int]							):







								'''simple docstring'''







								a							,					a									=				self.get_unet_model(model_id="""CompVis/stable-diffusion-v1-4"""						, fpaa=__magic_name__							)
								a									=				self.get_latents(__magic_name__						, fpaa=__magic_name__							)
								a									=				self.get_encoder_hidden_states(__magic_name__						, fpaa=__magic_name__							)

								a									=				model.apply(
								    {"""params""": params}						, __magic_name__						, jnp.array(__magic_name__						, dtype=jnp.intaa							)						, encoder_hidden_states=__magic_name__						, ).sample

								assert sample.shape == latents.shape

								a									=				jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())							)						, dtype=jnp.floataa							)
								a									=				jnp.array(__magic_name__						, dtype=jnp.floataa							)

								# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, in the same hardware
								assert jnp.allclose(__magic_name__						, __magic_name__						, atol=1E-2							)





							@parameterized.expand(
							    [
							        # fmt: off
							        [83, 4, [0.1514, 0.0807, 0.1624, 0.1016, -0.1896, 0.0263, 0.0677, 0.2310]],
							        [17, 0.55, [0.1164, -0.0216, 0.0170, 0.1589, -0.3120, 0.1005, -0.0581, -0.1458]],
							        [8, 0.89, [-0.1758, -0.0169, 0.1004, -0.1411, 0.1312, 0.1103, -0.1996, 0.2139]],
							        [3, 1000, [0.1214, 0.0352, -0.0731, -0.1562, -0.0994, -0.0906, -0.2340, -0.0539]],
							        # fmt: on
							    ]							)
							def  lowerCamelCase__				(		self		:Optional[Any]						, __magic_name__		:List[Any]						, __magic_name__		:int						, __magic_name__		:str							):







								'''simple docstring'''







								a							,					a									=				self.get_unet_model(model_id="""stabilityai/stable-diffusion-2"""						, fpaa=__magic_name__							)
								a									=				self.get_latents(__magic_name__						, shape=(4, 4, 96, 96)						, fpaa=__magic_name__							)
								a									=				self.get_encoder_hidden_states(__magic_name__						, shape=(4, 77, 1024)						, fpaa=__magic_name__							)

								a									=				model.apply(
								    {"""params""": params}						, __magic_name__						, jnp.array(__magic_name__						, dtype=jnp.intaa							)						, encoder_hidden_states=__magic_name__						, ).sample

								assert sample.shape == latents.shape

								a									=				jnp.asarray(jax.device_get((sample[-1, -2:, -2:, :2].flatten())							)						, dtype=jnp.floataa							)
								a									=				jnp.array(__magic_name__						, dtype=jnp.floataa							)

								# Found torch (float16) and flax (bfloat16) outputs to be within this tolerance, on the same hardware
								assert jnp.allclose(__magic_name__						, __magic_name__						, atol=1E-2							)

