


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

if TYPE_CHECKING:
						import pyspark







@dataclass
class        lowercase__      (							datasets.BuilderConfig       ):
			_UpperCAmelCase		:Dict             =			None


def 			_snake_case    (					lowerCamelCase__     :   str    ,						lowerCamelCase__     :   Optional[Any]    ,						)    ->     Any:
							import pyspark

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

							return generate_fn







class        lowercase__      (							_BaseExamplesIterable       ):


			def __init__(				self					:  List[str]       ,	snake_case__					:  "pyspark.sql.DataFrame"       ,	snake_case__					:  int=None       ,	):
										lowerCamelCase_	:							List[Any]									=df
										lowerCamelCase_	:							int									=partition_order or range(self.df.rdd.getNumPartitions() )
										lowerCamelCase_	:							int									=_generate_iterable_examples(self.df       ,	self.partition_order )


			def __iter__(				self					:  Union[str, Any] ):
										yield from self.generate_examples_fn()


			def      UpperCAmelCase__						(				self					:  Dict       ,	snake_case__					:  np.random.Generator ):
										lowerCamelCase_	:							List[str]									=list(range(self.df.rdd.getNumPartitions() ) )
										generator.shuffle(UpperCamelCase__ )
										return SparkExamplesIterable(self.df       ,	partition_order=UpperCamelCase__ )


			def      UpperCAmelCase__						(				self					:  Any       ,	snake_case__					:  int       ,	snake_case__					:  int ):
										lowerCamelCase_	:							int									=self.split_shard_indices_by_worker(UpperCamelCase__       ,	UpperCamelCase__ )
										return SparkExamplesIterable(self.df       ,	partition_order=UpperCamelCase__ )


			@property
			def      UpperCAmelCase__						(				self					:  List[Any] ):
										return len(self.partition_order )







class        lowercase__      (							datasets.DatasetBuilder       ):
			_UpperCAmelCase		:Optional[int]             =			SparkConfig


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

										lowerCamelCase_	:							Optional[int]									=pyspark.sql.SparkSession.builder.getOrCreate()
										lowerCamelCase_	:							Union[str, Any]									=df
										lowerCamelCase_	:							str									=working_dir

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


			def      UpperCAmelCase__						(				self					:  List[Any] ):







										def create_cache_and_write_probe(snake_case__					:  Tuple ):
																	# 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=UpperCamelCase__ )
																	lowerCamelCase_	:							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(UpperCamelCase__       ,	"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:
																	lowerCamelCase_	:							Union[str, Any]									=(
																	    self._spark.sparkContext.parallelize(range(1 )       ,	1 ).mapPartitions(UpperCamelCase__ ).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      UpperCAmelCase__						(				self					:  Optional[int] ):
										return datasets.DatasetInfo(features=self.config.features )


			def      UpperCAmelCase__						(				self					:  Optional[int]       ,	snake_case__					:  datasets.download.download_manager.DownloadManager ):
										return [datasets.SplitGenerator(name=datasets.Split.TRAIN )]


			def      UpperCAmelCase__						(				self					:  str       ,	snake_case__					:  Optional[int] ):
										import pyspark

										def get_arrow_batch_size(snake_case__					:  Union[str, Any] ):
																	for batch in it:
																								yield pa.RecordBatch.from_pydict({"batch_bytes": [batch.nbytes]} )

										lowerCamelCase_	:							Optional[Any]									=self.df.count()
										lowerCamelCase_	:							Dict									=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.
										lowerCamelCase_	:							Optional[Any]									=(
										    self.df.limit(UpperCamelCase__ )
										    .repartition(1 )
										    .mapInArrow(UpperCamelCase__       ,	"batch_bytes: long" )
										    .agg(pyspark.sql.functions.sum("batch_bytes" ).alias("sample_bytes" ) )
										    .collect()[0]
										    .sample_bytes
										    / sample_num_rows
										)
										lowerCamelCase_	:							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.
																	lowerCamelCase_	:							Optional[int]									=min(UpperCamelCase__       ,	int(approx_total_size / max_shard_size ) )
																	lowerCamelCase_	:							Tuple									=self.df.repartition(UpperCamelCase__ )


			def      UpperCAmelCase__						(				self					:  Tuple       ,	snake_case__					:  str       ,	snake_case__					:  str       ,	snake_case__					:  int       ,	):
										import pyspark

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

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

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

										lowerCamelCase_	:							Any									=(
										    self.df.mapInArrow(UpperCamelCase__       ,	"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      UpperCAmelCase__						(				self					:  Optional[Any]       ,	snake_case__					:  "datasets.SplitGenerator"       ,	snake_case__					:  str = "arrow"       ,	snake_case__					:  Optional[Union[str, int]] = None       ,	snake_case__					:  Optional[int] = None       ,	**snake_case__					:  Optional[int]       ,	):
										self._validate_cache_dir()

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

										lowerCamelCase_	:							Tuple									='''-TTTTT-SSSSS-of-NNNNN'''
										lowerCamelCase_	:							List[str]									=F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
										lowerCamelCase_	:							Union[str, Any]									=path_join(self._output_dir       ,	UpperCamelCase__ )

										lowerCamelCase_	:							List[str]									=0
										lowerCamelCase_	:							int									=0
										lowerCamelCase_	:							Dict									=0
										lowerCamelCase_	:							Dict									=[]
										lowerCamelCase_	:							Dict									=[]

										for task_id, content in self._prepare_split_single(UpperCamelCase__       ,	UpperCamelCase__       ,	UpperCamelCase__ ):
																	(
																	    lowerCamelCase_
																	)	:							Union[str, Any]									=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(UpperCamelCase__ )

										lowerCamelCase_	:							Union[str, Any]									=total_num_examples
										lowerCamelCase_	:							List[Any]									=total_num_bytes

										# should rename everything at the end
										logger.debug(F"""Renaming {total_shards} shards.""" )
										if total_shards > 1:
																	lowerCamelCase_	:							List[str]									=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.
																	lowerCamelCase_	:							Union[str, Any]									=self._fs

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

																	lowerCamelCase_	:							Optional[Any]									=[]
																	lowerCamelCase_	:							List[str]									=0
																	for i in range(len(UpperCamelCase__ ) ):
																								lowerCamelCase_	:							str									=task_id_and_num_shards[i]
																								for shard_id in range(UpperCamelCase__ ):
																															args.append([task_id, shard_id, global_shard_id] )
																															global_shard_id += 1
																	self._spark.sparkContext.parallelize(UpperCamelCase__       ,	len(UpperCamelCase__ ) ).map(lambda snake_case__ : _rename_shard(*UpperCamelCase__ ) ).collect()
										else:
																	# don't use any pattern
																	lowerCamelCase_	:							Optional[Any]									=0
																	lowerCamelCase_	:							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(UpperCamelCase__       ,	"" )       ,	)


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











"""simple docstring"""



# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os

from accelerate.test_utils import execute_subprocess_async

def 			_snake_case    (					lowerCamelCase__     :   int=None  )    ->     Union[str, Any]:
							if subparsers is not None:
														lowerCamelCase_	:							List[Any]									=subparsers.add_parser("test"  )
							else:
														lowerCamelCase_	:							List[str]									=argparse.ArgumentParser("Accelerate test command"  )

							parser.add_argument(
							    "--config_file"    ,						default=lowerCamelCase__    ,						help=(
							        "The path to use to store the config file. Will default to a file named default_config.yaml in the cache "
							        "location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have "
							        "such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed "
							        "with 'huggingface'."
							    )    ,						)

							if subparsers is not None:
														parser.set_defaults(func=lowerCamelCase__  )
							return parser

def 			_snake_case    (					lowerCamelCase__     :   List[Any]  )    ->     Any:
							lowerCamelCase_	:							Optional[Any]									=os.path.sep.join(__file__.split(os.path.sep  )[:-2] + ["test_utils", "scripts", "test_script.py"]  )

							if args.config_file is None:
														lowerCamelCase_	:							List[Any]									=script_name
							else:
														lowerCamelCase_	:							Union[str, Any]									=F"""--config_file={args.config_file} {script_name}"""

							lowerCamelCase_	:							List[str]									=["accelerate-launch"] + test_args.split()
							lowerCamelCase_	:							Tuple									=execute_subprocess_async(lowerCamelCase__    ,						env=os.environ.copy()  )
							if result.returncode == 0:
														print("Test is a success! You are ready for your distributed training!"  )

def 			_snake_case    (					)    ->     Tuple:
							lowerCamelCase_	:							Any									=test_command_parser()
							lowerCamelCase_	:							List[Any]									=parser.parse_args()
							test_command(lowerCamelCase__  )


if __name__ == "__main__":
						main()



