





'''simple docstring'''






from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)
def    UpperCamelCase__      (		lowerCAmelCase							,	lowerCAmelCase  ):





      """simple docstring"""


      _lowerCAmelCase							     =							[]
      for part_id in partition_order:
            _lowerCAmelCase							     =							df.where(f"SPARK_PARTITION_ID() = {part_id}"  ).collect()
            for row_idx, row in enumerate(_a  ):
                  expected_row_ids_and_row_dicts.append((f"{part_id}_{row_idx}", row.asDict())  )
      return expected_row_ids_and_row_dicts
@require_not_windows
@require_dill_gt_0_3_2
def    UpperCamelCase__      (		):





      """simple docstring"""


      _lowerCAmelCase							     =							pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
      _lowerCAmelCase							     =							spark.range(1_00  ).repartition(1  )
      _lowerCAmelCase							     =							Spark(_a  )
      # The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
      # that each partition can hold 2 rows.
      spark_builder._repartition_df_if_needed(max_shard_size=16  )
      # Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
      assert spark_builder.df.rdd.getNumPartitions() == 50
@require_not_windows
@require_dill_gt_0_3_2
def    UpperCamelCase__      (		):





      """simple docstring"""


      _lowerCAmelCase							     =							pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
      _lowerCAmelCase							     =							spark.range(10  ).repartition(2  )
      _lowerCAmelCase							     =							[1, 0]
      _lowerCAmelCase							     =							_generate_iterable_examples(_a							,	_a  )  # Reverse the partitions.
      _lowerCAmelCase							     =							_get_expected_row_ids_and_row_dicts_for_partition_order(_a							,	_a  )

      for i, (row_id, row_dict) in enumerate(generate_fn()  ):
            _lowerCAmelCase							     =							expected_row_ids_and_row_dicts[i]
            assert row_id == expected_row_id
            assert row_dict == expected_row_dict
@require_not_windows
@require_dill_gt_0_3_2
def    UpperCamelCase__      (		):





      """simple docstring"""


      _lowerCAmelCase							     =							pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
      _lowerCAmelCase							     =							spark.range(10  ).repartition(1  )
      _lowerCAmelCase							     =							SparkExamplesIterable(_a  )
      assert it.n_shards == 1
      for i, (row_id, row_dict) in enumerate(_a  ):
            assert row_id == f"0_{i}"
            assert row_dict == {"id": i}
@require_not_windows
@require_dill_gt_0_3_2
def    UpperCamelCase__      (		):





      """simple docstring"""


      _lowerCAmelCase							     =							pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
      _lowerCAmelCase							     =							spark.range(30  ).repartition(3  )
      # Mock the generator so that shuffle reverses the partition indices.
      with patch("""numpy.random.Generator"""  ) as generator_mock:
            _lowerCAmelCase							     =							lambda lowerCAmelCase  : x.reverse()
            _lowerCAmelCase							     =							_get_expected_row_ids_and_row_dicts_for_partition_order(_a							,	[2, 1, 0]  )

            _lowerCAmelCase							     =							SparkExamplesIterable(_a  ).shuffle_data_sources(_a  )
            assert shuffled_it.n_shards == 3
            for i, (row_id, row_dict) in enumerate(_a  ):
                  _lowerCAmelCase							     =							expected_row_ids_and_row_dicts[i]
                  assert row_id == expected_row_id
                  assert row_dict == expected_row_dict
@require_not_windows
@require_dill_gt_0_3_2
def    UpperCamelCase__      (		):





      """simple docstring"""


      _lowerCAmelCase							     =							pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
      _lowerCAmelCase							     =							spark.range(20  ).repartition(4  )

      # Partitions 0 and 2
      _lowerCAmelCase							     =							SparkExamplesIterable(_a  ).shard_data_sources(worker_id=0							,	num_workers=2  )
      assert shard_it_a.n_shards == 2
      _lowerCAmelCase							     =							_get_expected_row_ids_and_row_dicts_for_partition_order(_a							,	[0, 2]  )
      for i, (row_id, row_dict) in enumerate(_a  ):
            _lowerCAmelCase							     =							expected_row_ids_and_row_dicts_a[i]
            assert row_id == expected_row_id
            assert row_dict == expected_row_dict

      # Partitions 1 and 3
      _lowerCAmelCase							     =							SparkExamplesIterable(_a  ).shard_data_sources(worker_id=1							,	num_workers=2  )
      assert shard_it_a.n_shards == 2
      _lowerCAmelCase							     =							_get_expected_row_ids_and_row_dicts_for_partition_order(_a							,	[1, 3]  )
      for i, (row_id, row_dict) in enumerate(_a  ):
            _lowerCAmelCase							     =							expected_row_ids_and_row_dicts_a[i]
            assert row_id == expected_row_id
            assert row_dict == expected_row_dict





@require_not_windows
@require_dill_gt_0_3_2
def    UpperCamelCase__      (		):





      """simple docstring"""


      _lowerCAmelCase							     =							pyspark.sql.SparkSession.builder.master("""local[*]"""  ).appName("""pyspark"""  ).getOrCreate()
      _lowerCAmelCase							     =							spark.range(1_00  ).repartition(1  )
      _lowerCAmelCase							     =							Spark(_a  )
      # Choose a small max_shard_size for maximum partitioning.
      spark_builder._repartition_df_if_needed(max_shard_size=1  )
      # The new number of partitions should not be greater than the number of rows.
      assert spark_builder.df.rdd.getNumPartitions() == 1_00











"""simple docstring"""

import shutil
import tempfile
import unittest
from unittest.mock import patch

from transformers import (
    DefaultFlowCallback,
    IntervalStrategy,
    PrinterCallback,
    ProgressCallback,
    Trainer,
    TrainerCallback,
    TrainingArguments,
    is_torch_available,
)
from transformers.testing_utils import require_torch


if is_torch_available():
     from transformers.trainer import DEFAULT_CALLBACKS

     from .test_trainer import RegressionDataset, RegressionModelConfig, RegressionPreTrainedModel




class        _UpperCAmelCase       (			lowerCAmelCase__):

 def __init__(     self		:       Optional[int]				):
     snake_case_ :     str				=	[]

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       Any  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       Tuple				):
     self.events.append('''on_init_end'''				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       str  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_train_begin'''				)

 def 						_snake_case					(     self		:       Any  ,     lowercase_		:       List[str]  ,     lowercase_		:       Tuple  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       Optional[int]				):
     self.events.append('''on_train_end'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       int  ,     lowercase_		:       Optional[Any]  ,     **lowercase_		:       List[Any]				):
     self.events.append('''on_epoch_begin'''				)

 def 						_snake_case					(     self		:       Tuple  ,     lowercase_		:       List[str]  ,     lowercase_		:       Dict  ,     lowercase_		:       Union[str, Any]  ,     **lowercase_		:       Optional[Any]				):
     self.events.append('''on_epoch_end'''				)

 def 						_snake_case					(     self		:       List[str]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       Optional[Any]  ,     lowercase_		:       int  ,     **lowercase_		:       Optional[Any]				):
     self.events.append('''on_step_begin'''				)

 def 						_snake_case					(     self		:       int  ,     lowercase_		:       int  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_step_end'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       int  ,     lowercase_		:       Dict  ,     lowercase_		:       List[str]  ,     **lowercase_		:       List[str]				):
     self.events.append('''on_evaluate'''				)

 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       Any  ,     lowercase_		:       List[Any]  ,     **lowercase_		:       str				):
     self.events.append('''on_predict'''				)

 def 						_snake_case					(     self		:       List[Any]  ,     lowercase_		:       Union[str, Any]  ,     lowercase_		:       List[Any]  ,     lowercase_		:       int  ,     **lowercase_		:       Union[str, Any]				):
     self.events.append('''on_save'''				)

 def 						_snake_case					(     self		:       str  ,     lowercase_		:       Tuple  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     **lowercase_		:       Any				):
     self.events.append('''on_log'''				)






 def 						_snake_case					(     self		:       Dict  ,     lowercase_		:       Optional[int]  ,     lowercase_		:       List[str]  ,     lowercase_		:       Union[str, Any]  ,     **lowercase_		:       Optional[int]				):
     self.events.append('''on_prediction_step'''				)





@require_torch
class        _UpperCAmelCase       (			unittest.TestCase):

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Tuple				=	tempfile.mkdtemp()

 def 						_snake_case					(     self		:       Tuple				):
     shutil.rmtree(self.output_dir				)

 def 						_snake_case					(     self		:       int  ,     lowercase_		:       Union[str, Any]=0  ,     lowercase_		:       Dict=0  ,     lowercase_		:       List[str]=64  ,     lowercase_		:       Union[str, Any]=64  ,     lowercase_		:       Union[str, Any]=None  ,     lowercase_		:       Any=False  ,     **lowercase_		:       List[Any]				):
     # disable_tqdm in TrainingArguments has a flaky default since it depends on the level of logging. We make sure
     # its set to False since the tests later on depend on its value.
     snake_case_ :     int				=	RegressionDataset(length=lowercase_				)
     snake_case_ :     Any				=	RegressionDataset(length=lowercase_				)
     snake_case_ :     int				=	RegressionModelConfig(a=lowercase_  ,     b=lowercase_				)
     snake_case_ :     Tuple				=	RegressionPreTrainedModel(lowercase_				)

     snake_case_ :     Any				=	TrainingArguments(self.output_dir  ,     disable_tqdm=lowercase_  ,     report_to=[]  ,     **lowercase_				)
     return Trainer(
         lowercase_  ,     lowercase_  ,     train_dataset=lowercase_  ,     eval_dataset=lowercase_  ,     callbacks=lowercase_  ,     )

 def 						_snake_case					(     self		:       Optional[int]  ,     lowercase_		:       Any  ,     lowercase_		:       List[Any]				):
     self.assertEqual(len(lowercase_				)  ,     len(lowercase_				)				)

     # Order doesn't matter
     snake_case_ :     Any				=	sorted(lowercase_  ,     key=lambda lowercase_				: cb.__name__ if isinstance(lowercase_  ,     lowercase_				) else cb.__class__.__name__				)
     snake_case_ :     List[str]				=	sorted(lowercase_  ,     key=lambda lowercase_				: cb.__name__ if isinstance(lowercase_  ,     lowercase_				) else cb.__class__.__name__				)

     for cba, cba in zip(lowercase_  ,     lowercase_				):
         if isinstance(lowercase_  ,     lowercase_				) and isinstance(lowercase_  ,     lowercase_				):
             self.assertEqual(lowercase_  ,     lowercase_				)
         elif isinstance(lowercase_  ,     lowercase_				) and not isinstance(lowercase_  ,     lowercase_				):
             self.assertEqual(lowercase_  ,     cba.__class__				)
         elif not isinstance(lowercase_  ,     lowercase_				) and isinstance(lowercase_  ,     lowercase_				):
             self.assertEqual(cba.__class__  ,     lowercase_				)
         else:
             self.assertEqual(lowercase_  ,     lowercase_				)

 def 						_snake_case					(     self		:       Optional[Any]  ,     lowercase_		:       Tuple				):
     snake_case_ :     Tuple				=	['''on_init_end''', '''on_train_begin''']
     snake_case_ :     List[Any]				=	0
     snake_case_ :     Union[str, Any]				=	len(trainer.get_eval_dataloader()				)
     snake_case_ :     List[Any]				=	['''on_prediction_step'''] * len(trainer.get_eval_dataloader()				) + ['''on_log''', '''on_evaluate''']
     for _ in range(trainer.state.num_train_epochs				):
         expected_events.append('''on_epoch_begin'''				)
         for _ in range(lowercase_				):
             step += 1
             expected_events += ["on_step_begin", "on_step_end"]
             if step % trainer.args.logging_steps == 0:
                 expected_events.append('''on_log'''				)
             if trainer.args.evaluation_strategy == IntervalStrategy.STEPS and step % trainer.args.eval_steps == 0:
                 expected_events += evaluation_events.copy()
             if step % trainer.args.save_steps == 0:
                 expected_events.append('''on_save'''				)
         expected_events.append('''on_epoch_end'''				)
         if trainer.args.evaluation_strategy == IntervalStrategy.EPOCH:
             expected_events += evaluation_events.copy()
     expected_events += ["on_log", "on_train_end"]
     return expected_events

 def 						_snake_case					(     self		:       List[str]				):
     snake_case_ :     Union[str, Any]				=	self.get_trainer()
     snake_case_ :     Dict				=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     # Callbacks passed at init are added to the default callbacks
     snake_case_ :     Optional[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]				)
     expected_callbacks.append(lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     # TrainingArguments.disable_tqdm controls if use ProgressCallback or PrinterCallback
     snake_case_ :     Optional[int]				=	self.get_trainer(disable_tqdm=lowercase_				)
     snake_case_ :     List[Any]				=	DEFAULT_CALLBACKS.copy() + [PrinterCallback]
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

 def 						_snake_case					(     self		:       int				):
     snake_case_ :     int				=	DEFAULT_CALLBACKS.copy() + [ProgressCallback]
     snake_case_ :     List[Any]				=	self.get_trainer()

     # We can add, pop, or remove by class name
     trainer.remove_callback(lowercase_				)
     expected_callbacks.remove(lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     snake_case_ :     Dict				=	self.get_trainer()
     snake_case_ :     Optional[int]				=	trainer.pop_callback(lowercase_				)
     self.assertEqual(cb.__class__  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     trainer.add_callback(lowercase_				)
     expected_callbacks.insert(0  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     # We can also add, pop, or remove by instance
     snake_case_ :     Optional[int]				=	self.get_trainer()
     snake_case_ :     List[Any]				=	trainer.callback_handler.callbacks[0]
     trainer.remove_callback(lowercase_				)
     expected_callbacks.remove(lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     snake_case_ :     List[Any]				=	self.get_trainer()
     snake_case_ :     Optional[int]				=	trainer.callback_handler.callbacks[0]
     snake_case_ :     Optional[Any]				=	trainer.pop_callback(lowercase_				)
     self.assertEqual(lowercase_  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)

     trainer.add_callback(lowercase_				)
     expected_callbacks.insert(0  ,     lowercase_				)
     self.check_callbacks_equality(trainer.callback_handler.callbacks  ,     lowercase_				)






 def 						_snake_case					(     self		:       List[Any]				):
     import warnings

     # XXX: for now ignore scatter_gather warnings in this test since it's not relevant to what's being tested
     warnings.simplefilter(action='''ignore'''  ,     category=lowercase_				)

     snake_case_ :     int				=	self.get_trainer(callbacks=[MyTestTrainerCallback]				)
     trainer.train()
     snake_case_ :     Union[str, Any]				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     # Independent log/save/eval
     snake_case_ :     int				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     logging_steps=5				)
     trainer.train()
     snake_case_ :     str				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     snake_case_ :     List[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     save_steps=5				)
     trainer.train()
     snake_case_ :     int				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     snake_case_ :     List[Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     eval_steps=5  ,     evaluation_strategy='''steps'''				)
     trainer.train()
     snake_case_ :     Union[str, Any]				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     snake_case_ :     Union[str, Any]				=	self.get_trainer(callbacks=[MyTestTrainerCallback]  ,     evaluation_strategy='''epoch'''				)
     trainer.train()
     snake_case_ :     Dict				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     # A bit of everything
     snake_case_ :     str				=	self.get_trainer(
         callbacks=[MyTestTrainerCallback]  ,     logging_steps=3  ,     save_steps=10  ,     eval_steps=5  ,     evaluation_strategy='''steps'''  ,     )
     trainer.train()
     snake_case_ :     str				=	trainer.callback_handler.callbacks[-2].events
     self.assertEqual(lowercase_  ,     self.get_expected_events(lowercase_				)				)

     # warning should be emitted for duplicated callbacks
     with patch('''transformers.trainer_callback.logger.warning'''				) as warn_mock:
         snake_case_ :     Dict				=	self.get_trainer(
             callbacks=[MyTestTrainerCallback, MyTestTrainerCallback]  ,     )
         assert str(lowercase_				) in warn_mock.call_args[0][0]





