#NEWLINE# def shuffle( job_name , input_file_list , output_file_list , callback ) : #NEWLINE# #INDENT# 'Shuffle mapreduce files using the shuffler service.\n\n  Args:\n    job_name: unique shuffle job name as string.\n    input_file_list: list of files api file names to shuffle. Files should be\n      in records format with serialized KeyValue protocol buffer as record.\n    output_file_list: list of files api file names to store shuffle result.\n      Files should not be finalized. They will be of records format with\n      serialized KeyValues protocol buffer as record.\n    callback: shuffle service call back specification. Can be either\n      url - the task in default queue with default parameters will be enqueued.\n      It can also be a dict with following keys:\n        url: url to call back\n        version: app version to call\n        method: HTTP method to use (POST or GET)\n        queue: queue name to enqueue a task in.\n  Raises:\n    ShufflerUnavailableError if shuffler service is not available.\n  ' #NEWLINE# if ( not available ( ) ) : #NEWLINE# #INDENT# raise ShufflerUnavailableError ( ) #NEWLINE# #UNINDENT# request = file_service_pb . ShuffleRequest ( ) #NEWLINE# response = file_service_pb . ShuffleResponse ( ) #NEWLINE# request . set_shuffle_name ( job_name ) #NEWLINE# if isinstance ( callback , dict ) : #NEWLINE# #INDENT# request . mutable_callback ( ) . set_url ( callback [ 'url' ] ) #NEWLINE# if ( 'version' in callback ) : #NEWLINE# #INDENT# request . mutable_callback ( ) . set_app_version_id ( callback [ 'version' ] ) #NEWLINE# #UNINDENT# if ( 'method' in callback ) : #NEWLINE# #INDENT# request . mutable_callback ( ) . set_method ( callback [ 'method' ] ) #NEWLINE# #UNINDENT# if ( 'queue' in callback ) : #NEWLINE# #INDENT# request . mutable_callback ( ) . set_queue ( callback [ 'queue' ] ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# request . mutable_callback ( ) . set_url ( callback ) #NEWLINE# #UNINDENT# request . set_shuffle_size_bytes ( 0 ) #NEWLINE# for file_name in input_file_list : #NEWLINE# #INDENT# shuffle_input = request . add_input ( ) #NEWLINE# shuffle_input . set_format ( file_service_pb . ShuffleEnums . RECORDS_KEY_VALUE_PROTO_INPUT ) #NEWLINE# shuffle_input . set_path ( file_name ) #NEWLINE# #UNINDENT# shuffle_output = request . mutable_output ( ) #NEWLINE# shuffle_output . set_format ( file_service_pb . ShuffleEnums . RECORDS_KEY_MULTI_VALUE_PROTO_OUTPUT ) #NEWLINE# for file_name in output_file_list : #NEWLINE# #INDENT# shuffle_output . add_path ( file_name ) #NEWLINE# #UNINDENT# files . _make_call ( 'Shuffle' , request , job_name )