class_docstr: Class for managing threads which run concurrent Spark jobs.
  
  This maintains a primary dispatcher thread, plus 1 thread per Hyperopt trial.
  Each trial's thread runs 1 Spark job with 1 task.
code: def __init__(
      self,
      spark,
      resource_profile,
      eval_function,
      space,
      trials,
      early_stop_fn,
  ):
      self.spark = spark
      self.resource_profile = resource_profile
      self.eval_function = eval_function
      self.space = space
      self.trials = trials
      self.early_stop_fn = early_stop_fn
      self.early_stop_args = []
      self._fmin_done = False
      self._dispatcher_thread = None
      self._task_threads = set()
  
      if self.trials._spark_supports_job_cancelling:
          spark_context = spark.sparkContext
          self._job_group_id = spark_context.getLocalProperty(
              "spark.jobGroup.id"
          )
          self._job_desc = spark_context.getLocalProperty(
              "spark.job.description"
          )
          interrupt_on_cancel = spark_context.getLocalProperty(
              "spark.job.interruptOnCancel"
          )
          if interrupt_on_cancel is None:
              self._job_interrupt_on_cancel = False
          else:
              self._job_interrupt_on_cancel = (
                  "true" == interrupt_on_cancel.lower()
              )
          # In certain Spark deployments, the local property "spark.jobGroup.id"
          # value is None, so we create one to use for SparkTrials.
          if self._job_group_id is None:
              self._job_group_id = (
                  "Hyperopt_SparkTrials_" + _get_random_id()
              )
          if self._job_desc is None:
              self._job_desc = (
                  "Trial evaluation jobs launched by hyperopt fmin"
              )
          logger.debug(
              "Job group id: {g}, job desc: {d}, job interrupt on cancel: {i}".format(
                  g=self._job_group_id,
                  d=self._job_desc,
                  i=self._job_interrupt_on_cancel,
              )
          )
  
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 1 to create a new _SparkFMinState object  with example: obj = _SparkFMinState(spark, resource_profile, eval_function, space, trials, early_stop_fn)