





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       ):
						_UpperCamelCase      : Tuple         =   []
						for part_id in partition_order:
												_UpperCamelCase      : int         =   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_    (							):
						_UpperCamelCase      : Any         =   pyspark.sql.SparkSession.builder.master('''local[*]'''       ).appName('''pyspark'''       ).getOrCreate()
						_UpperCamelCase      : Union[str, Any]         =   spark.range(100       ).repartition(1       )
						_UpperCamelCase      : Dict         =   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_    (							):
						_UpperCamelCase      : Union[str, Any]         =   pyspark.sql.SparkSession.builder.master('''local[*]'''       ).appName('''pyspark'''       ).getOrCreate()
						_UpperCamelCase      : str         =   spark.range(10       ).repartition(2       )
						_UpperCamelCase      : List[str]         =   [1, 0]
						_UpperCamelCase      : str         =   _generate_iterable_examples(_lowercase	,						_lowercase       )  # Reverse the partitions.
						_UpperCamelCase      : str         =   _get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase	,						_lowercase       )

						for i, (row_id, row_dict) in enumerate(generate_fn()       ):
												_UpperCamelCase      : Union[str, Any]         =   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_    (							):
						_UpperCamelCase      : Optional[Any]         =   pyspark.sql.SparkSession.builder.master('''local[*]'''       ).appName('''pyspark'''       ).getOrCreate()
						_UpperCamelCase      : List[str]         =   spark.range(10       ).repartition(1       )
						_UpperCamelCase      : Optional[int]         =   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_    (							):
						_UpperCamelCase      : List[Any]         =   pyspark.sql.SparkSession.builder.master('''local[*]'''       ).appName('''pyspark'''       ).getOrCreate()
						_UpperCamelCase      : int         =   spark.range(30       ).repartition(3       )
						# Mock the generator so that shuffle reverses the partition indices.
						with patch('''numpy.random.Generator'''       ) as generator_mock:
												_UpperCamelCase      : Optional[int]         =   lambda _lowercase       : x.reverse()
												_UpperCamelCase      : Optional[Any]         =   _get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase	,						[2, 1, 0]       )

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

						# Partitions 0 and 2
						_UpperCamelCase      : Union[str, Any]         =   SparkExamplesIterable(_lowercase       ).shard_data_sources(worker_id=0	,						num_workers=2       )
						assert shard_it_a.n_shards == 2
						_UpperCamelCase      : Any         =   _get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase	,						[0, 2]       )
						for i, (row_id, row_dict) in enumerate(_lowercase       ):
												_UpperCamelCase      : List[Any]         =   expected_row_ids_and_row_dicts_a[i]
												assert row_id == expected_row_id
												assert row_dict == expected_row_dict

						# Partitions 1 and 3
						_UpperCamelCase      : List[Any]         =   SparkExamplesIterable(_lowercase       ).shard_data_sources(worker_id=1	,						num_workers=2       )
						assert shard_it_a.n_shards == 2
						_UpperCamelCase      : List[Any]         =   _get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase	,						[1, 3]       )
						for i, (row_id, row_dict) in enumerate(_lowercase       ):
												_UpperCamelCase      : int         =   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_    (							):
						_UpperCamelCase      : Union[str, Any]         =   pyspark.sql.SparkSession.builder.master('''local[*]'''       ).appName('''pyspark'''       ).getOrCreate()
						_UpperCamelCase      : Any         =   spark.range(100       ).repartition(1       )
						_UpperCamelCase      : Optional[int]         =   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() == 100










"""simple docstring"""




def 	a_    (							_lowercase	,						_lowercase       ):
						if discount_rate < 0:
												raise ValueError('''Discount rate cannot be negative'''       )
						if not cash_flows:
												raise ValueError('''Cash flows list cannot be empty'''       )
						_UpperCamelCase      : Optional[int]         =   sum(
						    cash_flow / ((1 + discount_rate) ** i) for i, cash_flow in enumerate(_lowercase       )       )
						return round(_lowercase	,						ndigits=2       )


if __name__ == "__main__":
						import doctest

						doctest.testmod()



