#NEWLINE# def run_reducer( self , step_num = 0 ) : #NEWLINE# #INDENT# "Run the reducer for the given step.\n\n        :type step_num: int\n        :param step_num: which step to run (0-indexed)\n\n        If we encounter a line that can't be decoded by our input protocol,\n        or a tuple that can't be encoded by our output protocol, we'll\n        increment a counter rather than raising an exception. If\n        --strict-protocols is set, then an exception is raised\n\n        Called from :py:meth:`run`. You'd probably only want to call this\n        directly from automated tests.\n        " #NEWLINE# steps = self . steps ( ) #NEWLINE# if ( not ( 0 <= step_num < len ( steps ) ) ) : #NEWLINE# #INDENT# raise ValueError ( ( 'Out-of-range step: %d' % step_num ) ) #NEWLINE# #UNINDENT# step = steps [ step_num ] #NEWLINE# reducer = step [ 'reducer' ] #NEWLINE# reducer_init = step [ 'reducer_init' ] #NEWLINE# reducer_final = step [ 'reducer_final' ] #NEWLINE# if ( reducer is None ) : #NEWLINE# #INDENT# raise ValueError ( ( 'No reducer in step %d' % step_num ) ) #NEWLINE# #UNINDENT# ( read_lines , write_line ) = self . _wrap_protocols ( step_num , 'reducer' ) #NEWLINE# if reducer_init : #NEWLINE# #INDENT# for ( out_key , out_value ) in ( reducer_init ( ) or ( ) ) : #NEWLINE# #INDENT# write_line ( out_key , out_value ) #NEWLINE# #UNINDENT# #UNINDENT# for ( key , kv_pairs ) in itertools . groupby ( read_lines ( ) , key = ( lambda k_v : k_v [ 0 ] ) ) : #NEWLINE# #INDENT# values = ( v for ( k , v ) in kv_pairs ) #NEWLINE# for ( out_key , out_value ) in ( reducer ( key , values ) or ( ) ) : #NEWLINE# #INDENT# write_line ( out_key , out_value ) #NEWLINE# #UNINDENT# #UNINDENT# if reducer_final : #NEWLINE# #INDENT# for ( out_key , out_value ) in ( reducer_final ( ) or ( ) ) : #NEWLINE# #INDENT# write_line ( out_key , out_value )