class_docstr: Same as sklearn `BaseSearchCV` but with distributed
  training using spark
  
  Args:
      estimator (estimator object):
          This is assumed to implement the scikit-learn estimator interface.
          Either estimator needs to provide a ``score`` function,
          or ``scoring`` must be passed.
      sc (sparkContext): Spark context for spark broadcasting and rdd operations.
      partitions (int or 'auto'): default 'auto'
          Number of partitions to use for parallelization of parameter
          search space. Integer values or None will be used directly for `numSlices`,
          while 'auto' will set `numSlices` to the number required fits.
      preds (bool): keep predictions as attribute
class_docstr_tok: ["Same","as","sklearn","`","BaseSearchCV","`","but","with","distributed","training","using","spark","Args",":","estimator","(","estimator","object",")",":","This","is","assumed","to","implement","the","scikit-learn","estimator","interface",".","Either","estimator","needs","to","provide","a","``","score","``","function",",","or","``","scoring","``","must","be","passed",".","sc","(","sparkContext",")",":","Spark","context","for","spark","broadcasting","and","rdd","operations",".","partitions","(","int","or","'auto","'",")",":","default","'auto'","Number","of","partitions","to","use","for","parallelization","of","parameter","search","space",".","Integer","values","or","None","will","be","used","directly","for","`","numSlices","`",",","while","'auto","'","will","set","`","numSlices","`","to","the","number","required","fits",".","preds","(","bool",")",":","keep","predictions","as","attribute"]
docstr: Run fit with all sets of parameters. Parallelize fit operations
  using spark.
  
  Args:
      X (array-like, shape = [n_samples, n_features]): training vector,
          where n_samples is the number of samples and
          n_features is the number of features
      y (array-like, shape = [n_samples] or [n_samples, n_output]): target
          relative to X for classification or regression
      groups (array-like, with shape (n_samples,)): group labels for
          the samples used while splitting the dataset into
          train/test set
      **fit_params (dict of string -> object): parameters passed
          to the ``fit`` method of the estimator
docstr_tok: ["Run","fit","with","all","sets","of","parameters",".","Parallelize","fit","operations","using","spark",".","Args",":","X","(","array-like",",","shape","=","[","n_samples",",","n_features","]",")",":","training","vector",",","where","n_samples","is","the","number","of","samples","and","n_features","is","the","number","of","features","y","(","array-like",",","shape","=","[","n_samples","]","or","[","n_samples",",","n_output","]",")",":","target","relative","to","X","for","classification","or","regression","groups","(","array-like",",","with","shape","(","n_samples",",",")",")",":","group","labels","for","the","samples","used","while","splitting","the","dataset","into","train\/test","set","*","*","fit_params","(","dict","of","string","-",">","object",")",":","parameters","passed","to","the","``","fit","``","method","of","the","estimator"]
code: def fit(self, X, y=None, groups=None, **fit_params):
      """
      Run fit with all sets of parameters. Parallelize fit operations
      using spark.
  
      Args:
          X (array-like, shape = [n_samples, n_features]): training vector,
              where n_samples is the number of samples and
              n_features is the number of features
          y (array-like, shape = [n_samples] or [n_samples, n_output]): target
              relative to X for classification or regression
          groups (array-like, with shape (n_samples,)): group labels for
              the samples used while splitting the dataset into
              train/test set
          **fit_params (dict of string -> object): parameters passed
              to the ``fit`` method of the estimator
      """
      _check_estimator(self, verbose=self.verbose)
      estimator = self.estimator
      cv = check_cv(self.cv, y, classifier=is_classifier(estimator))
  
      scorers, self.multimetric_ = _check_multimetric_scoring(
          self.estimator, scoring=self.scoring
      )
  
      if self.multimetric_:
          if self.refit is not False and (
              not isinstance(self.refit, six.string_types)
              or
              # This will work for both dict / list (tuple)
              self.refit not in scorers
          ):
              raise ValueError(
                  "For multi-metric scoring, the parameter "
                  "refit must be set to a scorer key "
                  "to refit an estimator with the best "
                  "parameter setting on the whole data and "
                  "make the best_* attributes "
                  "available for that metric. If this is not "
                  "needed, refit should be set to False "
                  "explicitly. %r was passed." % self.refit
              )
          else:
              refit_metric = self.refit
      else:
          refit_metric = "score"
  
      X, y, groups = indexable(X, y, groups)
      n_splits = cv.get_n_splits(X, y, groups)
      # Regenerate parameter iterable for each fit
      candidate_params = list(self._get_param_iterator())
      n_candidates = len(candidate_params)
      if self.verbose > 0:
          print(
              (
                  "Fitting {0} folds for each of {1} candidates, totalling"
                  " {2} fits".format(
                      n_splits, n_candidates, n_candidates * n_splits
                  )
              )
          )
  
      base_estimator = _clone(self.estimator)
      pre_dispatch = self.pre_dispatch
  
      fit_sets = []
      cv_splitted = list(cv.split(X, y, groups))
      count = -1
      for fit_set in product(candidate_params, cv_splitted):
          count += 1
          fit_sets.append((count,) + fit_set)
      verbose = self.verbose
      return_train_score = self.return_train_score
      error_score = self.error_score
  
      if self.sc is None:
          base_estimator_ = base_estimator
          out = Parallel(n_jobs=self.n_jobs)(
              delayed(_fit_and_score)(
                  base_estimator_,
                  X,
                  y,
                  scorers,
                  x[2][0],
                  x[2][1],
                  verbose,
                  x[1],
                  fit_params=fit_params,
                  return_train_score=return_train_score,
                  return_n_test_samples=True,
                  return_times=True,
                  return_parameters=False,
                  error_score=error_score,
              )
              for x in fit_sets
          )
          out = [
              [fit_sets[ind][0], out[ind]]
              for ind in range(len(fit_sets))
          ]
      else:
          base_estimator_ = self.sc.broadcast(base_estimator)
          partitions = _parse_partitions(self.partitions, len(fit_sets))
          out = (
              self.sc.parallelize(fit_sets, numSlices=partitions)
              .map(
                  lambda x: [
                      x[0],
                      _fit_and_score(
                          base_estimator_,
                          X,
                          y,
                          scorers,
                          x[2][0],
                          x[2][1],
                          verbose,
                          x[1],
                          fit_params=fit_params,
                          return_train_score=return_train_score,
                          return_n_test_samples=True,
                          return_times=True,
                          return_parameters=False,
                          error_score=error_score,
                      ),
                  ]
              )
              .collect()
          )
  
      out = [out[i][1] for i in np.argsort([x[0] for x in out])]
  
      # if one choose to see train score, "out" will contain train score info
      if self.return_train_score:
          (
              train_score_dicts,
              test_score_dicts,
              test_sample_counts,
              fit_time,
              score_time,
          ) = list(zip(*out))
      else:
          (
              test_score_dicts,
              test_sample_counts,
              fit_time,
              score_time,
          ) = list(zip(*out))
  
      # test_score_dicts and train_score dicts are lists of dictionaries and
      # we make them into dict of lists
      test_scores = _aggregate_score_dicts(test_score_dicts)
      if self.return_train_score:
          train_scores = _aggregate_score_dicts(train_score_dicts)
  
      results = {}
  
      def _store(
          key_name, array, weights=None, splits=False, rank=False
      ):
          """A small helper to store the scores/times to the cv_results_"""
          array = np.array(array, dtype=np.float64).reshape(
              n_candidates, n_splits
          )
          if splits:
              for split_i in range(n_splits):
                  # Uses closure to alter the results
                  results["split%d_%s" % (split_i, key_name)] = array[
                      :, split_i
                  ]
  
          array_means = np.average(array, axis=1, weights=weights)
          results["mean_%s" % key_name] = array_means
          # Weighted std is not directly available in numpy
          array_stds = np.sqrt(
              np.average(
                  (array - array_means[:, np.newaxis]) ** 2,
                  axis=1,
                  weights=weights,
              )
          )
          results["std_%s" % key_name] = array_stds
  
          if rank:
              results["rank_%s" % key_name] = np.asarray(
                  rankdata(-array_means, method="min"), dtype=np.int32
              )
  
      _store("fit_time", fit_time)
      _store("score_time", score_time)
      param_results = defaultdict(
          partial(
              MaskedArray,
              np.empty(
                  n_candidates,
              ),
              mask=True,
              dtype=object,
          )
      )
      for cand_i, params in enumerate(candidate_params):
          for name, value in list(params.items()):
              # An all masked empty array gets created for the key
              # `"param_%s" % name` at the first occurence of `name`.
              # Setting the value at an index also unmasks that index
              param_results["param_%s" % name][cand_i] = value
  
      results.update(param_results)
      # Store a list of param dicts at the key 'params'
      results["params"] = candidate_params
  
      # NOTE test_sample counts (weights) remain the same for all candidates
      test_sample_counts = np.array(
          test_sample_counts[:n_splits], dtype=np.int
      )
      for scorer_name in list(scorers.keys()):
          # Computed the (weighted) mean and std for test scores alone
          _store(
              "test_%s" % scorer_name,
              test_scores[scorer_name],
              splits=True,
              rank=True,
              weights=test_sample_counts if self.iid else None,
          )
          if self.return_train_score:
              prev_keys = set(results.keys())
              _store(
                  "train_%s" % scorer_name,
                  train_scores[scorer_name],
                  splits=True,
              )
  
              if self.return_train_score == "warn":
                  for key in set(results.keys()) - prev_keys:
                      message = (
                          "You are accessing a training score ({!r}), "
                          "which will not be available by default "
                          "any more in 0.21. If you need training scores, "
                          "please set return_train_score=True"
                      ).format(key)
                      # warn on key access
                      results.add_warning(key, message, FutureWarning)
  
      # For multi-metric evaluation, store the best_index_, best_params_ and
      # best_score_ iff refit is one of the scorer names
      # In single metric evaluation, refit_metric is "score"
      if self.refit or not self.multimetric_:
          self.best_index_ = results[
              "rank_test_%s" % refit_metric
          ].argmin()
          self.best_params_ = candidate_params[self.best_index_]
          self.best_score_ = results["mean_test_%s" % refit_metric][
              self.best_index_
          ]
  
      if self.refit:
          self.best_estimator_ = _clone(base_estimator).set_params(
              **self.best_params_
          )
          if y is not None:
              self.best_estimator_.fit(X, y, **fit_params)
          else:
              self.best_estimator_.fit(X, **fit_params)
          if self.preds:
              preds = []
              for train_index, test_index in cv_splitted:
                  estimator_ = _clone(base_estimator).set_params(
                      **self.best_params_
                  )
                  estimator_.fit(X[train_index], y[train_index])
                  try:
                      preds.append(
                          estimator_.predict_proba(X[test_index])
                      )
                  except:
                      preds.append(estimator_.predict(X[test_index]))
              self.preds_ = np.vstack(preds)
  
      # Store the only scorer not as a dict for single metric evaluation
      self.scorer_ = scorers if self.multimetric_ else scorers["score"]
  
      self.cv_results_ = results
      self.n_splits_ = n_splits
  
      del self.sc
      if hasattr(self.estimator, "sc"):
          del self.estimator.sc
      return self
  
imports: import time
  import numbers
  import random
  import numpy
  import pandas
  from copy import copy
  from abc import ABCMeta
  from joblib import Parallel, delayed
  from sklearn.model_selection import ParameterGrid, GridSearchCV, RandomizedSearchCV, ParameterSampler, check_cv
  from sklearn.metrics import check_scoring
  from sklearn.base import BaseEstimator, is_classifier
  from sklearn.utils.metaestimators import if_delegate_has_method
  from sklearn.utils.validation import indexable
  from functools import partial
  from numpy.ma import MaskedArray
  from scipy.stats import rankdata
  from scipy.sparse import issparse
  from itertools import product
  from collections import defaultdict
  from .validation import _check_estimator, _check_base_estimator, _validate_params, _validate_models, _validate_names, _validate_estimators, _check_n_iter, _is_arraylike, _num_samples, _safe_indexing, _check_is_fitted
  from .utils import _multimetric_score, _num_samples, _aggregate_score_dicts, _score, _check_multimetric_scoring, _safe_split, _dict_slice_remove
  from .base import _clone, _get_value, _parse_partitions
label_desc: Use image node_id 2 for calling the DistBaseSearchCV obj's underlying member method code with example usage: obj.fit(X, y, groups) and returns: self