





'''simple docstring'''







from functools import reduce

lowerCAmelCase_					:    Optional[Any]      	=			(
    '73167176531330624919225119674426574742355349194934'
    '96983520312774506326239578318016984801869478851843'
    '85861560789112949495459501737958331952853208805511'
    '12540698747158523863050715693290963295227443043557'
    '66896648950445244523161731856403098711121722383113'
    '62229893423380308135336276614282806444486645238749'
    '30358907296290491560440772390713810515859307960866'
    '70172427121883998797908792274921901699720888093776'
    '65727333001053367881220235421809751254540594752243'
    '52584907711670556013604839586446706324415722155397'
    '53697817977846174064955149290862569321978468622482'
    '83972241375657056057490261407972968652414535100474'
    '82166370484403199890008895243450658541227588666881'
    '16427171479924442928230863465674813919123162824586'
    '17866458359124566529476545682848912883142607690042'
    '24219022671055626321111109370544217506941658960408'
    '07198403850962455444362981230987879927244284909188'
    '84580156166097919133875499200524063689912560717606'
    '05886116467109405077541002256983155200055935729725'
    '71636269561882670428252483600823257530420752963450'
)






def  _lowerCamelCase				( lowercase						:      str = N						)	->	int:
				return max(
				    # mypy cannot properly interpret reduce
				    int(reduce(lambda lowercase    ,							lowercase						: str(int(lowercase						) * int(lowercase						)						)    ,							n[i : i + 13]						)						)
				    for i in range(len(lowercase						) - 12						)						)


if __name__ == "__main__":
					print(f"""{solution() = }""")

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  snake_case  (       snake_case__			:Union[str, Any]   ,	snake_case__			:Dict)						->   Any:
       _A        =  []
       for part_id in partition_order:
              _A        =  df.where(F'''SPARK_PARTITION_ID() = {part_id}''').collect()
              for row_idx, row in enumerate(snake_case__):
                     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  snake_case  (       )						->   Optional[Any]:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(100).repartition(1)
       _A        =  Spark(snake_case__)
       # 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  snake_case  (       )						->   Union[str, Any]:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(10).repartition(2)
       _A        =  [1, 0]
       _A        =  _generate_iterable_examples(snake_case__   ,	snake_case__)  # Reverse the partitions.
       _A        =  _get_expected_row_ids_and_row_dicts_for_partition_order(snake_case__   ,	snake_case__)

       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  snake_case  (       )						->   int:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(10).repartition(1)
       _A        =  SparkExamplesIterable(snake_case__)
       assert it.n_shards == 1
       for i, (row_id, row_dict) in enumerate(snake_case__):
              assert row_id == F'''0_{i}'''
              assert row_dict == {"id": i}






@require_not_windows
@require_dill_gt_0_3_2
def  snake_case  (       )						->   Union[str, Any]:
       _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 snake_case__: x.reverse()
              _A        =  _get_expected_row_ids_and_row_dicts_for_partition_order(snake_case__   ,	[2, 1, 0])

              _A        =  SparkExamplesIterable(snake_case__).shuffle_data_sources(snake_case__)
              assert shuffled_it.n_shards == 3
              for i, (row_id, row_dict) in enumerate(snake_case__):
                     _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  snake_case  (       )						->   List[str]:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(20).repartition(4)

       # Partitions 0 and 2
       _A        =  SparkExamplesIterable(snake_case__).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(snake_case__   ,	[0, 2])
       for i, (row_id, row_dict) in enumerate(snake_case__):
              _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(snake_case__).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(snake_case__   ,	[1, 3])
       for i, (row_id, row_dict) in enumerate(snake_case__):
              _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  snake_case  (       )						->   Tuple:
       _A        =  pyspark.sql.SparkSession.builder.master("""local[*]""").appName("""pyspark""").getOrCreate()
       _A        =  spark.range(100).repartition(1)
       _A        =  Spark(snake_case__)
       # 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() == 100




