docstr: Create Orca MPI Estimator
  :param model_creator: A model creator function that takes the parameter "config"
         and returns a model
  :param optimizer_creator: An optimizer creator function that has two parameters "model" and
         "config" and returns a optimizer.
  :param loss_creator: An creater function to return a loss.
         Default: None if loss computation is not needed.
  :param metrics: One or a list of validation metrics. Function(s) that computes the
         metrics between the output and target tensors are also supported.
  :param scheduler_creator: A scheduler creator function that has two parameters "optimizer"
         and "config" and returns a learning rate scheduler wrapping the optimizer.
         By default a scheduler will take effect automatically every epoch.
         Default: None if no scheduler is needed.
  :param config: A parameter config dict, that plays a role of
         configuration to create model, loss, optimizer, scheduler and data.
         Default: None if no config is needed.
  :param init_func: A function takes the parameter "config" to init the distributed
          environment for MPI if any.
  :param hosts: host information to be run distributedly.
         It can be None, 'all' or list of hostname/ip.
         If hosts is None, means it runs on single(self) node.
         If hosts is 'all', it will get executor hosts from current Spark Context.
         Default: None.
  :param workers_per_node: The number of workers on each node.
  :param env: Special environment should be passed to MPI environment.
docstr_tok: ["Create","Orca","MPI","Estimator",":","param","model_creator",":","A","model","creator","function","that","takes","the","parameter","``","config","''","and","returns","a","model",":","param","optimizer_creator",":","An","optimizer","creator","function","that","has","two","parameters","``","model","''","and","``","config","''","and","returns","a","optimizer",".",":","param","loss_creator",":","An","creater","function","to","return","a","loss",".","Default",":","None","if","loss","computation","is","not","needed",".",":","param","metrics",":","One","or","a","list","of","validation","metrics",".","Function","(","s",")","that","computes","the","metrics","between","the","output","and","target","tensors","are","also","supported",".",":","param","scheduler_creator",":","A","scheduler","creator","function","that","has","two","parameters","``","optimizer","''","and","``","config","''","and","returns","a","learning","rate","scheduler","wrapping","the","optimizer",".","By","default","a","scheduler","will","take","effect","automatically","every","epoch",".","Default",":","None","if","no","scheduler","is","needed",".",":","param","config",":","A","parameter","config","dict",",","that","plays","a","role","of","configuration","to","create","model",",","loss",",","optimizer",",","scheduler","and","data",".","Default",":","None","if","no","config","is","needed",".",":","param","init_func",":","A","function","takes","the","parameter","``","config","''","to","init","the","distributed","environment","for","MPI","if","any",".",":","param","hosts",":","host","information","to","be","run","distributedly",".","It","can","be","None",",","'all","'","or","list","of","hostname\/ip",".","If","hosts","is","None",",","means","it","runs","on","single","(","self",")","node",".","If","hosts","is","'all","'",",","it","will","get","executor","hosts","from","current","Spark","Context",".","Default",":","None",".",":","param","workers_per_node",":","The","number","of","workers","on","each","node",".",":","param","env",":","Special","environment","should","be","passed","to","MPI","environment","."]
code: def __init__(
      self,
      model_creator,
      optimizer_creator,
      loss_creator,
      metrics=None,
      scheduler_creator=None,
      config=None,
      init_func=None,  # Init the distributed environment for MPI if any
      hosts=None,
      workers_per_node=1,
      env=None,
  ):
      """
      Create Orca MPI Estimator
      :param model_creator: A model creator function that takes the parameter "config"
             and returns a model
      :param optimizer_creator: An optimizer creator function that has two parameters "model" and
             "config" and returns a optimizer.
      :param loss_creator: An creater function to return a loss.
             Default: None if loss computation is not needed.
      :param metrics: One or a list of validation metrics. Function(s) that computes the
             metrics between the output and target tensors are also supported.
      :param scheduler_creator: A scheduler creator function that has two parameters "optimizer"
             and "config" and returns a learning rate scheduler wrapping the optimizer.
             By default a scheduler will take effect automatically every epoch.
             Default: None if no scheduler is needed.
      :param config: A parameter config dict, that plays a role of
             configuration to create model, loss, optimizer, scheduler and data.
             Default: None if no config is needed.
      :param init_func: A function takes the parameter "config" to init the distributed
              environment for MPI if any.
      :param hosts: host information to be run distributedly.
             It can be None, 'all' or list of hostname/ip.
             If hosts is None, means it runs on single(self) node.
             If hosts is 'all', it will get executor hosts from current Spark Context.
             Default: None.
      :param workers_per_node: The number of workers on each node.
      :param env: Special environment should be passed to MPI environment.
      """
      self.dir = os.getcwd()
      self.mpi_runner = MPIRunner(
          hosts=hosts, processes_per_node=workers_per_node, env=env
      )
      with open("saved_mpi_estimator.pkl", "wb") as f:
          cloudpickle.dump(
              (
                  model_creator,
                  optimizer_creator,
                  loss_creator,
                  metrics,
                  scheduler_creator,
                  config,
                  init_func,
              ),
              f,
          )
      self.mpi_runner.scp_file("saved_mpi_estimator.pkl", self.dir)
      # Need to put mpi_train.py in the current directory so that the PYTHONPATH is the same.
      train_file = os.path.abspath(__file__ + "/../mpi_train.py")
      p = subprocess.Popen(["cp", train_file, self.dir])
      os.waitpid(p.pid, 0)
      self.mpi_runner.scp_file(train_file, self.dir)
  
imports: import os
  import subprocess
  import types
  import cloudpickle
  from pyspark.rdd import RDD
  from pyspark.sql import DataFrame
  from torch.utils.data import Dataset, DataLoader
  from bigdl.dllib.utils.log4Error import *
  from bigdl.dllib.utils.utils import get_node_ip
  from bigdl.orca.learn.mpi.mpi_runner import MPIRunner
  from bigdl.orca.learn.mpi.utils import *
label_desc: Use image node_id 1 to create a new MPIEstimator object  with example: obj = MPIEstimator(model_creator, optimizer_creator, loss_creator, metrics, scheduler_creator, config, init_func, hosts, workers_per_node, env)