Airflow 中的延迟通知系统

Question

当 DAG 或 Task Fails 或 Success 或 Delayed 时，我正在尝试实施 Airflow 警报/通知系统。我成功地实现了 FAILURE 和 SUCCESS 通知，并且还成功地使用 SLA 部分实现了延迟通知系统，但 SLA 的局限性在于它只能与计划的 DAG 一起使用，我们系统中的许多 DAG 都是基于触发器的。有没有办法在 Airflow 中实现没有 SLA 的延迟通知系统？提前致谢！

Answer 1

服务水平协议 (SLA) 提供了发送电子邮件的功能，以防任务超出其从 DAG 执行开始后的预期时间范围，并且根据 official documentation 和 medium article，Airflow 当前没有t 在触发的 DAG 中支持此功能；在这种情况下，我建议您在Airflow Issue Tracker 中打开一个改进问题以请求支持此功能。

另一方面，您可以使用以下方法在 DAG 通过on_success_callback 完成运行后验证 SLA：

def success_function (context):
   s_date = context.get('task_instance').start_date.replace(tzinfo = None)
   e_date = datetime.now().replace(tzinfo = None)
  
   execution_time = e_date - s_date
   sla = timedelta (seconds = 30)
 
   if execution_time> sla:
       print('send_email')
   else:
       print('no send email')

t1 = BashOperator (
   task_id = 'print_date',
   bash_command = 'date',
   dag = dag,
   on_success_callback = success_function
)

如果您实施此解决方法，请牢记以下注意事项：

• 上下文不包含任务中配置的SLA信息；因此，有必要在函数中为要验证 SLA 的每个任务指定它
• 错过的 SLA 不会出现在 Airflow UI 中
• 需要创建发送电子邮件的逻辑