code: def __init__(
      self, config=None, instances_per_node=1, num_ps=1, protocol="grpc"
  ):
      self.config = config
      self.num_ps = num_ps
      self.protocol = protocol
      ray_ctx = OrcaContext.get_ray_context()
      cores_per_instance = (
          ray_ctx.ray_node_cpu_cores // instances_per_node
      )
      self.num_instances = ray_ctx.num_ray_nodes * instances_per_node
      self.num_workers = self.num_instances - self.num_ps
      num_chief = 1  # Normally there would always be only one chief.
      invalidInputError(
          self.num_instances >= 2,
          "There should be at least two instances, one parameter server and one worker",
      )
      invalidInputError(
          self.num_workers > 0,
          "Do not have enough resources to launch {} parameter servers. "
          "Try to reduce num_ps".format(self.num_ps),
      )
  
      RemoteRunner = ray.remote(num_cpus=cores_per_instance)(RayWorker)
      self.remote_ps = [
          RemoteRunner.remote(
              task_index=i, task_type="ps", config=self.config
          )
          for i in range(num_ps)
      ]
      chief = RemoteRunner.remote(
          task_index=0, task_type="chief", config=self.config
      )
      workers = [
          RemoteRunner.remote(
              task_index=i, task_type="worker", config=self.config
          )
          for i in range(0, self.num_workers - num_chief)
      ]
      self.remote_workers = [chief] + workers
      self.remote_instances = self.remote_ps + self.remote_workers
      self.setup_workers()
  
imports: import time
  import argparse
  import tensorflow
  import os
  import sys
  import math
  import collections
  from tensorflow.python.client import timeline
  import json
  from tensorflow.python.ops import partitioned_variables
  import ray
  from bigdl.dllib.utils.log4Error import *
  from bigdl.orca import init_orca_context, stop_orca_context, OrcaContext
  from bigdl.orca.data.utils import process_spark_xshards
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards
  from bigdl.friesian.feature import FeatureTable
label_desc: Use image node_id 1 to create a new RayDeepRecCluster object  with example: obj = RayDeepRecCluster(config, instances_per_node, num_ps, protocol)