#NEWLINE# def fit( dataset , save_results = True , show = False ) : #NEWLINE# #INDENT# from disco.worker.pipeline.worker import Worker , Stage #NEWLINE# from disco.core import Job #NEWLINE# job = Job ( worker = Worker ( save_results = save_results ) ) #NEWLINE# job . pipeline = [ ( 'split' , Stage ( 'map' , input_chain = dataset . params [ 'input_chain' ] , init = simple_init , process = map_fit ) ) , ( 'group_all' , Stage ( 'reduce' , init = simple_init , process = reduce_fit , combine = True ) ) ] #NEWLINE# job . params = dataset . params #NEWLINE# job . run ( name = 'linreg_fit' , input = dataset . params [ 'data_tag' ] ) #NEWLINE# fitmodel_url = job . wait ( show = show ) #NEWLINE# return { #NEWLINE# #INDENT# 'linreg_fitmodel' : fitmodel_url , #NEWLINE# #UNINDENT# }