实时任务修改状态表-增加重试次数

Pyspark_job/utils/templates.py

@@ -673,15 +673,20 @@ class Templates(object):
                 wx_msg = f"站点： {self.site_name}, {self.date_type}, {self.date_info} asin详情实时消费数据到redis准备工作已完成，可以开启详情爬取！"
             else:
                 pass
-            try:
             sql = f"UPDATE selection.workflow_progress SET {kafka_field}=3, updated_at=CURRENT_TIMESTAMP where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}' and page='asin详情'"
+            for retry in range(5):
+                try:
                     DBUtil.exec_sql('mysql', 'us', sql)
                     CommonUtil.send_wx_msg(wx_users, f"asin详情kafka消费", wx_msg)
+                    break
                 except Exception as e:
-                print(e, traceback.format_exc())
+                    print(f"UPDATE workflow_progress 失败（第{retry + 1}次），等待10s重试", e, traceback.format_exc())
+                    if retry == 4:
                         CommonUtil.send_wx_msg(wx_users, f"\u26A0asin详情kafka实时消费\u26A0",
                                                f"站点： {self.site_name} asin详情实时消费准备失败，请等待处理！")
                     else:
+                        time.sleep(10)
+        else:
             pass
 
     def run_kafka(self):