#NEWLINE# def main( cl_args = None ) : #NEWLINE# #INDENT# usage = 'usage: %prog CLUSTER_ID [options] "command string"' #NEWLINE# description = 'Run a command on the master and all slaves of an EMR cluster. Store stdout/stderr for results in OUTPUT_DIR.' #NEWLINE# option_parser = OptionParser ( usage = usage , description = description ) #NEWLINE# option_parser . add_option ( '-o' , '--output-dir' , dest = 'output_dir' , default = None , help = 'Specify an output directory (default: CLUSTER_ID)' ) #NEWLINE# _add_basic_opts ( option_parser ) #NEWLINE# _add_emr_connect_opts ( option_parser ) #NEWLINE# scrape_options_into_new_groups ( MRJob ( ) . all_option_groups ( ) , { #NEWLINE# #INDENT# option_parser : ( 'ec2_key_pair_file' , 'ssh_bin' ) , #NEWLINE# #UNINDENT# } ) #NEWLINE# _alphabetize_options ( option_parser ) #NEWLINE# ( options , args ) = option_parser . parse_args ( cl_args ) #NEWLINE# MRJob . set_up_logging ( quiet = options . quiet , verbose = options . verbose ) #NEWLINE# runner_kwargs = options . __dict__ . copy ( ) #NEWLINE# for unused_arg in ( 'output_dir' , 'quiet' , 'verbose' ) : #NEWLINE# #INDENT# del runner_kwargs [ unused_arg ] #NEWLINE# #UNINDENT# if ( len ( args ) < 2 ) : #NEWLINE# #INDENT# option_parser . print_help ( ) #NEWLINE# sys . exit ( 1 ) #NEWLINE# #UNINDENT# ( cluster_id , cmd_string ) = args [ : 2 ] #NEWLINE# cmd_args = shlex_split ( cmd_string ) #NEWLINE# output_dir = os . path . abspath ( ( options . output_dir or output_dir ) ) #NEWLINE# with EMRJobRunner ( cluster_id = cluster_id , ** runner_kwargs ) as runner : #NEWLINE# #INDENT# _run_on_all_nodes ( runner , output_dir , cmd_args )