#NEWLINE# def run( self , job_name , mapper_spec , reducer_spec , input_reader_spec , output_writer_spec = None , mapper_params = None , reducer_params = None , shards = None , combiner_spec = None ) : #NEWLINE# #INDENT# map_pipeline = ( yield MapPipeline ( job_name , mapper_spec , input_reader_spec , params = mapper_params , shards = shards ) ) #NEWLINE# shuffler_pipeline = ( yield ShufflePipeline ( job_name , map_pipeline ) ) #NEWLINE# reducer_pipeline = ( yield ReducePipeline ( job_name , reducer_spec , output_writer_spec , reducer_params , shuffler_pipeline , combiner_spec = combiner_spec ) ) #NEWLINE# with pipeline . After ( reducer_pipeline ) : #NEWLINE# #INDENT# all_temp_files = ( yield pipeline_common . Extend ( map_pipeline , shuffler_pipeline ) ) #NEWLINE# ( yield CleanupPipeline ( all_temp_files ) ) #NEWLINE# #UNINDENT# ( yield pipeline_common . Return ( reducer_pipeline ) )