Parallelize single-node hyperparameter tuning with Hyperopt SparkTrials

1.Start from a single-machine fmin() workflow

You can't parallelize a tuning run until you have one, so start with the plain single-machine Hyperopt workflow. Databricks describes four steps. First, define a function to minimize. Second, define a search space over the hyperparameters. Third, select a search algorithm. Fourth, run the tuning algorithm with Hyperopt fmin(). The Databricks example notebook tunes a scikit-learn support vector classifier on the Iris dataset, and the only hyperparameter it searches is the regularization parameter C.

Most of the code is in the objective function. It receives a candidate value of C, trains a model, scores it, and returns a loss. Hyperopt always *minimizes*. If your score is one where higher is better, such as accuracy, return its negative. The notebook scores each candidate with cross_val_score, but this objective only needs to return a loss. How you validate the model inside it is up to you.

def objective(C):
    # Create a support vector classifier model
    clf = SVC(C=C)

    # Use the cross-validation accuracy to compare the models' performance
    accuracy = cross_val_score(clf, X, y).mean()

    # Hyperopt tries to minimize the objective function. A higher accuracy value means a better model, so you must return the negative accuracy.
    return {'loss': -accuracy, 'status': STATUS_OK}

The search space in the example is hp.lognormal('C', 0, 1.0). For the algorithm, the notebook names two main choices. hyperopt.tpe.suggest is Tree of Parzen Estimators, a Bayesian method that picks new settings based on past results. hyperopt.rand.suggest is random search, which samples the space without adapting. Keep this difference in mind, because it matters again when you turn up parallelism. The table below lists the core fmin() arguments.

argmin = fmin(
  fn=objective,
  space=search_space,
  algo=algo,
   ? =16)

Sources12

2.Distribute the same run by adding SparkTrials

Only one thing changes. To distribute tuning, you add one more argument to fmin(): a Trials class called SparkTrials. Databricks developed SparkTrials so that a scikit-learn model, which only knows how to train on one machine, can be tuned faster on a cluster. Each model still trains on a single machine, but many models train at once on different workers. The objective, the search space, the algorithm and max_evals all stay the same.

from hyperopt import SparkTrials

# To display the API documentation for the SparkTrials class, uncomment the following line.
# help(SparkTrials)

spark_trials = SparkTrials()

with mlflow.start_run():
  argmin = fmin(
    fn=objective,
    space=search_space,
    algo=algo,
    max_evals=16,
    trials=spark_trials)

# Print the best value found for C
print("Best value found: ", argmin)

Note two details in this sample. SparkTrials() is created with no arguments, so it uses the defaults for both of its optional arguments, which the next section covers. The call to fmin() also sits inside with mlflow.start_run():. Automated MLflow tracking is on by default, and this wrapper is what Databricks recommends. Section 5 explains why.

Sources12

3.How SparkTrials schedules trials: parallelism, timeout and the cluster

In Hyperopt, a trial generally means fitting one model on one setting of hyperparameters. With SparkTrials, the work is split between the driver and the workers. The driver generates new trials, and the workers evaluate them. Each trial is generated with a Spark job that has one task, and that task runs on a worker. If the cluster allows several tasks per worker, one worker can evaluate several trials at the same time.

SparkTrials takes two optional arguments.

**parallelism** is the maximum number of trials to evaluate at the same time. A higher value lets you test more hyperparameter settings at once. Because Hyperopt proposes new trials based on past results, there is a trade-off between parallelism and adaptivity. The concepts page gives the default as the number of Spark executors available and the maximum as 128. If you ask for more parallelism than the cluster configuration allows in concurrent tasks, SparkTrials lowers it to that limit.

**timeout** is the maximum number of seconds an fmin() call can take. When it is exceeded, all runs are terminated and fmin() exits. Information about completed runs is saved, so the trials that finished before the timeout are kept.

Two fmin() arguments interact with these. max_queue_len sets how many settings Hyperopt generates ahead of time. TPE generation can be slow, so raising it above the default of 1 can help, but it should generally be no larger than the parallelism setting. early_stop_fn can end the run before max_evals is reached. Under SparkTrials this function is polled, so it isn't guaranteed to run after every trial.

Sources23

4.SparkTrials or Trials: match the class to the model

SparkTrials is only for models that are *not* already distributed. When the objective function trains a model with a distributed algorithm such as Spark MLlib or Horovod, that training already uses the whole cluster. In that case, don't pass SparkTrials. Use Hyperopt's default Trials class, or leave out the trials argument altogether. Hyperopt then evaluates each trial on the driver, and from there the algorithm can start distributed training across the cluster's resources.

Sources24

5.What SparkTrials logs to MLflow

Running many trials in parallel would be hard to review without tracking. Databricks Runtime ML supports logging to MLflow from workers, and SparkTrials records the tuning as nested runs:

- Main (parent) run: the fmin() call itself. If a run is already active, SparkTrials logs to it and leaves it open when fmin() returns. If no run is active, SparkTrials creates a run, logs to it, and ends it before fmin() returns. - Child runs: each hyperparameter setting tested gets its own child run under the main run. Anything logged to MLflow from the workers is stored under the matching child run.

Databricks recommends wrapping each fmin() call in with mlflow.start_run():. That gives every tuning call its own main run and makes it easier to add extra tags, parameters or metrics to it. Inside the objective function, you don't manage runs yourself. A call such as mlflow.log_param there logs to the child run of that trial.

To see the effect of C in the notebook example, select the resulting runs in the MLflow UI, click Compare, and plot C on the X-axis against loss on the Y-axis in the scatter plot.

Sources2

6.When distributing trials pays off, and Hyperopt's status

Distributing trials doesn't speed up every job. Each trial costs a Spark job plus Hyperopt bookkeeping. In the Iris example, the objective runs so quickly that the cost of starting Spark jobs dominates the calculation time, and the distributed version takes *longer* than the single-machine one. The best-practices page draws the general lesson: when trials take only a few tens of seconds, the overhead from Hyperopt and Spark can dominate, and the speedup may be small or even zero. For typical real-world problems, the objective function does more work, and distributing it with SparkTrials is faster than tuning on one machine.

Two more points help when you read results. The search algorithms are stochastic, so the loss usually doesn't fall steadily from one run to the next. A reported loss of NaN usually means the objective returned NaN for that setting. It doesn't affect other runs, and adjusting the search space or the objective can prevent it. Because SparkTrials runs trials on worker nodes, the data the objective function trains on has to reach those workers too.

Sources132

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.SparkTrials is the right choice for any model you want to tune on a cluster, including Spark MLlib or Horovod models.Why is that wrong?

   SparkTrials is for single-machine models such as scikit-learn. Distributed algorithms already use the whole cluster, so use the default Trials class, which runs each trial from the driver.

   Covered in SparkTrials or Trials: match the class to the model

2. 2.Setting parallelism as high as possible always gives the best hyperparameters, just faster.Why is that wrong?

   For a fixed max_evals, higher parallelism means each new trial is proposed with fewer completed results, so an adaptive algorithm like TPE can produce worse results.

   Covered in How SparkTrials schedules trials: parallelism, timeout and the cluster

3. 3.Running SparkTrials on an autoscaling cluster lets the tuning job grow its parallelism as workers are added.Why is that wrong?

   Hyperopt picks the parallelism value when execution begins and can't use workers the cluster adds later. Don't use SparkTrials on autoscaling clusters.

   Covered in How SparkTrials schedules trials: parallelism, timeout and the cluster

4. 4.Tuning with the default Trials class logs every trial to MLflow automatically, just as SparkTrials does.Why is that wrong?

   Automatic MLflow logging comes with SparkTrials. With the Trials class, you must log trials to MLflow yourself.

   Covered in SparkTrials or Trials: match the class to the model

5. 5.Distributing trials with SparkTrials always finishes faster than single-machine tuning.Why is that wrong?

   For short trials, Spark and Hyperopt overhead can dominate, and the speedup can be small, zero or even negative.

   Covered in When distributing trials pays off, and Hyperopt's status

Practise it for real

Turn the single-machine SVC tuning run from the Databricks notebook into a distributed SparkTrials run and inspect its MLflow runs (needs Databricks Runtime ML 16.4 LTS or earlier, where Hyperopt is still included).

1. 1.In a notebook on a non-autoscaling cluster, load the Iris data and define the objective(C) function that returns {'loss': -accuracy, 'status': STATUS_OK}.

   Why: Hyperopt minimizes the loss, so a higher-is-better accuracy has to be negated.

   You should see: Calling objective(1.0) directly returns a dictionary with a negative loss.

2. 2.Set search_space = hp.lognormal('C', 0, 1.0) and algo=tpe.suggest, then run fmin(fn=objective, space=search_space, algo=algo, max_evals=16) without a trials argument.

   Why: This gives you the single-machine baseline before you distribute anything.

   You should see: fmin() prints a best value for C after 16 evaluations.

3. 3.Create spark_trials = SparkTrials() and rerun the same fmin() call with trials=spark_trials inside a with mlflow.start_run(): block.

   Why: SparkTrials evaluates the trials on workers, and the explicit run gives this fmin() call its own main MLflow run.

   You should see: The 16 trials run as Spark jobs and fmin() returns a best value for C.

4. 4.Open the notebook's Experiment panel, select the child runs, click Compare, and plot C on the X-axis against loss on the Y-axis.

   Why: Each trial was logged as a child run under the main run, so you can compare hyperparameter settings side by side.

   You should see: One parent run with 16 child runs, and a scatter plot of loss against C.

5. 5.Compare the wall-clock time of the single-machine run with the SparkTrials run.

   Why: The Iris objective is very fast, so Spark job overhead should dominate.

   You should see: The distributed run is likely no faster, and may be slower, than the single-machine run.