我正在根據(jù)序列號(hào)以并行方式運(yùn)行作業(yè)。我正在采取每一項(xiàng)工作的狀態(tài)， 如 - 成功或失敗。然后，在獲得每個(gè)作業(yè)的狀態(tài)后，我正在發(fā)送帶有每個(gè)作業(yè)狀態(tài)的郵件。但是郵件是在整個(gè)過(guò)程完成后生成的。但我希望如果任何作業(yè)失敗，該過(guò)程將停止在那里，郵件將生成。你能幫我怎么做嗎？我正在運(yùn)行的代碼： df_mail_final = pd.DataFrame() df_mail_final1 = pd.DataFrame() '''Getting the status of every job''' for m_job in df_main4.master_job.unique(): list_df = [] dict_mail = OrderedDict() temp_df1 = df_main4[df_main4['master_job'] == m_job].copy() temp_df1['duration'] = pd.to_datetime(temp_df1['end_time'].unique()[-1]) - pd.to_datetime(temp_df1['start_time'].unique()[0]) temp_df1['duration'] = temp_df1['duration'].replace('0 days' ,'') status_list = temp_df1.status.unique() if(0 in status_list): dict_mail['Master Job Name'] = m_job idx = temp_df1['status'] == 0 dict_mail['Execution_Seq'] = temp_df1.loc[idx]["exec_seq"].unique()[0] dict_mail['Start_time'] = temp_df1.loc[idx]["start_time"].unique()[0] dict_mail['End_time'] = temp_df1.loc[idx]["end_time"].unique()[-1] dict_mail['Status'] = 'Failed' dict_mail['Duration'] = temp_df1.loc[idx]["duration"].unique()[-1] dict_mail['Reason'] = temp_df1.loc[idx]["error_msg"].unique()[0] dict_mail['Function_Name'] = temp_df1.loc[idx]["error_func"].unique()[0] list_df.append(dict_mail) df_mail = pd.DataFrame(list_df) if(0 not in status_list): print(m_job) dict_mail['Master Job Name'] = m_job dict_mail['Execution_Seq'] = temp_df1.exec_seq.unique()[0] dict_mail['Start_time'] = temp_df1.start_time.unique()[0] dict_mail['End_time'] = temp_df1.end_time.unique()[-1] dict_mail['Status'] = 'Success'
1 回答
慕標(biāo)5832272
TA貢獻(xiàn)1966條經(jīng)驗(yàn) 獲得超4個(gè)贊
分享如何實(shí)現(xiàn)此作業(yè)的要點(diǎn)/系統(tǒng)設(shè)計(jì)邏輯
def my_parallel_job(*args, **kwargs):
# do your stuff here
pass
def parallel_job_wrapper(*args, **kwargs):
try:
my_parallel_job(*args, **kwargs)
# if errors following will not run
return "success"
except:
# if errors comes
return "fail"
def main(*args, **kwargs):
# call you parallel jobs from here
p1 = parallel_job_wrapper(*args, **kwargs)
# preferably you are using something like python's multithreading pool methods
在上面的代碼中，第二個(gè)函數(shù)是充當(dāng)緩沖，以防第一個(gè)函數(shù)發(fā)生任何故障。這可確保即使任何并行作業(yè)失敗，您也不會(huì)停止。main
添加回答
舉報(bào)
0/150
提交
取消