



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


UpperCamelCase						:				int             =      datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
	import pyspark



@dataclass
class 			lowerCamelCase__						(   datasets.BuilderConfig			):
						lowerCAmelCase       =     None







def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			)      ->		str:







		"""simple docstring"""





		import pyspark

		def generate_fn():
				A					       = df.select('*' ,			pyspark.sql.functions.spark_partition_id().alias('part_id'	)	)
				for partition_id in partition_order:
						A					       = df_with_partition_id.select('*'	).where(f'part_id = {partition_id}'	).drop('part_id'	)
						A					       = partition_df.collect()
						A					       = 0
						for row in rows:
								yield f'{partition_id}_{row_id}', row.asDict()
								row_id += 1

		return generate_fn



class 			lowerCamelCase__						(   _BaseExamplesIterable			):







						def __init__(							self   :    Tuple			,			_lowercase   :    "pyspark.sql.DataFrame"			,			_lowercase   :    List[Any]=None			,			):
								A					       = df
								A					       = partition_order or range(self.df.rdd.getNumPartitions()		)
								A					       = _generate_iterable_examples(self.df			,			self.partition_order		)







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







						def __a						(							self   :    Dict			,			_lowercase   :    np.random.Generator		):
								A					       = list(range(self.df.rdd.getNumPartitions()		)		)
								generator.shuffle(lowerCamelCase__		)
								return SparkExamplesIterable(self.df			,			partition_order=lowerCamelCase__		)







						def __a						(							self   :    Optional[Any]			,			_lowercase   :    int			,			_lowercase   :    int		):
								A					       = self.split_shard_indices_by_worker(lowerCamelCase__			,			lowerCamelCase__		)
								return SparkExamplesIterable(self.df			,			partition_order=lowerCamelCase__		)







						@property
						def __a						(							self   :    List[str]		):
								return len(self.partition_order		)



class 			lowerCamelCase__						(   datasets.DatasetBuilder			):
						lowerCAmelCase       =     SparkConfig







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

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

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







						def __a						(							self   :    str		):






								def create_cache_and_write_probe(_lowercase   :    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=lowerCamelCase__		)
										A					       = 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(lowerCamelCase__			,			'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					       = (
										    self._spark.sparkContext.parallelize(range(1		)			,			1		).mapPartitions(lowerCamelCase__		).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 __a						(							self   :    int		):
								return datasets.DatasetInfo(features=self.config.features		)







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







						def __a						(							self   :    Any			,			_lowercase   :    Union[str, Any]		):
								import pyspark

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

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







						def __a						(							self   :    str			,			_lowercase   :    str			,			_lowercase   :    str			,			_lowercase   :    int			,			):
								import pyspark

								A					       = ParquetWriter if file_format == '''parquet''' else ArrowWriter
								A					       = os.path.join(self._working_dir			,			os.path.basename(lowerCamelCase__		)		) if self._working_dir else fpath
								A					       = 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					       = self.config.features
								A					       = self._writer_batch_size
								A					       = self._fs.storage_options

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

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

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

								A					       = convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE		)
								self._repartition_df_if_needed(lowerCamelCase__		)
								A					       = not is_remote_filesystem(self._fs		)
								A					       = os.path.join if is_local else posixpath.join

								A					       = '''-TTTTT-SSSSS-of-NNNNN'''
								A					       = f'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
								A					       = path_join(self._output_dir			,			lowerCamelCase__		)

								A					       = 0
								A					       = 0
								A					       = 0
								A					       = []
								A					       = []

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

								A					       = total_num_examples
								A					       = total_num_bytes

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

										# use the -SSSSS-of-NNNNN pattern
										def _rename_shard(
										    _lowercase   :    int			,			_lowercase   :    int			,			_lowercase   :    int			,			):
												rename(
												    lowerCamelCase__			,			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					       = []
										A					       = 0
										for i in range(len(lowerCamelCase__		)		):
												A					       = task_id_and_num_shards[i]
												for shard_id in range(lowerCamelCase__		):
														args.append([task_id, shard_id, global_shard_id]		)
														global_shard_id += 1
										self._spark.sparkContext.parallelize(lowerCamelCase__			,			len(lowerCamelCase__		)		).map(lambda _lowercase		: _rename_shard(*lowerCamelCase__		)		).collect()
								else:
										# don't use any pattern
										A					       = 0
										A					       = task_id_and_num_shards[0][0]
										self._rename(
										    fpath.replace('SSSSS'			,			f'{shard_id:05d}'		).replace('TTTTT'			,			f'{task_id:05d}'		)			,			fpath.replace(lowerCamelCase__			,			''		)			,			)







						def __a						(							self   :    Dict			,			_lowercase   :    "datasets.SplitGenerator"			,			):
								return SparkExamplesIterable(self.df		)









"""simple docstring"""


import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
	from sagemaker import Session, TrainingJobAnalytics
	from sagemaker.huggingface import HuggingFace



@pytest.mark.skipif(
    literal_eval(os.getenv("""TEST_SAGEMAKER"""							,    """False"""			)			) is not True							,    reason="""Skipping test because should only be run when releasing minor transformers version"""							,    )
@pytest.mark.usefixtures("""sm_env"""			)
@parameterized_class(
    [
        {
            """framework""": """pytorch""",
            """script""": """run_glue.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 650, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_ddp.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf_dist.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ]			)
class 			lowerCamelCase__						(   unittest.TestCase			):







						def __a						(							self   :    int		):
								if self.framework == "pytorch":
										subprocess.run(
										    f'cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py'.split()			,			encoding='utf-8'			,			check=_lowercase			,			)
								assert hasattr(self			,			'env'		)







						def __a						(							self   :    Optional[int]			,			_lowercase   :    int		):
								A					       = f'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
								# distributed data settings
								A					       = {'smdistributed': {'dataparallel': {'enabled': True}}} if self.script != 'run_ddp.py' else None

								# creates estimator
								return HuggingFace(
								    entry_point=self.script			,			source_dir=self.env.test_path			,			role=self.env.role			,			image_uri=self.env.image_uri			,			base_job_name=_lowercase			,			instance_count=_lowercase			,			instance_type=self.instance_type			,			debugger_hook_config=_lowercase			,			hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}			,			metric_definitions=self.env.metric_definitions			,			distribution=_lowercase			,			py_version='py36'			,			)







						def __a						(							self   :    Tuple			,			_lowercase   :    List[str]		):
								TrainingJobAnalytics(_lowercase		).export_csv(f'{self.env.test_path}/{job_name}_metrics.csv'		)







						@parameterized.expand([(2,)]		)
						def __a						(							self   :    List[Any]			,			_lowercase   :    Union[str, Any]		):
								# create estimator
								A					       = self.create_estimator(_lowercase		)

								# run training
								estimator.fit()

								# result dataframe
								A					       = TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

								# extract kpis
								A					       = list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']		)
								A					       = list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']		)
								# get train time from SageMaker job, this includes starting, preprocessing, stopping
								A					       = (
								    Session().describe_training_job(estimator.latest_training_job.name		).get('TrainingTimeInSeconds'			,			999_999		)
								)

								# assert kpis
								assert train_runtime <= self.results["train_runtime"]
								assert all(t >= self.results['eval_accuracy'] for t in eval_accuracy		)
								assert all(t <= self.results['eval_loss'] for t in eval_loss		)

								# dump tests result into json file to share in PR
								with open(f'{estimator.latest_training_job.name}.json'			,			'w'		) as outfile:
										json.dump({'train_time': train_runtime, 'eval_accuracy': eval_accuracy, 'eval_loss': eval_loss}			,			_lowercase		)





