
# 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 warnings
from typing import List
from unittest.mock import Mock

import torch
from torch.utils.data import DataLoader, IterableDataset, TensorDataset

from accelerate.accelerator import Accelerator
from accelerate.utils.dataclasses import DistributedType





class 				A__						(      _SCREAMING_SNAKE_CASE):
      def __init__( self       ,    __magic_name__ ):
       lowerCamelCase						:      int							     = data






      def __iter__( self ):
       for element in self.data:
        yield element





def        _a   (		lowerCamelCase=True				):
 lowerCamelCase						:      Optional[Any]							     = Accelerator(even_batches=lowerCamelCase_				)
 assert accelerator.num_processes == 2, "this script expects that two GPUs are available"
 return accelerator





def        _a   (		lowerCamelCase,      lowerCamelCase,      lowerCamelCase,      lowerCamelCase = False				):
 if iterable:
  lowerCamelCase						:      List[str]							     = DummyIterableDataset(torch.as_tensor(range(lowerCamelCase_				)				)				)
 else:
  lowerCamelCase						:      Optional[int]							     = TensorDataset(torch.as_tensor(range(lowerCamelCase_				)				)				)

 lowerCamelCase						:      Optional[Any]							     = DataLoader(lowerCamelCase_,      batch_size=lowerCamelCase_				)
 lowerCamelCase						:      Dict							     = accelerator.prepare(lowerCamelCase_				)

 return dl





def        _a   (		lowerCamelCase,      lowerCamelCase,      lowerCamelCase,      lowerCamelCase,      lowerCamelCase,      ):
 lowerCamelCase						:      List[Any]							     = create_dataloader(accelerator=lowerCamelCase_,      dataset_size=lowerCamelCase_,      batch_size=lowerCamelCase_				)

 lowerCamelCase						:      List[str]							     = [len(batch[0]				) for batch in dl]

 if accelerator.process_index == 0:
  assert batch_sizes == process_0_expected_batch_sizes
 elif accelerator.process_index == 1:
  assert batch_sizes == process_1_expected_batch_sizes





def        _a   (		):
 lowerCamelCase						:      List[Any]							     = create_accelerator()

 # without padding, we would expect a different number of batches
 verify_dataloader_batch_sizes(
     lowerCamelCase_,      dataset_size=3,      batch_size=1,      process_0_expected_batch_sizes=[1, 1],      process_1_expected_batch_sizes=[1, 1],      )

 # without padding, we would expect the same number of batches, but different sizes
 verify_dataloader_batch_sizes(
     lowerCamelCase_,      dataset_size=7,      batch_size=2,      process_0_expected_batch_sizes=[2, 2],      process_1_expected_batch_sizes=[2, 2],      )





def        _a   (		):
 lowerCamelCase						:      str							     = create_accelerator(even_batches=lowerCamelCase_				)

 verify_dataloader_batch_sizes(
     lowerCamelCase_,      dataset_size=3,      batch_size=1,      process_0_expected_batch_sizes=[1, 1],      process_1_expected_batch_sizes=[1],      )

 verify_dataloader_batch_sizes(
     lowerCamelCase_,      dataset_size=7,      batch_size=2,      process_0_expected_batch_sizes=[2, 2],      process_1_expected_batch_sizes=[2, 1],      )





def        _a   (		):
 lowerCamelCase						:      Union[str, Any]							     = create_accelerator(even_batches=lowerCamelCase_				)

 lowerCamelCase						:      Optional[int]							     = torch.nn.Linear(1,      1				)
 lowerCamelCase						:      Dict							     = accelerator.prepare(lowerCamelCase_				)

 lowerCamelCase						:      Tuple							     = create_dataloader(lowerCamelCase_,      dataset_size=3,      batch_size=1				)

 lowerCamelCase						:      Optional[int]							     = []
 with accelerator.join_uneven_inputs([ddp_model]				):
  for batch_idx, batch in enumerate(lowerCamelCase_				):
   lowerCamelCase						:      Any							     = ddp_model(batch[0].float()				)
   lowerCamelCase						:      Optional[Any]							     = output.sum()
   loss.backward()
   batch_idxs.append(lowerCamelCase_				)

 accelerator.wait_for_everyone()

 if accelerator.process_index == 0:
  assert batch_idxs == [0, 1]
 elif accelerator.process_index == 1:
  assert batch_idxs == [0]





def        _a   (		lowerCamelCase				):
 with warnings.catch_warnings(record=lowerCamelCase_				) as w:
  with accelerator.join_uneven_inputs([Mock()]				):
   pass

  assert issubclass(w[-1].category,      lowerCamelCase_				)
  assert "only supported for multi-GPU" in str(w[-1].message				)





def        _a   (		):
 lowerCamelCase						:      Any							     = True
 lowerCamelCase						:      Union[str, Any]							     = False
 lowerCamelCase						:      str							     = create_accelerator(even_batches=lowerCamelCase_				)
 lowerCamelCase						:      Optional[int]							     = torch.nn.Linear(1,      1				)
 lowerCamelCase						:      Optional[Any]							     = accelerator.prepare(lowerCamelCase_				)
 lowerCamelCase						:      List[str]							     = create_dataloader(lowerCamelCase_,      dataset_size=3,      batch_size=1				)
 lowerCamelCase						:      int							     = create_dataloader(lowerCamelCase_,      dataset_size=3,      batch_size=1				)

 with accelerator.join_uneven_inputs([ddp_model],      even_batches=lowerCamelCase_				):
  lowerCamelCase						:      List[Any]							     = train_dl.batch_sampler.even_batches
  lowerCamelCase						:      Tuple							     = valid_dl.batch_sampler.even_batches

 assert train_dl_overridden_value == overridden_even_batches
 assert valid_dl_overridden_value == overridden_even_batches
 assert train_dl.batch_sampler.even_batches == default_even_batches
 assert valid_dl.batch_sampler.even_batches == default_even_batches





def        _a   (		):
 lowerCamelCase						:      Tuple							     = True
 lowerCamelCase						:      Tuple							     = False
 lowerCamelCase						:      List[Any]							     = create_accelerator(even_batches=lowerCamelCase_				)
 lowerCamelCase						:      Tuple							     = torch.nn.Linear(1,      1				)
 lowerCamelCase						:      Optional[Any]							     = accelerator.prepare(lowerCamelCase_				)
 create_dataloader(lowerCamelCase_,      dataset_size=3,      batch_size=1,      iterable=lowerCamelCase_				)
 lowerCamelCase						:      List[str]							     = create_dataloader(lowerCamelCase_,      dataset_size=3,      batch_size=1				)

 with warnings.catch_warnings():
  warnings.filterwarnings("""ignore"""				)
  try:
   with accelerator.join_uneven_inputs([ddp_model],      even_batches=lowerCamelCase_				):
    lowerCamelCase						:      Tuple							     = batch_dl.batch_sampler.even_batches
  except AttributeError:
   # ensure attribute error is not raised when processing iterable dl
   raise AssertionError

 assert batch_dl_overridden_value == overridden_even_batches
 assert batch_dl.batch_sampler.even_batches == default_even_batches





def        _a   (		):
 lowerCamelCase						:      Union[str, Any]							     = create_accelerator()
 lowerCamelCase						:      List[Any]							     = torch.nn.Linear(1,      1				)
 lowerCamelCase						:      Union[str, Any]							     = accelerator.prepare(lowerCamelCase_				)
 create_dataloader(lowerCamelCase_,      dataset_size=3,      batch_size=1,      iterable=lowerCamelCase_				)

 with warnings.catch_warnings(record=lowerCamelCase_				) as w:
  with accelerator.join_uneven_inputs([ddp_model],      even_batches=lowerCamelCase_				):
   pass

  assert issubclass(w[-1].category,      lowerCamelCase_				)
  assert "only supported for map-style datasets" in str(w[-1].message				)





def        _a   (		):
 lowerCamelCase						:      Union[str, Any]							     = create_accelerator()

 accelerator.print("""Test that even_batches variable ensures uniform batches across processes"""				)
 test_default_ensures_even_batch_sizes()

 accelerator.print("""Run tests with even_batches disabled"""				)
 test_can_disable_even_batches()

 accelerator.print("""Test joining uneven inputs"""				)
 test_can_join_uneven_inputs()

 accelerator.print("""Test overriding even_batches when joining uneven inputs"""				)
 test_join_can_override_even_batches()

 accelerator.print("""Test overriding even_batches for mixed dataloader types"""				)
 test_join_can_override_for_mixed_type_dataloaders()

 accelerator.print("""Test overriding even_batches raises a warning for iterable dataloaders"""				)
 test_join_raises_warning_for_iterable_when_overriding_even_batches()

 accelerator.print("""Test join with non DDP distributed raises warning"""				)
 lowerCamelCase						:      List[str]							     = accelerator.state.distributed_type
 lowerCamelCase						:      str							     = DistributedType.FSDP
 test_join_raises_warning_for_non_ddp_distributed(lowerCamelCase_				)
 lowerCamelCase						:      List[Any]							     = original_state


if __name__ == "__main__":
      main()







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


lowercase			:				Dict   =     datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
		import pyspark







@dataclass
class 					__lowercase  (						datasets.BuilderConfig				):





							"""simple docstring"""

							UpperCAmelCase_						:		Optional[datasets.Features]								=	None






def        snake_case__		(    lowerCamelCase_     ,				lowerCamelCase_     ,				):
							import pyspark

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




							def __iter__(					self						)		->       str:
														yield from self.generate_examples_fn()




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       "SparkExamplesIterable":
														A      :					Tuple      		=					list(range(self.df.rdd.getNumPartitions()						)						)
														generator.shuffle(__UpperCAmelCase						)
														return SparkExamplesIterable(self.df    ,						partition_order=__UpperCAmelCase						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase						)		->       "SparkExamplesIterable":
														A      :					Tuple      		=					self.split_shard_indices_by_worker(__UpperCAmelCase    ,						__UpperCAmelCase						)
														return SparkExamplesIterable(self.df    ,						partition_order=__UpperCAmelCase						)






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







class 					__lowercase  (						datasets.DatasetBuilder				):





							"""simple docstring"""

							UpperCAmelCase_						:		List[str]								=	SparkConfig




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

														A      :					Tuple      		=					pyspark.sql.SparkSession.builder.getOrCreate()
														A      :					List[Any]      		=					df
														A      :					List[Any]      		=					working_dir

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




							def       snake_case			(					self						)		->       Optional[Any]:
														# Returns the path of the created file.
														def create_cache_and_write_probe(__UpperCAmelCase						):
																					# 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						)
																					A      :					Optional[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:
																					A      :					Optional[int]      		=					(
																					    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       snake_case			(					self						)		->       List[Any]:
														return datasets.DatasetInfo(features=self.config.features						)




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       str:
														return [datasets.SplitGenerator(name=datasets.Split.TRAIN						)]




							def       snake_case			(					self    ,						__UpperCAmelCase						)		->       List[Any]:
														import pyspark

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

														A      :					Any      		=					self.df.count()
														A      :					Tuple      		=					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.
														A      :					Dict      		=					(
														    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
														)
														A      :					str      		=					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(__UpperCAmelCase    ,						int(approx_total_size / max_shard_size						)						)
																					A      :					List[Any]      		=					self.df.repartition(__UpperCAmelCase						)




							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase    ,						)		->       Iterable[Tuple[int, bool, Union[int, tuple]]]:
														import pyspark

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

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

																					if writer._num_bytes > 0:
																												A		,		A      :					Tuple      		=					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						)						):
																																			A      :					Union[str, Any]      		=					os.path.join(os.path.dirname(__UpperCAmelCase						)    ,						os.path.basename(__UpperCAmelCase						)						)
																																			shutil.move(__UpperCAmelCase    ,						__UpperCAmelCase						)

														A      :					Union[str, 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       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase = "arrow"    ,						__UpperCAmelCase = None    ,						__UpperCAmelCase = None    ,						**__UpperCAmelCase    ,						)		->       List[str]:
														self._validate_cache_dir()

														A      :					int      		=					convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE						)
														self._repartition_df_if_needed(__UpperCAmelCase						)
														A      :					Any      		=					not is_remote_filesystem(self._fs						)
														A      :					Union[str, Any]      		=					os.path.join if is_local else posixpath.join

														A      :					Any      		=					'''-TTTTT-SSSSS-of-NNNNN'''
														A      :					Any      		=					f'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
														A      :					List[Any]      		=					path_join(self._output_dir    ,						__UpperCAmelCase						)

														A      :					Union[str, Any]      		=					0
														A      :					Any      		=					0
														A      :					Tuple      		=					0
														A      :					Union[str, Any]      		=					[]
														A      :					List[str]      		=					[]

														for task_id, content in self._prepare_split_single(__UpperCAmelCase    ,						__UpperCAmelCase    ,						__UpperCAmelCase						):
																					(
																					    (
																					    A
																					)		,		(
																					    A
																					)		,		(
																					    A
																					)		,		(
																					    A
																					)		,		
																					)      :					List[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						)

														A      :					List[str]      		=					total_num_examples
														A      :					List[str]      		=					total_num_bytes

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

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

																					A      :					Tuple      		=					[]
																					A      :					Optional[int]      		=					0
																					for i in range(len(__UpperCAmelCase						)						):
																												A		,		A      :					Tuple      		=					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 __UpperCAmelCase						: _rename_shard(*__UpperCAmelCase						)						).collect()
														else:
																					# don't use any pattern
																					A      :					int      		=					0
																					A      :					Tuple      		=					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       snake_case			(					self    ,						__UpperCAmelCase    ,						)		->       SparkExamplesIterable:
														return SparkExamplesIterable(self.df						)






