


"""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 _A						(						lowercase	,   lowercase		):







    """simple docstring"""
    a					=[]
    for part_id in partition_order:
        a					=df.where(f'''SPARK_PARTITION_ID() = {part_id}'''		).collect()
        for row_idx, row in enumerate(lowercase		):
            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 _A						(						):







    """simple docstring"""
    a					=pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
    a					=spark.range(1_00		).repartition(1		)
    a					=Spark(lowercase		)
    # 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 _A						(						):







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

    for i, (row_id, row_dict) in enumerate(generate_fn()		):
        a		,				a					=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 _A						(						):







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







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

        a					=SparkExamplesIterable(lowercase		).shuffle_data_sources(lowercase		)
        assert shuffled_it.n_shards == 3
        for i, (row_id, row_dict) in enumerate(lowercase		):
            a		,				a					=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 _A						(						):







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

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

    # Partitions 1 and 3
    a					=SparkExamplesIterable(lowercase		).shard_data_sources(worker_id=1	,   num_workers=2		)
    assert shard_it_a.n_shards == 2
    a					=_get_expected_row_ids_and_row_dicts_for_partition_order(lowercase	,   [1, 3]		)
    for i, (row_id, row_dict) in enumerate(lowercase		):
        a		,				a					=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 _A						(						):







    """simple docstring"""
    a					=pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
    a					=spark.range(1_00		).repartition(1		)
    a					=Spark(lowercase		)
    # 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 unittest
from pathlib import Path
from tempfile import TemporaryDirectory

from transformers import AutoConfig, TFAutoModel, is_tensorflow_text_available, is_tf_available
from transformers.models.bert.tokenization_bert import BertTokenizer
from transformers.testing_utils import require_tensorflow_text, require_tf, slow


if is_tf_available():
       import tensorflow as tf

if is_tensorflow_text_available():
       from transformers.models.bert import TFBertTokenizer


lowerCamelCase_       :       str      						=  ["""bert-base-uncased""", """bert-base-cased"""]
lowerCamelCase_       :       List[str]      						=  """hf-internal-testing/tiny-bert-tf-only"""

if is_tf_available():






       class 				__A    (	tf.keras.Model      ):






        """simple docstring"""




        def __init__(		self			,   __A						) ->					Dict:
            super().__init__()
            a					=tokenizer
            a					=AutoConfig.from_pretrained(__A						)
            a					=TFAutoModel.from_config(__A						)





        def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					int:
            a					=self.tokenizer(__A						)
            a					=self.bert(**__A						)
            return out["pooler_output"]






@require_tf
@require_tensorflow_text
class 				__A    (	unittest.TestCase      ):






 """simple docstring"""




 def 				SCREAMING_SNAKE_CASE							(		self						) ->					str:
     super().setUp()

     a					=[
         BertTokenizer.from_pretrained(__A						) for checkpoint in (TOKENIZER_CHECKPOINTS * 2)
     ]  # repeat for when fast_bert_tokenizer=false
     a					=[TFBertTokenizer.from_pretrained(__A						) for checkpoint in TOKENIZER_CHECKPOINTS] + [
         TFBertTokenizer.from_pretrained(__A			,   use_fast_bert_tokenizer=__A						)
         for checkpoint in TOKENIZER_CHECKPOINTS
     ]
     assert len(self.tokenizers						) == len(self.tf_tokenizers						)

     a					=[
         '''This is a straightforward English test sentence.''',
         '''This one has some weird characters\rto\nsee\r\nif  those\u00E9break things.''',
         '''Now we\'re going to add some Chinese: 一 二 三 一二三''',
         '''And some much more rare Chinese: 齉 堃 齉堃''',
         '''Je vais aussi écrire en français pour tester les accents''',
         '''Classical Irish also has some unusual characters, so in they go: Gaelaċ, ꝼ''',
     ]
     a					=list(zip(self.test_sentences			,   self.test_sentences[::-1]						)						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Optional[int]:
     for tokenizer, tf_tokenizer in zip(self.tokenizers			,   self.tf_tokenizers						):
         for test_inputs in (self.test_sentences, self.paired_sentences):
             a					=tokenizer(__A			,   return_tensors='''tf'''			,   padding='''longest'''						)
             a					=tf_tokenizer(__A						)

             for key in python_outputs.keys():
                 self.assertTrue(tf.reduce_all(python_outputs[key].shape == tf_outputs[key].shape						)						)
                 self.assertTrue(tf.reduce_all(tf.cast(python_outputs[key]			,   tf.intaa						) == tf_outputs[key]						)						)


 @slow
 def 				SCREAMING_SNAKE_CASE							(		self						) ->					str:
     for tf_tokenizer in self.tf_tokenizers:
         a					=tf_tokenizer(self.paired_sentences						)
         a					=tf_tokenizer(
             text=[sentence[0] for sentence in self.paired_sentences]			,   text_pair=[sentence[1] for sentence in self.paired_sentences]			,   )
         for key in merged_outputs.keys():
             self.assertTrue(tf.reduce_all(tf.cast(merged_outputs[key]			,   tf.intaa						) == separated_outputs[key]						)						)


 @slow
 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[str]:
     for tf_tokenizer in self.tf_tokenizers:
         a					=tf.function(__A						)
         for test_inputs in (self.test_sentences, self.paired_sentences):
             a					=tf.constant(__A						)
             a					=compiled_tokenizer(__A						)
             a					=tf_tokenizer(__A						)

             for key in eager_outputs.keys():
                 self.assertTrue(tf.reduce_all(eager_outputs[key] == compiled_outputs[key]						)						)





 @slow
 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Tuple:
     for tf_tokenizer in self.tf_tokenizers:
         a					=ModelToSave(tokenizer=__A						)
         a					=tf.convert_to_tensor(self.test_sentences						)
         a					=model(__A						)  # Build model with some sample inputs
         with TemporaryDirectory() as tempdir:
             a					=Path(__A						) / '''saved.model'''
             model.save(__A						)
             a					=tf.keras.models.load_model(__A						)
         a					=loaded_model(__A						)
         # We may see small differences because the loaded model is compiled, so we need an epsilon for the test
         self.assertLessEqual(tf.reduce_max(tf.abs(out - loaded_output						)						)			,   1E-5						)