




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


_SCREAMING_SNAKE_CASE							:       List[Any]						=			datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
		import pyspark




@dataclass
class 			_snake_case		(     datasets.BuilderConfig							):
			lowerCAmelCase_					:				Optional[datasets.Features]							=   None


def 					UpperCamelCase_(     snake_case							:	"pyspark.sql.DataFrame"							,    snake_case							:	List[int]							,    ):

							'''simple docstring'''






							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 			_snake_case		(     _BaseExamplesIterable							):




			def __init__( self    ,							a__    ,							a__=None    ,							)    ->    Optional[Any]:

										'''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							)    ->    int:

										'''simple docstring'''




										yield from self.generate_examples_fn()




			def        lowerCAmelCase__    ( self    ,							a__							)    ->    "SparkExamplesIterable":

										'''simple docstring'''




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




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__							)    ->    "SparkExamplesIterable":

										'''simple docstring'''




										snake_case_      =				self.split_shard_indices_by_worker(a__    ,							a__							)
										return SparkExamplesIterable(self.df    ,							partition_order=a__							)




			@property
			def        lowerCAmelCase__    ( self							)    ->    int:

										'''simple docstring'''




										return len(self.partition_order							)




class 			_snake_case		(     datasets.DatasetBuilder							):
			lowerCAmelCase_					:				Optional[Any]							=   SparkConfig




			def __init__( self    ,							a__    ,							a__ = None    ,							a__ = None    ,							**a__    ,							)    ->    Optional[Any]:

										'''simple docstring'''




										import pyspark

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

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




			def        lowerCAmelCase__    ( self							)    ->    Optional[Any]:

										'''simple docstring'''




										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__							)
																	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(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_      =				(
																	    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        lowerCAmelCase__    ( self							)    ->    Optional[Any]:

										'''simple docstring'''




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




			def        lowerCAmelCase__    ( self    ,							a__							)    ->    Any:

										'''simple docstring'''




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




			def        lowerCAmelCase__    ( self    ,							a__							)    ->    Optional[Any]:

										'''simple docstring'''




										import pyspark

										def get_arrow_batch_size(a__							):
																	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 <= 100 else 100
										# Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
										snake_case_      =				(
										    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_      =				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(a__    ,							int(approx_total_size / max_shard_size							)							)
																	snake_case_      =				self.df.repartition(a__							)




			def        lowerCAmelCase__    ( self    ,							a__    ,							a__    ,							a__    ,							)    ->    Iterable[Tuple[int, bool, Union[int, tuple]]]:

										'''simple docstring'''




										import pyspark

										snake_case_      =				ParquetWriter if file_format == "parquet" else ArrowWriter
										snake_case_      =				os.path.join(self._working_dir    ,							os.path.basename(a__							)							) 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(a__							):
																	# Within the same SparkContext, no two task attempts will share the same attempt ID.
																	snake_case_      =				pyspark.TaskContext().taskAttemptId()
																	snake_case_      =				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_      =				0
																	snake_case_      =				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_      =				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_      =				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=a__    ,							storage_options=a__    ,							embed_local_files=a__    ,							)
																								snake_case_      =				pa.Table.from_batches([batch]							)
																								writer.write_table(a__							)

																	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(a__							)							):
																															snake_case_      =				os.path.join(os.path.dirname(a__							)    ,							os.path.basename(a__							)							)
																															shutil.move(a__    ,							a__							)

										snake_case_      =				(
										    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        lowerCAmelCase__    ( self    ,							a__    ,							a__ = "arrow"    ,							a__ = None    ,							a__ = None    ,							**a__    ,							)    ->    Dict:

										'''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(a__							)
										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    ,							a__							)

										snake_case_      =				0
										snake_case_      =				0
										snake_case_      =				0
										snake_case_      =				[]
										snake_case_      =				[]

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

										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(
																	    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}'							)    ,							)

																	snake_case_      =				[]
																	snake_case_      =				0
																	for i in range(len(a__							)							):
																								snake_case_ ,				snake_case_      =				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_      =				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(a__    ,							""							)    ,							)




			def        lowerCAmelCase__    ( self    ,							a__    ,							)    ->    SparkExamplesIterable:

										'''simple docstring'''




										return SparkExamplesIterable(self.df							)










'''simple docstring'''





from __future__ import annotations

import numpy as np


def 					UpperCamelCase_(     snake_case							:	np.ndarray					):

							'''simple docstring'''






							snake_case_ ,				snake_case_      =				np.shape(snake_case					)
							if rows != columns:
														snake_case_      =				(
														    "'table' has to be of square shaped array but got a "
														    f'{rows}x{columns} array:\n{table}'
														)
														raise ValueError(snake_case					)

							snake_case_      =				np.zeros((rows, columns)					)
							snake_case_      =				np.zeros((rows, columns)					)
							for i in range(snake_case					):
														for j in range(snake_case					):
																					snake_case_      =				sum(lower[i][k] * upper[k][j] for k in range(snake_case					)					)
																					if upper[j][j] == 0:
																												raise ArithmeticError("No LU decomposition exists"					)
																					snake_case_      =				(table[i][j] - total) / upper[j][j]
														snake_case_      =				1
														for j in range(snake_case							,    snake_case					):
																					snake_case_      =				sum(lower[i][k] * upper[k][j] for k in range(snake_case					)					)
																					snake_case_      =				table[i][j] - total
							return lower, upper


if __name__ == "__main__":
		import doctest

		doctest.testmod()




