class_docstr: Implementation of hyperopt.Trials supporting
  distributed execution using Apache Spark clusters.
  This requires fmin to be run on a Spark cluster.
  
  Plugging SparkTrials into hyperopt.fmin() allows hyperopt
  to send model training and evaluation tasks to Spark workers,
  parallelizing hyperparameter search.
  Each trial (set of hyperparameter values) is handled within
  a single Spark task; i.e., each model will be fit and evaluated
  on a single worker machine.  Trials are run asynchronously.
  
  See hyperopt.Trials docs for general information about Trials.
  
  The fields we store in our trial docs match the base Trials class.  The fields include:
   - 'tid': trial ID
   - 'state': JOB_STATE_DONE, JOB_STATE_ERROR, etc.
   - 'result': evaluation result for completed trial run
   - 'refresh_time': timestamp for last status update
   - 'misc': includes:
      - 'error': (error type, error message)
   - 'book_time': timestamp for trial run start
docstr: This should not be called directly but is called via :func:`hyperopt.fmin`
  Refer to :func:`hyperopt.fmin` for docs on each argument
docstr_tok: ["This","should","not","be","called","directly","but","is","called","via",":","func",":","`","hyperopt.fmin","`","Refer","to",":","func",":","`","hyperopt.fmin","`","for","docs","on","each","argument"]
code: def fmin(
      self,
      fn,
      space,
      algo,
      max_evals,
      timeout,
      loss_threshold,
      max_queue_len,
      rstate,
      verbose,
      pass_expr_memo_ctrl,
      catch_eval_exceptions,
      return_argmin,
      show_progressbar,
      early_stop_fn,
      trials_save_file="",
  ):
      """
      This should not be called directly but is called via :func:`hyperopt.fmin`
      Refer to :func:`hyperopt.fmin` for docs on each argument
      """
  
      if timeout is not None:
          if self.timeout is not None:
              logger.warning(
                  "Timeout param was defined in Trials object, ignoring fmin definition"
              )
          else:
              validate_timeout(timeout)
              self.timeout = timeout
  
      if loss_threshold is not None:
          validate_loss_threshold(loss_threshold)
          self.loss_threshold = loss_threshold
  
      assert (
          not pass_expr_memo_ctrl
      ), "SparkTrials does not support `pass_expr_memo_ctrl`"
      assert (
          not catch_eval_exceptions
      ), "SparkTrials does not support `catch_eval_exceptions`"
  
      state = _SparkFMinState(
          self._spark,
          self._resource_profile,
          fn,
          space,
          self,
          early_stop_fn=early_stop_fn,
      )
  
      # Will launch a dispatcher thread which runs each trial task as one spark job.
      state.launch_dispatcher()
  
      try:
          res = fmin(
              fn,
              space,
              algo,
              max_evals,
              timeout=timeout,
              loss_threshold=loss_threshold,
              max_queue_len=max_queue_len,
              trials=self,
              allow_trials_fmin=False,  # -- prevent recursion
              rstate=rstate,
              pass_expr_memo_ctrl=None,  # not supported
              catch_eval_exceptions=catch_eval_exceptions,
              verbose=verbose,
              return_argmin=return_argmin,
              points_to_evaluate=None,  # not supported
              show_progressbar=show_progressbar,
              # do not check early stopping in fmin. SparkTrials early stopping is implemented in run_dispatcher
              early_stop_fn=None,
              trials_save_file="",  # not supported
          )
      except KeyboardInterrupt as e:
          self._fmin_cancelled = True
          self._fmin_cancelled_reason = FMIN_CANCELLED_REASON_USER
          logger.debug("fmin thread terminated by user.")
          raise e
      except BaseException as e:
          logger.debug("fmin thread exits with an exception raised.")
          raise e
      else:
          logger.debug("fmin thread exits normally.")
          return res
      finally:
          state.wait_for_all_threads()
  
          logger.info(
              f"Total Trials: {self.count_total_trials()}: {self.count_successful_trials()} succeeded, {self.count_failed_trials()} failed, {self.count_cancelled_trials()} cancelled."
          )
  
imports: import copy
  import os
  import threading
  import time
  import timeit
  import traceback
  from hyperopt import base, fmin, Trials
  from hyperopt.base import validate_timeout, validate_loss_threshold, STATUS_OK
  from hyperopt.utils import coarse_utcnow, _get_logger, _get_random_id
label_desc: Use image node_id 11 for calling the SparkTrials obj's underlying member method code with example usage: obj.fmin(fn, space, algo, max_evals, timeout, loss_threshold, max_queue_len, rstate, verbose, pass_expr_memo_ctrl, catch_eval_exceptions, return_argmin, show_progressbar, early_stop_fn, trials_save_file) and returns: res