#NEWLINE# def job_start( self , p ) : #NEWLINE# #INDENT# self . createJob ( p ) #NEWLINE# if ( not self . job_machine . remote . isdir ( p [ 'workdir' ] ) ) : #NEWLINE# #INDENT# self . job_machine . remote . mkdir ( p [ 'workdir' ] ) #NEWLINE# #UNINDENT# self . set_job_machine ( p , self . job_machine ) #NEWLINE# self . job . reset ( ) #NEWLINE# config = self . get_hadoop_config ( self . job_machine ) #NEWLINE# command = ( 'jar %s' % config [ 'streaming.jar' ] ) #NEWLINE# generics = '' #NEWLINE# arguments = '' #NEWLINE# if ( '://' not in p [ 'input' ] ) : #NEWLINE# #INDENT# p [ 'input' ] = self . add_prefix ( p [ 'input' ] , self . job_machine ) #NEWLINE# #UNINDENT# if ( '://' not in p [ 'output' ] ) : #NEWLINE# #INDENT# p [ 'output' ] = self . add_prefix ( p [ 'output' ] , self . job_machine ) #NEWLINE# #UNINDENT# arguments += ( ' -input %s -output %s' % ( p [ 'input' ] , p [ 'output' ] ) ) #NEWLINE# if ( self . localMapper != None ) : #NEWLINE# #INDENT# tempfile = self . job_machine . remote . send_command ( 'mktemp' ) . strip ( ) #NEWLINE# result = self . job_machine . sendfile ( self . localMapper . name , tempfile ) #NEWLINE# mapperFileName = os . path . split ( tempfile ) [ 1 ] #NEWLINE# p [ 'files' ] . append ( result ) #NEWLINE# arguments += ( ' -mapper %s' % mapperFileName ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# arguments += ' -mapper org.apache.hadoop.mapred.lib.IdentityMapper' #NEWLINE# #UNINDENT# if ( self . localCombiner != None ) : #NEWLINE# #INDENT# tempfile = self . job_machine . remote . send_command ( 'mktemp' ) . strip ( ) #NEWLINE# result = self . job_machine . sendfile ( self . localCombiner . name , tempfile ) #NEWLINE# combinerFileName = os . path . split ( tempfile ) [ 1 ] #NEWLINE# p [ 'files' ] . append ( tempfile ) #NEWLINE# arguments += ( ' -combiner %s' % combinerFileName ) #NEWLINE# #UNINDENT# if ( self . localReducer != None ) : #NEWLINE# #INDENT# tempfile = self . job_machine . remote . send_command ( 'mktemp' ) . strip ( ) #NEWLINE# result = self . job_machine . sendfile ( self . localReducer . name , tempfile ) #NEWLINE# reducerFileName = os . path . split ( tempfile ) [ 1 ] #NEWLINE# p [ 'files' ] . append ( tempfile ) #NEWLINE# arguments += ( ' -reducer %s' % reducerFileName ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# arguments += ' -numReduceTasks 0' #NEWLINE# #UNINDENT# for var in p [ 'envVars' ] : #NEWLINE# #INDENT# arguments += ( ' -cmdenv ' + var ) #NEWLINE# #UNINDENT# for cacheArchive in p [ 'cacheArchives' ] : #NEWLINE# #INDENT# arguments += ( ' -cacheArchive %s' % cacheArchive ) #NEWLINE# #UNINDENT# if p [ 'files' ] : #NEWLINE# #INDENT# generics += ( ' -files ' + ',' . join ( p [ 'files' ] ) ) #NEWLINE# #UNINDENT# arguments = ( ( command + generics ) + arguments ) #NEWLINE# result = self . call_hadoop ( arguments , p [ 'workdir' ] , p [ 'job_identifier' ] , self . job_machine ) #NEWLINE# return p