我正在尝试在气流 DAG 中实施 SLA。
我知道 SLA 是如何工作的，您设置了一个 timedelta 对象，如果任务在该持续时间内没有完成，它将发送一封电子邮件并通知任务尚未完成。
我想要一些类似的功能，但我不想给出持续时间，而是想在 SLA 中设置特定的时间。例如，如果任务由于上午 8:00 未完成，它会发送电子邮件并通知经理。像这样的：
'sla': time(hour=8, minute=0, second=0)
我搜索了很多，但没有找到。
对于这个特定问题有什么解决方案吗？还是 SLA 以外的任何其他解决方案？
提前致谢。
【问题讨论】：
标签： python python-3.x airflow airflow-scheduler
BaseOperator 的SLA 参数需要一个datetime.timedelta 对象，所以那里没有什么可做的了。考虑到SLA 表示计划周期结束后的时间增量。 docs 中的示例假设每天安排一个 DAG：
例如，如果您将 SLA 设置为 1 小时，如果 2016-01-01 实例尚未成功，调度程序将在 2016-01-02 凌晨 1:00 之后不久发送电子邮件。
关键是，它始终是计划周期的时间增量，这不是您想要的。
所以我认为你应该采取另一种方法，比如在你需要的时候安排你的 DAG，执行你想要的任务，然后添加一个 sensor operator 来检查你正在寻找的条件是否得到满足或不。 sensors 有几种类型，具体取决于您可以从中选择的上下文。
另一种选择是，创建一个新的 DAG 专用于检查您在原始 DAG 中执行的任务是否成功执行，并采取相应措施（例如，发送电子邮件等）。为此，您可以使用ExternalTaskSensor，在线查看有关如何实现它的教程，尽管如docs 中所述避免交叉 DAG 依赖关系可能更简单。
希望这可以为您指明正确的方向。
【讨论】：