<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _spark_map(fun, indexed_param_grid, sc, seed, X_bc):
 """We cannot pass a RandomState instance to each spark worker since it will behave identically across partitions. Instead, we explictly handle the partitions with a newly seeded instance. The seed for each partition will be the "seed" (MonteCarloProfile.seed) + "split_index" which is the partition index. Following this trick: https://wegetsignal.wordpress.com/2015/05/08/ generating-random-numbers-for-rdd-in-spark/ """


    def _wrap_random_state(split_index, partition):
        prng = np.random.RandomState(seed + split_index)
        yield map(partial(fun, prng=prng, X=X_bc), partition)

    par_param_grid = sc.parallelize(indexed_param_grid)
    indexed_results = par_param_grid.mapPartitionsWithIndex(
        _wrap_random_state
    ).collect()
    return [item for sublist in indexed_results for item in sublist]