#NEWLINE# def fit_predict( training_data , fitting_data , tau = 1 , samples_per_job = 0 , save_results = True , show = False ) : #NEWLINE# #INDENT# from disco.worker.pipeline.worker import Worker , Stage #NEWLINE# from disco.core import Job , result_iterator #NEWLINE# from disco.core import Disco #NEWLINE# '\n    training_data - training samples\n    fitting_data - dataset to be fitted to training data.\n    tau - controls how quickly the weight of a training sample falls off with distance of its x(i) from the query point x.\n    samples_per_job - define a number of samples that will be processed in single mapreduce job. If 0, algorithm will calculate number of samples per job.\n    ' #NEWLINE# try : #NEWLINE# #INDENT# tau = float ( tau ) #NEWLINE# if ( tau <= 0 ) : #NEWLINE# #INDENT# raise Exception ( 'Parameter tau should be >= 0.' ) #NEWLINE# #UNINDENT# #UNINDENT# except ValueError : #NEWLINE# #INDENT# raise Exception ( 'Parameter tau should be numerical.' ) #NEWLINE# #UNINDENT# if ( fitting_data . params [ 'id_index' ] == ( - 1 ) ) : #NEWLINE# #INDENT# raise Exception ( 'Predict data should have id_index set.' ) #NEWLINE# #UNINDENT# job = Job ( worker = Worker ( save_results = save_results ) ) #NEWLINE# job . pipeline = [ ( 'split' , Stage ( 'map' , input_chain = fitting_data . params [ 'input_chain' ] , init = simple_init , process = map_predict ) ) ] #NEWLINE# job . params = fitting_data . params #NEWLINE# job . run ( name = 'lwlr_read_data' , input = fitting_data . params [ 'data_tag' ] ) #NEWLINE# samples = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# results = [ ] #NEWLINE# tau = float ( ( 2 * ( tau ** 2 ) ) ) #NEWLINE# counter = 0 #NEWLINE# for ( test_id , x ) in result_iterator ( job . wait ( show = show ) ) : #NEWLINE# #INDENT# if ( samples_per_job == 0 ) : #NEWLINE# #INDENT# if ( len ( x ) <= 100 ) : #NEWLINE# #INDENT# samples_per_job = 100 #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# samples_per_job = ( ( ( len ( x ) * ( - 25 ) ) / 900.0 ) + 53 ) #NEWLINE# #UNINDENT# #UNINDENT# samples [ test_id ] = ddfs #NEWLINE# if ( counter == samples_per_job ) : #NEWLINE# #INDENT# results . append ( _fit_predict ( training_data , samples , tau , save_results , show ) ) #NEWLINE# counter = 0 #NEWLINE# samples = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# #UNINDENT# counter += 1 #NEWLINE# #UNINDENT# if ( len ( samples ) > 0 ) : #NEWLINE# #INDENT# results . append ( _fit_predict ( training_data , samples , tau , save_results , show ) ) #NEWLINE# #UNINDENT# ddfs = Disco ( ) . ddfs #NEWLINE# ddfs . tag ( job . name , [ [ list ( ddfs . blobs ( tag ) ) [ 0 ] [ 0 ] ] for tag in results ] ) #NEWLINE# return [ ( 'tag://' + job . name ) ]