






import pickle

import numpy as np
from matplotlib import pyplot as plt







class 	SCREAMING_SNAKE_CASE	:




       def __init__(					self      : List[Any]	,    A__      : List[Any]	,    A__      : Dict	,    A__      : Optional[int]	,    A__      : Union[str, Any]	,    A__      : Tuple	,    A__      : List[Any]=0.2	,    A__      : List[Any]=0.2					):


            """simple docstring"""







            __lowerCamelCase						:       List[Any]				   =				bp_numa
            __lowerCamelCase						:       Tuple				   =				bp_numa
            __lowerCamelCase						:       List[Any]				   =				bp_numa
            __lowerCamelCase						:       Optional[Any]				   =				conva_get[:2]
            __lowerCamelCase						:       Any				   =				conva_get[2]
            __lowerCamelCase						:       Tuple				   =				size_pa
            __lowerCamelCase						:       List[str]				   =				rate_w
            __lowerCamelCase						:       List[Any]				   =				rate_t
            __lowerCamelCase						:       Tuple				   =				[
                np.mat(-1 * np.random.rand(self.conva[0]	,    self.conva[0]					) + 0.5					)
                for i in range(self.conva[1]					)
            ]
            __lowerCamelCase						:       int				   =				np.mat(-1 * np.random.rand(self.num_bpa	,    self.num_bpa					) + 0.5					)
            __lowerCamelCase						:       Dict				   =				np.mat(-1 * np.random.rand(self.num_bpa	,    self.num_bpa					) + 0.5					)
            __lowerCamelCase						:       List[str]				   =				-2 * np.random.rand(self.conva[1]					) + 1
            __lowerCamelCase						:       Dict				   =				-2 * np.random.rand(self.num_bpa					) + 1
            __lowerCamelCase						:       Dict				   =				-2 * np.random.rand(self.num_bpa					) + 1




       def 					a_					(					self      : Optional[int]	,    A__      : int					):


            """simple docstring"""







            __lowerCamelCase						:       List[Any]				   =				{
                """num_bp1""": self.num_bpa,
                """num_bp2""": self.num_bpa,
                """num_bp3""": self.num_bpa,
                """conv1""": self.conva,
                """step_conv1""": self.step_conva,
                """size_pooling1""": self.size_poolinga,
                """rate_weight""": self.rate_weight,
                """rate_thre""": self.rate_thre,
                """w_conv1""": self.w_conva,
                """wkj""": self.wkj,
                """vji""": self.vji,
                """thre_conv1""": self.thre_conva,
                """thre_bp2""": self.thre_bpa,
                """thre_bp3""": self.thre_bpa,
            }
            with open(_SCREAMING_SNAKE_CASE	,    """wb"""					) as f:
                 pickle.dump(_SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					)

            print(f"Model saved： {save_path}"					)




       @classmethod
       def 					a_					(					cls      : str	,    A__      : List[Any]					):


            """simple docstring"""







            with open(_SCREAMING_SNAKE_CASE	,    """rb"""					) as f:
                 __lowerCamelCase						:       Tuple				   =				pickle.load(_SCREAMING_SNAKE_CASE					)  # noqa: S301

            __lowerCamelCase						:       Tuple				   =				model_dic.get("""conv1"""					)
            conv_get.append(model_dic.get("""step_conv1"""					)					)
            __lowerCamelCase						:       Dict				   =				model_dic.get("""size_pooling1"""					)
            __lowerCamelCase						:       Optional[Any]				   =				model_dic.get("""num_bp1"""					)
            __lowerCamelCase						:       str				   =				model_dic.get("""num_bp2"""					)
            __lowerCamelCase						:       str				   =				model_dic.get("""num_bp3"""					)
            __lowerCamelCase						:       Union[str, Any]				   =				model_dic.get("""rate_weight"""					)
            __lowerCamelCase						:       int				   =				model_dic.get("""rate_thre"""					)
            # create model instance
            __lowerCamelCase						:       Dict				   =				CNN(_SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					)
            # modify model parameter
            __lowerCamelCase						:       Any				   =				model_dic.get("""w_conv1"""					)
            __lowerCamelCase						:       Optional[int]				   =				model_dic.get("""wkj"""					)
            __lowerCamelCase						:       Optional[Any]				   =				model_dic.get("""vji"""					)
            __lowerCamelCase						:       Any				   =				model_dic.get("""thre_conv1"""					)
            __lowerCamelCase						:       str				   =				model_dic.get("""thre_bp2"""					)
            __lowerCamelCase						:       List[str]				   =				model_dic.get("""thre_bp3"""					)
            return conv_ins




       def 					a_					(					self      : int	,    A__      : List[Any]					):


            """simple docstring"""







            return 1 / (1 + np.exp(-1 * x					))




       def 					a_					(					self      : List[Any]	,    A__      : str					):


            """simple docstring"""







            return round(_SCREAMING_SNAKE_CASE	,    3					)




       def 					a_					(					self      : Optional[int]	,    A__      : Tuple	,    A__      : Tuple	,    A__      : Dict	,    A__      : str	,    A__      : Tuple					):


            """simple docstring"""







            __lowerCamelCase						:       Optional[Any]				   =				convs[0]
            __lowerCamelCase						:       Union[str, Any]				   =				convs[1]
            __lowerCamelCase						:       str				   =				np.shape(_SCREAMING_SNAKE_CASE					)[0]
            # get the data slice of original image data, data_focus
            __lowerCamelCase						:       Dict				   =				[]
            for i_focus in range(0	,    size_data - size_conv + 1	,    _SCREAMING_SNAKE_CASE					):
                 for j_focus in range(0	,    size_data - size_conv + 1	,    _SCREAMING_SNAKE_CASE					):
                      __lowerCamelCase						:       Optional[Any]				   =				data[
                          i_focus : i_focus + size_conv, j_focus : j_focus + size_conv
                      ]
                      data_focus.append(_SCREAMING_SNAKE_CASE					)
        # calculate the feature map of every single kernel, and saved as list of matrix
            __lowerCamelCase						:       int				   =				[]
            __lowerCamelCase						:       Tuple				   =				int((size_data - size_conv) / conv_step + 1					)
            for i_map in range(_SCREAMING_SNAKE_CASE					):
                 __lowerCamelCase						:       List[Any]				   =				[]
                 for i_focus in range(len(_SCREAMING_SNAKE_CASE					)					):
                      __lowerCamelCase						:       int				   =				(
                          np.sum(np.multiply(data_focus[i_focus]	,    w_convs[i_map]					)					)
                          - thre_convs[i_map]
                      )
                      featuremap.append(self.sig(_SCREAMING_SNAKE_CASE					)					)
                 __lowerCamelCase						:       str				   =				np.asmatrix(_SCREAMING_SNAKE_CASE					).reshape(
                     _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					)
                 data_featuremap.append(_SCREAMING_SNAKE_CASE					)

            # expanding the data slice to One dimenssion
            __lowerCamelCase						:       Dict				   =				[]
            for each_focus in data_focus:
                 focusa_list.extend(self.Expand_Mat(_SCREAMING_SNAKE_CASE					)					)
            __lowerCamelCase						:       Optional[Any]				   =				np.asarray(_SCREAMING_SNAKE_CASE					)
            return focus_list, data_featuremap




       def 					a_					(					self      : List[str]	,    A__      : str	,    A__      : Optional[Any]	,    A__      : List[Any]="average_pool"					):


            """simple docstring"""







            __lowerCamelCase						:       int				   =				len(featuremaps[0]					)
            __lowerCamelCase						:       Optional[Any]				   =				int(size_map / size_pooling					)
            __lowerCamelCase						:       Tuple				   =				[]
            for i_map in range(len(_SCREAMING_SNAKE_CASE					)					):
                 __lowerCamelCase						:       Union[str, Any]				   =				featuremaps[i_map]
                 __lowerCamelCase						:       List[Any]				   =				[]
                 for i_focus in range(0	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					):
                      for j_focus in range(0	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					):
                           __lowerCamelCase						:       Any				   =				feature_map[
                               i_focus : i_focus + size_pooling,
                               j_focus : j_focus + size_pooling,
                           ]
                           if pooling_type == "average_pool":
                                # average pooling
                                map_pooled.append(np.average(_SCREAMING_SNAKE_CASE					)					)
                           elif pooling_type == "max_pooling":
                                # max pooling
                                map_pooled.append(np.max(_SCREAMING_SNAKE_CASE					)					)
                 __lowerCamelCase						:       List[str]				   =				np.asmatrix(_SCREAMING_SNAKE_CASE					).reshape(_SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					)
                 featuremap_pooled.append(_SCREAMING_SNAKE_CASE					)
            return featuremap_pooled




       def 					a_					(					self      : Optional[Any]	,    A__      : Tuple					):


            """simple docstring"""







            __lowerCamelCase						:       Dict				   =				[]
            for i in range(len(_SCREAMING_SNAKE_CASE					)					):
                 __lowerCamelCase						:       Tuple				   =				np.shape(data[i]					)
                 __lowerCamelCase						:       str				   =				data[i].reshape(1	,    shapes[0] * shapes[1]					)
                 __lowerCamelCase						:       Any				   =				data_listed.getA().tolist()[0]
                 data_expanded.extend(_SCREAMING_SNAKE_CASE					)
            __lowerCamelCase						:       str				   =				np.asarray(_SCREAMING_SNAKE_CASE					)
            return data_expanded




       def 					a_					(					self      : List[str]	,    A__      : Tuple					):


            """simple docstring"""







            __lowerCamelCase						:       Optional[int]				   =				np.asarray(_SCREAMING_SNAKE_CASE					)
            __lowerCamelCase						:       Any				   =				np.shape(_SCREAMING_SNAKE_CASE					)
            __lowerCamelCase						:       Tuple				   =				data_mat.reshape(1	,    shapes[0] * shapes[1]					)
            return data_expanded




       def 					a_					(					self      : Optional[int]	,    A__      : Union[str, Any]	,    A__      : Tuple	,    A__      : Optional[Any]	,    A__      : Optional[Any]	,    A__      : Optional[int]					):


            """simple docstring"""







            __lowerCamelCase						:       List[str]				   =				[]
            __lowerCamelCase						:       List[str]				   =				0
            for i_map in range(_SCREAMING_SNAKE_CASE					):
                 __lowerCamelCase						:       Dict				   =				np.ones((size_map, size_map)					)
                 for i in range(0	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					):
                      for j in range(0	,    _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					):
                           __lowerCamelCase						:       Optional[Any]				   =				pd_pool[
                               i_pool
                           ]
                           __lowerCamelCase						:       int				   =				i_pool + 1
                 __lowerCamelCase						:       Any				   =				np.multiply(
                     _SCREAMING_SNAKE_CASE	,    np.multiply(out_map[i_map]	,    (1 - out_map[i_map])					)					)
                 pd_all.append(_SCREAMING_SNAKE_CASE					)
            return pd_all




       def 					a_					(					self      : Optional[int]	,    A__      : Optional[Any]	,    A__      : List[str]	,    A__      : Union[str, Any]	,    A__      : Optional[int]	,    A__      : Optional[int]	,    A__      : int=bool					):


            """simple docstring"""







            print("""----------------------Start Training-------------------------"""					)
            print((""" - - Shape: Train_Data  """, np.shape(_SCREAMING_SNAKE_CASE					))					)
            print((""" - - Shape: Teach_Data  """, np.shape(_SCREAMING_SNAKE_CASE					))					)
            __lowerCamelCase						:       Any				   =				0
            __lowerCamelCase						:       Dict				   =				[]
            __lowerCamelCase						:       Optional[int]				   =				10000
            while rp < n_repeat and mse >= error_accuracy:
                 __lowerCamelCase						:       Any				   =				0
                 print(f"-------------Learning Time {rp}--------------"					)
                 for p in range(len(_SCREAMING_SNAKE_CASE					)					):
                      # print('------------Learning Image: %d--------------'%p)
                      __lowerCamelCase						:       str				   =				np.asmatrix(datas_train[p]					)
                      __lowerCamelCase						:       Optional[Any]				   =				np.asarray(datas_teach[p]					)
                      __lowerCamelCase   ,      __lowerCamelCase						:       Tuple				   =				self.convolute(
                          _SCREAMING_SNAKE_CASE	,    self.conva	,    self.w_conva	,    self.thre_conva	,    conv_step=self.step_conva	,    )
                      __lowerCamelCase						:       List[Any]				   =				self.pooling(_SCREAMING_SNAKE_CASE	,    self.size_poolinga					)
                      __lowerCamelCase						:       Union[str, Any]				   =				np.shape(_SCREAMING_SNAKE_CASE					)
                      __lowerCamelCase						:       Tuple				   =				self._expand(_SCREAMING_SNAKE_CASE					)
                      __lowerCamelCase						:       Optional[int]				   =				data_bp_input

                      __lowerCamelCase						:       str				   =				np.dot(_SCREAMING_SNAKE_CASE	,    self.vji.T					) - self.thre_bpa
                      __lowerCamelCase						:       Tuple				   =				self.sig(_SCREAMING_SNAKE_CASE					)
                      __lowerCamelCase						:       Union[str, Any]				   =				np.dot(_SCREAMING_SNAKE_CASE	,    self.wkj.T					) - self.thre_bpa
                      __lowerCamelCase						:       int				   =				self.sig(_SCREAMING_SNAKE_CASE					)

                      # --------------Model Leaning ------------------------
                      # calculate error and gradient---------------
                      __lowerCamelCase						:       List[Any]				   =				np.multiply(
                          (data_teach - bp_outa)	,    np.multiply(_SCREAMING_SNAKE_CASE	,    (1 - bp_outa)					)					)
                      __lowerCamelCase						:       Dict				   =				np.multiply(
                          np.dot(_SCREAMING_SNAKE_CASE	,    self.wkj					)	,    np.multiply(_SCREAMING_SNAKE_CASE	,    (1 - bp_outa)					)					)
                      __lowerCamelCase						:       Union[str, Any]				   =				np.dot(_SCREAMING_SNAKE_CASE	,    self.vji					)

                      __lowerCamelCase						:       int				   =				pd_i_all / (self.size_poolinga * self.size_poolinga)
                      __lowerCamelCase						:       Optional[Any]				   =				pd_conva_pooled.T.getA().tolist()
                      __lowerCamelCase						:       Optional[Any]				   =				self._calculate_gradient_from_pool(
                          _SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE	,    shape_featuremapa[0]	,    shape_featuremapa[1]	,    self.size_poolinga	,    )
                      # weight and threshold learning process---------
                      # convolution layer
                      for k_conv in range(self.conva[1]					):
                           __lowerCamelCase						:       List[Any]				   =				self._expand_mat(pd_conva_all[k_conv]					)
                           __lowerCamelCase						:       Tuple				   =				self.rate_weight * np.dot(_SCREAMING_SNAKE_CASE	,    _SCREAMING_SNAKE_CASE					)

                           __lowerCamelCase						:       List[Any]				   =				self.w_conva[k_conv] + delta_w.reshape(
                               (self.conva[0], self.conva[0])					)

                           __lowerCamelCase						:       Optional[Any]				   =				(
                               self.thre_conva[k_conv]
                               - np.sum(pd_conva_all[k_conv]					) * self.rate_thre
                           )
                      # all connected layer
                      __lowerCamelCase						:       Tuple				   =				self.wkj + pd_k_all.T * bp_outa * self.rate_weight
                      __lowerCamelCase						:       int				   =				self.vji + pd_j_all.T * bp_outa * self.rate_weight
                      __lowerCamelCase						:       Any				   =				self.thre_bpa - pd_k_all * self.rate_thre
                      __lowerCamelCase						:       Dict				   =				self.thre_bpa - pd_j_all * self.rate_thre
                      # calculate the sum error of all single image
                      __lowerCamelCase						:       List[str]				   =				np.sum(abs(data_teach - bp_outa					)					)
                      error_count += errors
                      # print('   ----Teach      ',data_teach)
                      # print('   ----BP_output  ',bp_out3)
                 __lowerCamelCase						:       Optional[int]				   =				rp + 1
                 __lowerCamelCase						:       List[Any]				   =				error_count / patterns
                 all_mse.append(_SCREAMING_SNAKE_CASE					)

            def draw_error():
                 __lowerCamelCase						:       Optional[Any]				   =				[error_accuracy for i in range(int(n_repeat * 1.2					)					)]
                 plt.plot(_SCREAMING_SNAKE_CASE	,    """+-"""					)
                 plt.plot(_SCREAMING_SNAKE_CASE	,    """r--"""					)
                 plt.xlabel("""Learning Times"""					)
                 plt.ylabel("""All_mse"""					)
                 plt.grid(_SCREAMING_SNAKE_CASE	,    alpha=0.5					)
                 plt.show()

            print("""------------------Training Complished---------------------"""					)
            print((""" - - Training epoch: """, rp, f"     - - Mse: {mse:.6f}")					)
            if draw_e:
                 draw_error()
            return mse




       def 					a_					(					self      : Union[str, Any]	,    A__      : Optional[Any]					):


            """simple docstring"""







            __lowerCamelCase						:       Dict				   =				[]
            print("""-------------------Start Testing-------------------------"""					)
            print((""" - - Shape: Test_Data  """, np.shape(_SCREAMING_SNAKE_CASE					))					)
            for p in range(len(_SCREAMING_SNAKE_CASE					)					):
                 __lowerCamelCase						:       Tuple				   =				np.asmatrix(datas_test[p]					)
                 __lowerCamelCase   ,      __lowerCamelCase						:       Any				   =				self.convolute(
                     _SCREAMING_SNAKE_CASE	,    self.conva	,    self.w_conva	,    self.thre_conva	,    conv_step=self.step_conva	,    )
                 __lowerCamelCase						:       Optional[Any]				   =				self.pooling(_SCREAMING_SNAKE_CASE	,    self.size_poolinga					)
                 __lowerCamelCase						:       Optional[int]				   =				self._expand(_SCREAMING_SNAKE_CASE					)

                 __lowerCamelCase						:       str				   =				data_bp_input
                 __lowerCamelCase						:       Dict				   =				bp_outa * self.vji.T - self.thre_bpa
                 __lowerCamelCase						:       int				   =				self.sig(_SCREAMING_SNAKE_CASE					)
                 __lowerCamelCase						:       List[Any]				   =				bp_outa * self.wkj.T - self.thre_bpa
                 __lowerCamelCase						:       Tuple				   =				self.sig(_SCREAMING_SNAKE_CASE					)
                 produce_out.extend(bp_outa.getA().tolist()					)
            __lowerCamelCase						:       Dict				   =				[list(map(self.do_round	,    _SCREAMING_SNAKE_CASE					)					) for each in produce_out]
            return np.asarray(_SCREAMING_SNAKE_CASE					)




       def 					a_					(					self      : Optional[Any]	,    A__      : List[Any]					):


            """simple docstring"""







            __lowerCamelCase						:       List[str]				   =				np.asmatrix(_SCREAMING_SNAKE_CASE					)
            __lowerCamelCase   ,      __lowerCamelCase						:       str				   =				self.convolute(
                _SCREAMING_SNAKE_CASE	,    self.conva	,    self.w_conva	,    self.thre_conva	,    conv_step=self.step_conva	,    )
            __lowerCamelCase						:       str				   =				self.pooling(_SCREAMING_SNAKE_CASE	,    self.size_poolinga					)

            return data_conveda, data_pooleda


if __name__ == "__main__":
   pass











'''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     __lowercase  (_lowercase,						_lowercase     )   ->     List[str]:


     """simple docstring"""
     __lowerCamelCase						:       Union[str, Any]				   =				[]
     for part_id in partition_order:
          __lowerCamelCase						:       Union[str, Any]				   =				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     __lowercase  ()   ->     Optional[int]:


     """simple docstring"""
     __lowerCamelCase						:       int				   =				pyspark.sql.SparkSession.builder.master("""local[*]"""     ).appName("""pyspark"""     ).getOrCreate()
     __lowerCamelCase						:       List[str]				   =				spark.range(100     ).repartition(1     )
     __lowerCamelCase						:       str				   =				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     __lowercase  ()   ->     Optional[Any]:


     """simple docstring"""
     __lowerCamelCase						:       Any				   =				pyspark.sql.SparkSession.builder.master("""local[*]"""     ).appName("""pyspark"""     ).getOrCreate()
     __lowerCamelCase						:       Dict				   =				spark.range(10     ).repartition(2     )
     __lowerCamelCase						:       List[Any]				   =				[1, 0]
     __lowerCamelCase						:       Any				   =				_generate_iterable_examples(_lowercase,						_lowercase     )  # Reverse the partitions.
     __lowerCamelCase						:       int				   =				_get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase,						_lowercase     )

     for i, (row_id, row_dict) in enumerate(generate_fn()     ):
          __lowerCamelCase   ,      __lowerCamelCase						:       str				   =				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     __lowercase  ()   ->     int:


     """simple docstring"""
     __lowerCamelCase						:       Dict				   =				pyspark.sql.SparkSession.builder.master("""local[*]"""     ).appName("""pyspark"""     ).getOrCreate()
     __lowerCamelCase						:       Union[str, Any]				   =				spark.range(10     ).repartition(1     )
     __lowerCamelCase						:       Dict				   =				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     __lowercase  ()   ->     Tuple:


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

          __lowerCamelCase						:       Tuple				   =				SparkExamplesIterable(_lowercase     ).shuffle_data_sources(_lowercase     )
          assert shuffled_it.n_shards == 3
          for i, (row_id, row_dict) in enumerate(_lowercase     ):
               __lowerCamelCase   ,      __lowerCamelCase						:       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     __lowercase  ()   ->     str:


     """simple docstring"""
     __lowerCamelCase						:       Union[str, Any]				   =				pyspark.sql.SparkSession.builder.master("""local[*]"""     ).appName("""pyspark"""     ).getOrCreate()
     __lowerCamelCase						:       List[str]				   =				spark.range(20     ).repartition(4     )

     # Partitions 0 and 2
     __lowerCamelCase						:       Union[str, Any]				   =				SparkExamplesIterable(_lowercase     ).shard_data_sources(worker_id=0,						num_workers=2     )
     assert shard_it_a.n_shards == 2
     __lowerCamelCase						:       Optional[int]				   =				_get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase,						[0, 2]     )
     for i, (row_id, row_dict) in enumerate(_lowercase     ):
          __lowerCamelCase   ,      __lowerCamelCase						:       int				   =				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						:       List[str]				   =				SparkExamplesIterable(_lowercase     ).shard_data_sources(worker_id=1,						num_workers=2     )
     assert shard_it_a.n_shards == 2
     __lowerCamelCase						:       int				   =				_get_expected_row_ids_and_row_dicts_for_partition_order(_lowercase,						[1, 3]     )
     for i, (row_id, row_dict) in enumerate(_lowercase     ):
          __lowerCamelCase   ,      __lowerCamelCase						:       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     __lowercase  ()   ->     int:


     """simple docstring"""
     __lowerCamelCase						:       Union[str, Any]				   =				pyspark.sql.SparkSession.builder.master("""local[*]"""     ).appName("""pyspark"""     ).getOrCreate()
     __lowerCamelCase						:       Dict				   =				spark.range(100     ).repartition(1     )
     __lowerCamelCase						:       Any				   =				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



