

import logging
import os
from dataclasses import dataclass, field
from typing import Dict, Optional

import datasets
import numpy as np
import tensorflow as tf

from transformers import (
    AutoConfig,
    AutoTokenizer,
    EvalPrediction,
    HfArgumentParser,
    PreTrainedTokenizer,
    TFAutoModelForSequenceClassification,
    TFTrainer,
    TFTrainingArguments,
)
from transformers.utils import logging as hf_logging


hf_logging.set_verbosity_info()
hf_logging.enable_default_handler()
hf_logging.enable_explicit_format()


def        __lowercase    (  UpperCAmelCase__		,					UpperCAmelCase__		,					UpperCAmelCase__		,					UpperCAmelCase__		,					UpperCAmelCase__		,					UpperCAmelCase__ = None		,					):

					"""simple docstring"""






					__lowerCAmelCase					=							{}

					if train_file is not None:
										__lowerCAmelCase					=							[train_file]
					if eval_file is not None:
										__lowerCAmelCase					=							[eval_file]
					if test_file is not None:
										__lowerCAmelCase					=							[test_file]

					__lowerCAmelCase					=							datasets.load_dataset('csv'		,					data_files=UpperCAmelCase__      )
					__lowerCAmelCase					=							list(ds[list(files.keys()      )[0]].features.keys()      )
					__lowerCAmelCase					=							features_name.pop(UpperCAmelCase__      )
					__lowerCAmelCase					=							list(set(ds[list(files.keys()      )[0]][label_name]      )      )
					__lowerCAmelCase					=							{label: i for i, label in enumerate(UpperCAmelCase__      )}
					__lowerCAmelCase					=							tokenizer.model_input_names
					__lowerCAmelCase					=							{}

					if len(UpperCAmelCase__      ) == 1:
										for k in files.keys():
															__lowerCAmelCase					=							ds[k].map(
															    lambda UpperCAmelCase__      : tokenizer.batch_encode_plus(
															        example[features_name[0]]		,					truncation=UpperCAmelCase__		,					max_length=UpperCAmelCase__		,					padding='max_length'      )		,					batched=UpperCAmelCase__		,					)
					elif len(UpperCAmelCase__      ) == 2:
										for k in files.keys():
															__lowerCAmelCase					=							ds[k].map(
															    lambda UpperCAmelCase__      : tokenizer.batch_encode_plus(
															        (example[features_name[0]], example[features_name[1]])		,					truncation=UpperCAmelCase__		,					max_length=UpperCAmelCase__		,					padding='max_length'		,					)		,					batched=UpperCAmelCase__		,					)

					def gen_train():
										for ex in transformed_ds[datasets.Split.TRAIN]:
															__lowerCAmelCase					=							{k: v for k, v in ex.items() if k in input_names}
															__lowerCAmelCase					=							labelaid[ex[label_name]]
															yield (d, label)

					def gen_val():
										for ex in transformed_ds[datasets.Split.VALIDATION]:
															__lowerCAmelCase					=							{k: v for k, v in ex.items() if k in input_names}
															__lowerCAmelCase					=							labelaid[ex[label_name]]
															yield (d, label)

					def gen_test():
										for ex in transformed_ds[datasets.Split.TEST]:
															__lowerCAmelCase					=							{k: v for k, v in ex.items() if k in input_names}
															__lowerCAmelCase					=							labelaid[ex[label_name]]
															yield (d, label)

					__lowerCAmelCase					=							(
					    tf.data.Dataset.from_generator(
					        UpperCAmelCase__		,					({k: tf.intaa for k in input_names}, tf.intaa)		,					({k: tf.TensorShape([None]      ) for k in input_names}, tf.TensorShape([]      ))		,					)
					    if datasets.Split.TRAIN in transformed_ds
					    else None
					)

					if train_ds is not None:
										__lowerCAmelCase					=							train_ds.apply(tf.data.experimental.assert_cardinality(len(ds[datasets.Split.TRAIN]      )      )      )

					__lowerCAmelCase					=							(
					    tf.data.Dataset.from_generator(
					        UpperCAmelCase__		,					({k: tf.intaa for k in input_names}, tf.intaa)		,					({k: tf.TensorShape([None]      ) for k in input_names}, tf.TensorShape([]      ))		,					)
					    if datasets.Split.VALIDATION in transformed_ds
					    else None
					)

					if val_ds is not None:
										__lowerCAmelCase					=							val_ds.apply(tf.data.experimental.assert_cardinality(len(ds[datasets.Split.VALIDATION]      )      )      )

					__lowerCAmelCase					=							(
					    tf.data.Dataset.from_generator(
					        UpperCAmelCase__		,					({k: tf.intaa for k in input_names}, tf.intaa)		,					({k: tf.TensorShape([None]      ) for k in input_names}, tf.TensorShape([]      ))		,					)
					    if datasets.Split.TEST in transformed_ds
					    else None
					)

					if test_ds is not None:
										__lowerCAmelCase					=							test_ds.apply(tf.data.experimental.assert_cardinality(len(ds[datasets.Split.TEST]      )      )      )

					return train_ds, val_ds, test_ds, labelaid


lowerCamelCase      				=     logging.getLogger(__name__)





@dataclass
class     snake_case_			:


							"""simple docstring"""





							__UpperCAmelCase       =field(metadata={"""help""": """Which column contains the label"""}							)
							__UpperCAmelCase       =field(default=_a			, metadata={"""help""": """The path of the training file"""}							)
							__UpperCAmelCase       =field(default=_a			, metadata={"""help""": """The path of the development file"""}							)
							__UpperCAmelCase       =field(default=_a			, metadata={"""help""": """The path of the test file"""}							)
							__UpperCAmelCase       =field(
							    default=128			, metadata={
							        """help""": (
							            """The maximum total input sequence length after tokenization. Sequences longer """
							            """than this will be truncated, sequences shorter will be padded."""
							        )
							    }			, )
							__UpperCAmelCase       =field(
							    default=_a			, metadata={"""help""": """Overwrite the cached training and evaluation sets"""}							)





@dataclass
class     snake_case_			:


							"""simple docstring"""





							__UpperCAmelCase       =field(
							    metadata={"""help""": """Path to pretrained model or model identifier from huggingface.co/models"""}							)
							__UpperCAmelCase       =field(
							    default=_a			, metadata={"""help""": """Pretrained config name or path if not the same as model_name"""}							)
							__UpperCAmelCase       =field(
							    default=_a			, metadata={"""help""": """Pretrained tokenizer name or path if not the same as model_name"""}							)
							__UpperCAmelCase       =field(default=_a			, metadata={"""help""": """Set this flag to use fast tokenization."""}							)
							# If you want to tweak more attributes on your tokenizer, you should do it in a distinct script,
							# or just modify its tokenizer_config.json.
							__UpperCAmelCase       =field(
							    default=_a			, metadata={"""help""": """Where do you want to store the pretrained models downloaded from huggingface.co"""}			, )


def        __lowercase    (  ):

					"""simple docstring"""






					__lowerCAmelCase					=							HfArgumentParser((ModelArguments, DataTrainingArguments, TFTrainingArguments)      )
					__lowerCAmelCase,	__lowerCAmelCase,	__lowerCAmelCase					=							parser.parse_args_into_dataclasses()

					if (
					    os.path.exists(training_args.output_dir      )
					    and os.listdir(training_args.output_dir      )
					    and training_args.do_train
					    and not training_args.overwrite_output_dir
					):
										raise ValueError(
										    F"""Output directory ({training_args.output_dir}) already exists and is not empty. Use"""
										    ' --overwrite_output_dir to overcome.'      )

					# Setup logging
					logging.basicConfig(
					    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'		,					datefmt='%m/%d/%Y %H:%M:%S'		,					level=logging.INFO		,					)
					logger.info(
					    F"""n_replicas: {training_args.n_replicas}, distributed training: {bool(training_args.n_replicas > 1      )}, """
					    F"""16-bits training: {training_args.fpaa}"""      )
					logger.info(F"""Training/evaluation parameters {training_args}"""      )

					# Load pretrained model and tokenizer
					#
					# Distributed training:
					# The .from_pretrained methods guarantee that only one local process can concurrently
					# download model & vocab.

					__lowerCAmelCase					=							AutoTokenizer.from_pretrained(
					    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path		,					cache_dir=model_args.cache_dir		,					)

					__lowerCAmelCase,	__lowerCAmelCase,	__lowerCAmelCase,	__lowerCAmelCase					=							get_tfds(
					    train_file=data_args.train_file		,					eval_file=data_args.dev_file		,					test_file=data_args.test_file		,					tokenizer=UpperCAmelCase__		,					label_column_id=data_args.label_column_id		,					max_seq_length=data_args.max_seq_length		,					)

					__lowerCAmelCase					=							AutoConfig.from_pretrained(
					    model_args.config_name if model_args.config_name else model_args.model_name_or_path		,					num_labels=len(UpperCAmelCase__      )		,					labelaid=UpperCAmelCase__		,					idalabel={id: label for label, id in labelaid.items()}		,					finetuning_task='text-classification'		,					cache_dir=model_args.cache_dir		,					)

					with training_args.strategy.scope():
										__lowerCAmelCase					=							TFAutoModelForSequenceClassification.from_pretrained(
										    model_args.model_name_or_path		,					from_pt=bool('.bin' in model_args.model_name_or_path      )		,					config=UpperCAmelCase__		,					cache_dir=model_args.cache_dir		,					)

					def compute_metrics(UpperCAmelCase__      ) -> Dict:
										__lowerCAmelCase					=							np.argmax(p.predictions		,					axis=1      )

										return {"acc": (preds == p.label_ids).mean()}

					# Initialize our Trainer
					__lowerCAmelCase					=							TFTrainer(
					    model=UpperCAmelCase__		,					args=UpperCAmelCase__		,					train_dataset=UpperCAmelCase__		,					eval_dataset=UpperCAmelCase__		,					compute_metrics=UpperCAmelCase__		,					)

					# Training
					if training_args.do_train:
										trainer.train()
										trainer.save_model()
										tokenizer.save_pretrained(training_args.output_dir      )

					# Evaluation
					__lowerCAmelCase					=							{}
					if training_args.do_eval:
										logger.info('*** Evaluate ***'      )
										__lowerCAmelCase					=							trainer.evaluate()
										__lowerCAmelCase					=							os.path.join(training_args.output_dir		,					'eval_results.txt'      )

										with open(UpperCAmelCase__		,					'w'      ) as writer:
															logger.info('***** Eval results *****'      )

															for key, value in result.items():
																				logger.info(F"""  {key} = {value}"""      )
																				writer.write(F"""{key} = {value}\n"""      )

															results.update(UpperCAmelCase__      )

					return results


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


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

if TYPE_CHECKING:
		import pyspark





@dataclass
class     snake_case_			(				datasets.BuilderConfig							):


							"""simple docstring"""





							__UpperCAmelCase       =None


def        __lowercase    (  UpperCAmelCase__		,					UpperCAmelCase__		,					):

					"""simple docstring"""






					import pyspark

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

					return generate_fn





class     snake_case_			(				_BaseExamplesIterable							):


							"""simple docstring"""





							def __init__( self					,						_A					,						_A=None					,						):
												__lowerCAmelCase					=							df
												__lowerCAmelCase					=							partition_order or range(self.df.rdd.getNumPartitions()							)
												__lowerCAmelCase					=							_generate_iterable_examples(self.df					,						self.partition_order							)




							def __iter__( self							):
												yield from self.generate_examples_fn()




							def        A__  ( self					,						_A							):
												__lowerCAmelCase					=							list(range(self.df.rdd.getNumPartitions()							)							)
												generator.shuffle(_A							)
												return SparkExamplesIterable(self.df					,						partition_order=_A							)




							def        A__  ( self					,						_A					,						_A							):
												__lowerCAmelCase					=							self.split_shard_indices_by_worker(_A					,						_A							)
												return SparkExamplesIterable(self.df					,						partition_order=_A							)




							@property
							def        A__  ( self							):
												return len(self.partition_order							)





class     snake_case_			(				datasets.DatasetBuilder							):


							"""simple docstring"""





							__UpperCAmelCase       =SparkConfig




							def __init__( self					,						_A					,						_A = None					,						_A = None					,						**_A					,						):
												import pyspark

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

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




							def        A__  ( self							):
												# Returns the path of the created file.
												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							)
																	__lowerCAmelCase					=							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:
																	__lowerCAmelCase					=							(
																	    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        A__  ( self							):
												return datasets.DatasetInfo(features=self.config.features							)




							def        A__  ( self					,						_A							):
												return [datasets.SplitGenerator(name=datasets.Split.TRAIN							)]




							def        A__  ( self					,						_A							):
												import pyspark

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

												__lowerCAmelCase					=							self.df.count()
												__lowerCAmelCase					=							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.
												__lowerCAmelCase					=							(
												    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
												)
												__lowerCAmelCase					=							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					=							min(_A					,						int(approx_total_size / max_shard_size							)							)
																	__lowerCAmelCase					=							self.df.repartition(_A							)




							def        A__  ( self					,						_A					,						_A					,						_A					,						):
												import pyspark

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

												def write_arrow(_A							):
																	# Within the same SparkContext, no two task attempts will share the same attempt ID.
																	__lowerCAmelCase					=							pyspark.TaskContext().taskAttemptId()
																	__lowerCAmelCase					=							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']					,						)
																	__lowerCAmelCase					=							0
																	__lowerCAmelCase					=							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					,						)
																	__lowerCAmelCase					=							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:
																											__lowerCAmelCase,	__lowerCAmelCase					=							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					=							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					,						)
																						__lowerCAmelCase					=							pa.Table.from_batches([batch]							)
																						writer.write_table(_A							)

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

												__lowerCAmelCase					=							(
												    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        A__  ( self					,						_A					,						_A = "arrow"					,						_A = None					,						_A = None					,						**_A					,						):
												self._validate_cache_dir()

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

												__lowerCAmelCase					=							'-TTTTT-SSSSS-of-NNNNN'
												__lowerCAmelCase					=							F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
												__lowerCAmelCase					=							path_join(self._output_dir					,						_A							)

												__lowerCAmelCase					=							0
												__lowerCAmelCase					=							0
												__lowerCAmelCase					=							0
												__lowerCAmelCase					=							[]
												__lowerCAmelCase					=							[]

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

												__lowerCAmelCase					=							total_num_examples
												__lowerCAmelCase					=							total_num_bytes

												# should rename everything at the end
												logger.debug(F"""Renaming {total_shards} shards."""							)
												if total_shards > 1:
																	__lowerCAmelCase					=							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					=							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}"""							)					,						)

																	__lowerCAmelCase					=							[]
																	__lowerCAmelCase					=							0
																	for i in range(len(_A							)							):
																						__lowerCAmelCase,	__lowerCAmelCase					=							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
																	__lowerCAmelCase					=							0
																	__lowerCAmelCase					=							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        A__  ( self					,						_A					,						):
												return SparkExamplesIterable(self.df							)






