#NEWLINE# def fit( dataset , n_clusters = 5 , max_iterations = 10 , random_state = None , save_results = True , show = False ) : #NEWLINE# #INDENT# '\n    Optimize k-clustering for `iterations` iterations with cluster\n    center definitions as given in `center`.\n    ' #NEWLINE# from disco.job import Job #NEWLINE# from disco.worker.pipeline.worker import Worker , Stage #NEWLINE# from disco.core import result_iterator #NEWLINE# try : #NEWLINE# #INDENT# n_clusters = int ( n_clusters ) #NEWLINE# max_iterations = int ( max_iterations ) #NEWLINE# if ( n_clusters < 2 ) : #NEWLINE# #INDENT# raise Exception ( 'Parameter n_clusters should be greater than 1.' ) #NEWLINE# #UNINDENT# if ( max_iterations < 1 ) : #NEWLINE# #INDENT# raise Exception ( 'Parameter max_iterations should be greater than 0.' ) #NEWLINE# #UNINDENT# #UNINDENT# except ValueError : #NEWLINE# #INDENT# raise Exception ( 'Parameters should be numerical.' ) #NEWLINE# #UNINDENT# job = Job ( worker = Worker ( save_results = random_state ) ) #NEWLINE# job . pipeline = [ ( 'split' , Stage ( 'kmeans_init_map' , input_chain = dataset . params [ 'input_chain' ] , init = map_init , process = random_init_map ) ) , ( 'group_label' , Stage ( 'kmeans_init_reduce' , process = estimate_reduce , init = simple_init , combine = True ) ) ] #NEWLINE# job . params = dict ( ( dataset . params . items ( ) + mean_point_center . items ( ) ) ) #NEWLINE# job . params [ 'seed' ] = random_state #NEWLINE# job . params [ 'k' ] = n_clusters #NEWLINE# job . run ( input = dataset . params [ 'data_tag' ] , name = 'kmeans_init' ) #NEWLINE# init = job . wait ( show = show ) #NEWLINE# centers = [ ( i , c ) for ( i , c ) in result_iterator ( init ) ] #NEWLINE# for j in range ( max_iterations ) : #NEWLINE# #INDENT# job = Job ( worker = Worker ( save_results = save_results ) ) #NEWLINE# job . params = dict ( ( dataset . params . items ( ) + mean_point_center . items ( ) ) ) #NEWLINE# job . params [ 'k' ] = n_clusters #NEWLINE# job . params [ 'centers' ] = centers #NEWLINE# job . pipeline = [ ( 'split' , Stage ( ( 'kmeans_map_iter_%s' % ( ( j + 1 ) , ) ) , input_chain = dataset . params [ 'input_chain' ] , process = estimate_map , init = simple_init ) ) , ( 'group_label' , Stage ( ( 'kmeans_reduce_iter_%s' % ( ( j + 1 ) , ) ) , process = estimate_reduce , init = simple_init , combine = True ) ) ] #NEWLINE# job . run ( input = dataset . params [ 'data_tag' ] , name = ( 'kmeans_iter_%d' % ( ( j + 1 ) , ) ) ) #NEWLINE# fitmodel_url = job . wait ( show = show ) #NEWLINE# centers = [ ( i , c ) for ( i , c ) in result_iterator ( fitmodel_url ) ] #NEWLINE# #UNINDENT# return { #NEWLINE# #INDENT# 'kmeans_fitmodel' : fitmodel_url , #NEWLINE# #UNINDENT# }