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: :param parallelism: Maximum number of parallel trials to run,
                      i.e., maximum number of concurrent Spark tasks.
                      The actual parallelism is subject to available Spark task slots at
                      runtime.
                      If set to None (default) or a non-positive value, this will be set to
                      Spark's default parallelism or `1`.
                      We cap the value at `MAX_CONCURRENT_JOBS_ALLOWED=128`.
  :param timeout: Maximum time (in seconds) which fmin is allowed to take.
                  If this timeout is hit, then fmin will cancel running and proposed trials.
                  It will retain all completed trial runs and return the best result found
                  so far.
  :param spark_session: A SparkSession object. If None is passed, SparkTrials will attempt
                        to use an existing SparkSession or create a new one. SparkSession is
                        the entry point for various facilities provided by Spark. For more
                        information, visit the documentation for PySpark.
  :param resource_profile: A ResourceProfile object. If not None, SparkTrials will use resources
                           specified by the profile in Spark training tasks.
docstr_tok: [":","param","parallelism",":","Maximum","number","of","parallel","trials","to","run",",","i.e.",",","maximum","number","of","concurrent","Spark","tasks",".","The","actual","parallelism","is","subject","to","available","Spark","task","slots","at","runtime",".","If","set","to","None","(","default",")","or","a","non-positive","value",",","this","will","be","set","to","Spark","'s","default","parallelism","or","`","1","`",".","We","cap","the","value","at","`","MAX_CONCURRENT_JOBS_ALLOWED=128","`",".",":","param","timeout",":","Maximum","time","(","in","seconds",")","which","fmin","is","allowed","to","take",".","If","this","timeout","is","hit",",","then","fmin","will","cancel","running","and","proposed","trials",".","It","will","retain","all","completed","trial","runs","and","return","the","best","result","found","so","far",".",":","param","spark_session",":","A","SparkSession","object",".","If","None","is","passed",",","SparkTrials","will","attempt","to","use","an","existing","SparkSession","or","create","a","new","one",".","SparkSession","is","the","entry","point","for","various","facilities","provided","by","Spark",".","For","more","information",",","visit","the","documentation","for","PySpark",".",":","param","resource_profile",":","A","ResourceProfile","object",".","If","not","None",",","SparkTrials","will","use","resources","specified","by","the","profile","in","Spark","training","tasks","."]
code: def __init__(
      self,
      parallelism=None,
      timeout=None,
      loss_threshold=None,
      spark_session=None,
      resource_profile=None,
  ):
      """
      :param parallelism: Maximum number of parallel trials to run,
                          i.e., maximum number of concurrent Spark tasks.
                          The actual parallelism is subject to available Spark task slots at
                          runtime.
                          If set to None (default) or a non-positive value, this will be set to
                          Spark's default parallelism or `1`.
                          We cap the value at `MAX_CONCURRENT_JOBS_ALLOWED=128`.
      :param timeout: Maximum time (in seconds) which fmin is allowed to take.
                      If this timeout is hit, then fmin will cancel running and proposed trials.
                      It will retain all completed trial runs and return the best result found
                      so far.
      :param spark_session: A SparkSession object. If None is passed, SparkTrials will attempt
                            to use an existing SparkSession or create a new one. SparkSession is
                            the entry point for various facilities provided by Spark. For more
                            information, visit the documentation for PySpark.
      :param resource_profile: A ResourceProfile object. If not None, SparkTrials will use resources
                               specified by the profile in Spark training tasks.
      """
      super().__init__(exp_key=None, refresh=False)
      if not _have_spark:
          raise Exception(
              "SparkTrials cannot import pyspark classes.  Make sure that PySpark "
              "is available in your environment.  E.g., try running 'import pyspark'"
          )
      validate_timeout(timeout)
      validate_loss_threshold(loss_threshold)
      self._spark = (
          SparkSession.builder.getOrCreate()
          if spark_session is None
          else spark_session
      )
      self._spark_context = self._spark.sparkContext
      self._spark_pinned_threads_enabled = isinstance(
          self._spark_context._gateway, ClientServer
      )
      # The feature to support controlling jobGroupIds is in SPARK-22340
      self._spark_supports_job_cancelling = (
          self._spark_pinned_threads_enabled
          or hasattr(
              self._spark_context.parallelize([1]),
              "collectWithJobGroup",
          )
      )
      spark_default_parallelism = self._spark_context.defaultParallelism
      self.parallelism = self._decide_parallelism(
          requested_parallelism=parallelism,
          spark_default_parallelism=spark_default_parallelism,
      )
      self.user_specified_parallelism = parallelism
  
      self._spark_supports_resource_profile = hasattr(
          self._spark_context.parallelize([1]), "withResources"
      ) and not self._spark.conf.get("spark.master", "").startswith(
          "local"
      )
      if self._spark_supports_resource_profile:
          self._resource_profile = resource_profile
      else:
          self._resource_profile = None
          if resource_profile is not None:
              logger.warning(
                  "SparkTrials was constructed with a ResourceProfile, but this Apache "
                  "Spark version does not support stage-level scheduling."
              )
  
      if (
          not self._spark_supports_job_cancelling
          and timeout is not None
      ):
          logger.warning(
              "SparkTrials was constructed with a timeout specified, but this Apache "
              "Spark version does not support job group-based cancellation. The "
              "timeout will be respected when starting new Spark jobs, but "
              "SparkTrials will not be able to cancel running Spark jobs which exceed"
              " the timeout."
          )
  
      self.timeout = timeout
      self.loss_threshold = loss_threshold
      self._fmin_cancelled = False
      self._fmin_cancelled_reason = None
      self.refresh()
  
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 2 to create a new SparkTrials object from inherited base classes: Trials with example: obj = SparkTrials(parallelism, timeout, loss_threshold, spark_session, resource_profile)