code: def _fit(self, rdd, epochs, batch_size, verbose, validation_split):
          
          self._master_network.compile(optimizer=self.master_optimizer,
                                       loss=self.master_loss,
                                       metrics=self.master_metrics)
          if self.mode in ['asynchronous', 'hogwild']:
              self.start_server()
          train_config = self.get_train_config(
              epochs, batch_size, verbose, validation_split)
          mode = self.parameter_server_mode
          freq = self.frequency
          optimizer = self.master_optimizer
          loss = self.master_loss
          metrics = self.master_metrics
          custom = self.custom_objects
  
          yaml = self._master_network.to_yaml()
          init = self._master_network.get_weights()
          parameters = rdd.context.broadcast(init)
  
          if self.mode in ['asynchronous', 'hogwild']:
              print('>>> Initialize workers')
              worker = AsynchronousSparkWorker(
                  yaml, parameters, mode, 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(yaml, parameters, train_config,
                                   optimizer, loss, metrics, custom)
              gradients = rdd.mapPartitions(worker.train).collect()
              new_parameters = self._master_network.get_weights()
              for grad in gradients:  # simply accumulate gradients one by one
                  new_parameters = subtract_params(new_parameters, 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()
docs: Protected train method to make wrapping of modes easier