class_docstr: Tune driver
docstr: Prepare the train function for ray tune
  :param data: input data
  :param model_builder: model create function
  :param metric_name: the rewarding metric name
  :param metric_func: customized metric func
  :param mode: metric mode
  :param validation_data: validation data
  :param mc: if calculate uncertainty
  :param remote_dir: checkpoint will be uploaded to remote_dir in hdfs
  
  :return: the train function
docstr_tok: ["Prepare","the","train","function","for","ray","tune",":","param","data",":","input","data",":","param","model_builder",":","model","create","function",":","param","metric_name",":","the","rewarding","metric","name",":","param","metric_func",":","customized","metric","func",":","param","mode",":","metric","mode",":","param","validation_data",":","validation","data",":","param","mc",":","if","calculate","uncertainty",":","param","remote_dir",":","checkpoint","will","be","uploaded","to","remote_dir","in","hdfs",":","return",":","the","train","function"]
code: def _prepare_train_func(
      data,
      model_builder,
      validation_data=None,
      metric_name=None,
      metric_func=None,
      mode=None,
      mc=False,
      remote_dir=None,
      resources_per_trial=None,
      feature_cols=None,
      label_cols=None,
  ):
      """
      Prepare the train function for ray tune
      :param data: input data
      :param model_builder: model create function
      :param metric_name: the rewarding metric name
      :param metric_func: customized metric func
      :param mode: metric mode
      :param validation_data: validation data
      :param mc: if calculate uncertainty
      :param remote_dir: checkpoint will be uploaded to remote_dir in hdfs
  
      :return: the train function
      """
      from pyspark.sql import DataFrame
  
      if isinstance(data, DataFrame):
          from bigdl.orca.learn.utils import dataframe_to_xshards
          from bigdl.dllib.utils.common import get_node_and_core_number
          from bigdl.orca.data.utils import process_spark_xshards
  
          num_workers, _ = get_node_and_core_number()
          spark_xshards, val_spark_xshards = dataframe_to_xshards(
              data,
              validation_data=validation_data,
              feature_cols=feature_cols,
              label_cols=label_cols,
              mode="fit",
              num_workers=num_workers,
          )
          ray_xshards = process_spark_xshards(
              spark_xshards, num_workers=num_workers
          )
          val_ray_xshards = process_spark_xshards(
              val_spark_xshards, num_workers=num_workers
          )
          data_ref = ray_xshards.get_partition_refs()
          validation_data_ref = val_ray_xshards.get_partition_refs()
      else:
          data_ref = ray.put(data)
          validation_data_ref = ray.put(validation_data)
  
      def train_func(config):
          if isinstance(data_ref, list):
              train_data = get_data_from_part_refs(data_ref)
              val_data = get_data_from_part_refs(validation_data_ref)
          else:
              train_data = ray.get(data_ref)
              val_data = ray.get(validation_data_ref)
          config = convert_bayes_configs(config).copy()
          trial_model = model_builder.build(config)
  
          # no need to call build since it is called the first time fit_eval is called.
          # callbacks = [TuneCallback(tune_reporter)]
          # fit model
          best_reward = None
          for i in range(1, 101):
              result = trial_model.fit_eval(
                  data=train_data,
                  validation_data=val_data,
                  mc=mc,
                  metric=metric_name,
                  metric_func=metric_func,
                  resources_per_trial=resources_per_trial,
                  **config,
              )
              reward = result[metric_name]
              checkpoint_filename = "best.ckpt"
  
              # Save best reward iteration
              if mode == "max":
                  has_best_reward = (
                      best_reward is None or reward > best_reward
                  )
              elif mode == "min":
                  has_best_reward = (
                      best_reward is None or reward < best_reward
                  )
              else:
                  has_best_reward = True
  
              if has_best_reward:
                  best_reward = reward
                  trial_model.save(checkpoint_filename)
                  # Save to hdfs
                  if remote_dir is not None:
                      put_ckpt_hdfs(remote_dir, checkpoint_filename)
  
              report_dict = {
                  "training_iteration": i,
                  "checkpoint": checkpoint_filename,
                  "best_" + metric_name: best_reward,
              }
              report_dict.update(result)
              tune.report(**report_dict)
  
      return train_func
  
imports: import ray
  from ray import tune
  import os
  from bigdl.orca.automl.search.base import SearchEngine, TrialOutput, GoodError
  from bigdl.orca.automl.search.parameters import DEFAULT_LOGGER_NAME, DEFAULT_METRIC_NAME
  from bigdl.orca.automl.search.ray_tune.utils import convert_bayes_configs
  from bigdl.orca.automl.search.utils import get_ckpt_hdfs, put_ckpt_hdfs
  from ray.tune import Stopper
  from bigdl.orca.automl.model.abstract import ModelBuilder
  from bigdl.orca.data import ray_xshards
  from ray.tune.progress_reporter import TrialProgressCallback
  from bigdl.dllib.utils.log4Error import *
label_desc: Use image node_id 14 for calling the RayTuneSearchEngine obj's underlying member method code with example usage: obj._prepare_train_func(data, model_builder, validation_data, metric_name, metric_func, mode, mc, remote_dir, resources_per_trial, feature_cols, label_cols) and returns: train_func