docstr: Protected train method to make wrapping of modes easier
          
docstr_tok: ["Protected","train","method","to","make","wrapping","of","modes","easier"]
code: def _fit(self, rdd: RDD, **kwargs):
      """Protected train method to make wrapping of modes easier"""
      self._master_network.compile(
          optimizer=get_optimizer(self.master_optimizer),
          loss=self.master_loss,
          metrics=self.master_metrics,
      )
      if self.mode in ["asynchronous", "hogwild"]:
          self.start_server()
      train_config = kwargs
      freq = self.frequency
      optimizer = deserialize_optimizer(self.master_optimizer)
      loss = self.master_loss
      metrics = self.master_metrics
      custom = self.custom_objects
  
      model_json = self._master_network.to_json()
      init = self._master_network.get_weights()
      parameters = rdd.context.broadcast(init)
  
      if self.mode in ["asynchronous", "hogwild"]:
          print(">>> Initialize workers")
          worker = AsynchronousSparkWorker(
              model_json,
              parameters,
              self.client,
              train_config,
              freq,
              optimizer,
              loss,
              metrics,
              custom,
          )
          print(">>> Distribute load")
          rdd.mapPartitions(worker.train).collect()
          print(">>> Async training complete.")
          new_parameters = self.client.get_parameters()
      elif self.mode == "synchronous":
          worker = SparkWorker(
              model_json,
              parameters,
              train_config,
              optimizer,
              loss,
              metrics,
              custom,
          )
          training_outcomes = rdd.mapPartitions(worker.train).collect()
          new_parameters = self._master_network.get_weights()
          number_of_sub_models = len(training_outcomes)
          for training_outcome in training_outcomes:
              grad, history = training_outcome
              self.training_histories.append(history)
              weighted_grad = divide_by(grad, number_of_sub_models)
              new_parameters = subtract_params(
                  new_parameters, weighted_grad
              )
          print(">>> Synchronous training complete.")
      else:
          raise ValueError("Unsupported mode {}".format(self.mode))
      self._master_network.set_weights(new_parameters)
      if self.mode in ["asynchronous", "hogwild"]:
          self.stop_server()
  
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 12 for calling the SparkModel obj's underlying member method code with example usage: obj._fit(rdd) without return types