#NEWLINE# def _train( self , rdd , nb_epoch = 10 , batch_size = 32 , verbose = 0 , validation_split = 0.1 , master_url = 'localhost:5000' ) : #NEWLINE# #INDENT# '\n        Protected train method to make wrapping of modes easier\n        ' #NEWLINE# if ( self . mode in [ 'asynchronous' , 'hogwild' ] ) : #NEWLINE# #INDENT# self . start_server ( ) #NEWLINE# #UNINDENT# yaml = self . master_network . to_yaml ( ) #NEWLINE# train_config = self . get_train_config ( nb_epoch , batch_size , verbose , validation_split ) #NEWLINE# if ( self . mode in [ 'asynchronous' , 'hogwild' ] ) : #NEWLINE# #INDENT# worker = AsynchronousSparkWorker ( yaml , train_config , self . frequency , master_url ) #NEWLINE# rdd . mapPartitions ( worker . train ) . collect ( ) #NEWLINE# new_parameters = get_server_weights ( master_url ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( self . mode == 'synchronous' ) : #NEWLINE# #INDENT# init = self . master_network . get_weights ( ) #NEWLINE# parameters = self . spark_context . broadcast ( init ) #NEWLINE# worker = SparkWorker ( yaml , parameters , train_config ) #NEWLINE# deltas = rdd . mapPartitions ( worker . train ) . collect ( ) #NEWLINE# new_parameters = self . master_network . get_weights ( ) #NEWLINE# for delta in deltas : #NEWLINE# #INDENT# constraints = self . master_network . constraints #NEWLINE# new_parameters = self . optimizer . get_updates ( self . weights , constraints , delta ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# self . master_network . set_weights ( new_parameters ) #NEWLINE# if ( self . mode in [ 'asynchronous' , 'hogwild' ] ) : #NEWLINE# #INDENT# self . stop_server ( )