#NEWLINE# def execute_job( job_id , data ) : #NEWLINE# #INDENT# cluster_id = job_ex_dict [ 'cluster_id' ] #NEWLINE# configs = data . get ( 'job_configs' , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# interface = data . get ( 'interface' , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# input_id = data . get ( 'input_id' , None ) #NEWLINE# output_id = data . get ( 'output_id' , None ) #NEWLINE# job_execution_info = data . get ( 'job_execution_info' , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# configs [ 'job_execution_info' ] = job_execution_info #NEWLINE# job_ex_dict = { #NEWLINE# #INDENT# 'input_id' : input_id , #NEWLINE# 'output_id' : output_id , #NEWLINE# 'job_id' : job_id , #NEWLINE# 'cluster_id' : cluster_id , #NEWLINE# 'info' : { #NEWLINE# #INDENT# 'status' : edp . JOB_STATUS_PENDING , #NEWLINE# #UNINDENT# } , #NEWLINE# 'job_configs' : configs , #NEWLINE# 'extra' : { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , #NEWLINE# 'interface' : interface , #NEWLINE# #UNINDENT# } #NEWLINE# job_execution = conductor . job_execution_create ( context . ctx ( ) , job_ex_dict ) #NEWLINE# context . set_current_job_execution_id ( job_execution . id ) #NEWLINE# if p . job_execution_requires_proxy_user ( job_execution ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# p . create_proxy_user_for_job_execution ( job_execution ) #NEWLINE# #UNINDENT# except ex . SaharaException as e : #NEWLINE# #INDENT# LOG . error ( _LE ( "Can't run job execution. (Reasons: {reason})" ) . format ( reason = e ) ) #NEWLINE# conductor . job_execution_destroy ( context . ctx ( ) , job_execution ) #NEWLINE# raise e #NEWLINE# #UNINDENT# #UNINDENT# api . OPS . run_edp_job ( job_execution . id ) #NEWLINE# return job_execution