code: def __init__(
      self,
      config,
      model_creator,
      loss_creator=None,
      eval_metrics_creator=None,
      validation_metrics_creator=None,
      num_workers=None,
      num_servers=None,
      runner_cores=None,
  ):
      ray_ctx = OrcaRayContext.get()
      if not num_workers:
          num_workers = ray_ctx.num_ray_nodes
      self.config = {} if config is None else config
      invalidInputError(
          isinstance(config, dict), "config must be a dict"
      )
      for param in ["optimizer", "optimizer_params", "log_interval"]:
          invalidInputError(
              param in config, param + " must be specified in config"
          )
      self.model_creator = model_creator
      self.loss_creator = loss_creator
      self.validation_metrics_creator = validation_metrics_creator
      self.eval_metrics_creator = eval_metrics_creator
      self.num_workers = num_workers
      self.num_servers = (
          num_servers if num_servers else self.num_workers
      )
  
      # Generate actor class
      # Add a dummy custom resource: _mxnet_worker and _mxnet_server to diff worker from server
      # if runner_cores is specified so that we can place one worker and one server on a node
      # for better performance.
      Worker = (
          ray.remote(
              num_cpus=runner_cores, resources={"_mxnet_worker": 1}
          )(MXNetRunner)
          if runner_cores
          else ray.remote(MXNetRunner)
      )
      Server = (
          ray.remote(
              num_cpus=runner_cores, resources={"_mxnet_server": 1}
          )(MXNetRunner)
          if runner_cores
          else ray.remote(MXNetRunner)
      )
  
      # Start runners: workers followed by servers
      self.workers = [Worker.remote() for i in range(self.num_workers)]
      self.servers = [Server.remote() for i in range(self.num_servers)]
      self.runners = self.workers + self.servers
  
      env = {
          "DMLC_PS_ROOT_URI": str(get_host_ip()),
          "DMLC_PS_ROOT_PORT": str(find_free_port()),
          "DMLC_NUM_SERVER": str(self.num_servers),
          "DMLC_NUM_WORKER": str(self.num_workers),
      }
      envs = []
      for i in range(self.num_workers):
          current_env = env.copy()
          current_env["DMLC_ROLE"] = "worker"
          envs.append(current_env)
      for i in range(self.num_servers):
          current_env = env.copy()
          current_env["DMLC_ROLE"] = "server"
          envs.append(current_env)
  
      env["DMLC_ROLE"] = "scheduler"
      modified_env = os.environ.copy()
      modified_env.update(env)
      # Need to contain system env to run bash
      # TODO: Need to kill this process manually?
      subprocess.Popen(
          ["python", "-c", "import mxnet"],
          shell=False,
          env=modified_env,
      )
  
      ray.get(
          [
              runner.setup_distributed.remote(
                  envs[i],
                  self.config,
                  self.model_creator,
                  self.loss_creator,
                  self.validation_metrics_creator,
                  self.eval_metrics_creator,
              )
              for i, runner in enumerate(self.runners)
          ]
      )
  
imports: import os
  import types
  import itertools
  import subprocess
  import ray
  from dmlc_tracker.tracker import get_host_ip
  from bigdl.orca.data.utils import partitions_get_data_label, process_spark_xshards
  from bigdl.orca.ray import OrcaRayContext
  from bigdl.orca.learn.mxnet.mxnet_runner import MXNetRunner
  from bigdl.orca.learn.mxnet.utils import find_free_port
  from bigdl.orca.learn.ray_estimator import Estimator
  from bigdl.dllib.utils.log4Error import *
label_desc: Use image node_id 1 to create a new MXNetEstimator object from inherited base classes: OrcaRayEstimator with example: obj = MXNetEstimator(config, model_creator, loss_creator, eval_metrics_creator, validation_metrics_creator, num_workers, num_servers, runner_cores)