docstr: SparkModel
  
  Base class for distributed training on RDDs. Spark model takes a Keras
  model as master network, an optimization scheme, a parallelisation mode
  and an averaging frequency.
  
  :param model: Compiled Keras model
  :param mode: String, choose from `asynchronous`, `synchronous` and `hogwild`
  :param frequency: String, either `epoch` or `batch`
  :param parameter_server_mode: String, either `http` or `socket`
  :param num_workers: int, number of workers used for training (defaults to None)
  :param custom_objects: Keras custom objects
  :param batch_size: batch size used for training and inference
  :param port: port used in case of 'http' parameter server mode
docstr_tok: ["SparkModel","Base","class","for","distributed","training","on","RDDs",".","Spark","model","takes","a","Keras","model","as","master","network",",","an","optimization","scheme",",","a","parallelisation","mode","and","an","averaging","frequency",".",":","param","model",":","Compiled","Keras","model",":","param","mode",":","String",",","choose","from","`","asynchronous","`",",","`","synchronous","`","and","`","hogwild","`",":","param","frequency",":","String",",","either","`","epoch","`","or","`","batch","`",":","param","parameter_server_mode",":","String",",","either","`","http","`","or","`","socket","`",":","param","num_workers",":","int",",","number","of","workers","used","for","training","(","defaults","to","None",")",":","param","custom_objects",":","Keras","custom","objects",":","param","batch_size",":","batch","size","used","for","training","and","inference",":","param","port",":","port","used","in","case","of","'http","'","parameter","server","mode"]
code: def __init__(
      self,
      model,
      mode="asynchronous",
      frequency="epoch",
      parameter_server_mode="http",
      num_workers=None,
      custom_objects=None,
      batch_size=32,
      port=4000,
      *args,
      **kwargs,
  ):
      """SparkModel
  
      Base class for distributed training on RDDs. Spark model takes a Keras
      model as master network, an optimization scheme, a parallelisation mode
      and an averaging frequency.
  
      :param model: Compiled Keras model
      :param mode: String, choose from `asynchronous`, `synchronous` and `hogwild`
      :param frequency: String, either `epoch` or `batch`
      :param parameter_server_mode: String, either `http` or `socket`
      :param num_workers: int, number of workers used for training (defaults to None)
      :param custom_objects: Keras custom objects
      :param batch_size: batch size used for training and inference
      :param port: port used in case of 'http' parameter server mode
      """
      self._training_histories = []
      self._master_network = model
      if not hasattr(model, "loss"):
          raise Exception(
              "Compile your Keras model before initializing an Elephas model with it"
          )
      metrics = model.compiled_metrics._metrics
      loss = model.loss
      optimizer = serialize_optimizer(model.optimizer)
  
      if custom_objects is None:
          custom_objects = {}
      if metrics is None:
          metrics = ["accuracy"]
      self.mode = mode
      self.frequency = frequency
      self.num_workers = num_workers
      self.weights = self._master_network.get_weights()
      self.pickled_weights = None
      self.master_optimizer = optimizer
      self.master_loss = loss
      self.master_metrics = metrics
      self.custom_objects = custom_objects
      self.parameter_server_mode = parameter_server_mode
      self.batch_size = batch_size
      self.port = port
      self.kwargs = kwargs
  
      self.serialized_model = model_to_dict(model)
      if self.mode is not "synchronous":
          factory = ClientServerFactory.get_factory(
              self.parameter_server_mode
          )
          self.parameter_server = factory.create_server(
              self.serialized_model,
              self.port,
              self.mode,
              custom_objects=self.custom_objects,
          )
          self.client = factory.create_client(self.port)
  
imports: import json
  from functools import partial
  from itertools import tee
  from typing import Union
  import h5py
  import numpy
  import pyspark
  from pyspark import RDD
  from tensorflow.keras.models import load_model
  from tensorflow.keras.models import model_from_json
  from tensorflow.keras.optimizers import get
  from tensorflow.keras.optimizers import serialize, deserialize
  from .mllib import to_matrix, from_matrix, to_vector, from_vector
  from .parameter.factory import ClientServerFactory
  from .utils import lp_to_simple_rdd, to_simple_rdd
  from .utils import model_to_dict
  from .utils import subtract_params, divide_by
  from .worker import AsynchronousSparkWorker, SparkWorker
label_desc: Use image node_id 1 to create a new SparkModel object from inherited base classes: object with example: obj = SparkModel(model, mode, frequency, parameter_server_mode, num_workers, custom_objects, batch_size, port)