对Airflow的BaseSensorOperator参数感到困惑：超时，poke_interval和模式

问题描述

我对BaseSensorOperator的参数工作方式有些困惑：timeout和poke_interval。 考虑传感器的这种用法：

BaseSensorOperator(
  soft_fail=True,poke_interval = 4*60*60,# Poke every 4 hours
  timeout = 12*60*60,# Timeout after 12 hours
)

文档中提到了超时操作，用于在任务用完后将其设置为“失败”。但是我使用的是soft_fail=True，我不认为它会保留相同的行为，因为我发现在使用参数soft_fail和{{1 }}。

那这里会发生什么？

1. 传感器每4小时戳一次，每次戳一次，都将等待超时时间（12小时）吗？
2. 还是每4个小时戳一次，总共3次戳，然后超时？
3. 此外，如果我使用mode =“ reschedule”，这些参数会怎样？

这是BaseSensorOperator的文档

timeout

解决方法

定义术语

1. poke_interval：持续时间b / w连续的“戳”（评估“被感知”的必要条件）

2. timeout：不允许无限期地进行戳戳操作（例如，如果您的错误代码在一天中戳为2时，例如当月为2时，它会不断戳戳，直到4为止）年份）。因此，我们定义了一个最大时间段，超过该时间段我们将停止戳并终止（传感器标记为FAILED或SKIPPED）

3. soft_fail：通常（在soft_fail=False时），传感器在超时后被标记为FAILED。当soft_fail=True时，传感器将在超时后被标记为SKIPPED

4. mode：这有点复杂

   • 任何任务（包括传感器）在运行时都会吞噬某个池（slot池或显式指定的default）中的pool；本质上意味着它占用了一些资源。
   • 对于传感器，这是
     • 浪费：因为即使我们只是在等待（不做任何实际工作
     ，
     • 危险：如果您的工作流中有太多传感器同时进入 sensing ，它们可以冻结很多资源。实际上，拥有ExternalTaskSensor的太多notorious，无法将整个工作流（DAG）放入deadlocks
   • 要解决此问题，请在传感器中Airflow v1.10.2 introduced mode s
     • mode='poke'（默认）表示我们上面讨论的现有行为
     • mode='reschedule'表示在进行一次拨动尝试之后，而不是going to sleep时，传感器的行为就像是失败（当前尝试），并且其状态将从{{ 1}}至RUNNING。这样，它将释放它的插槽，在等待另一个>戳尝试
     时允许其他任务进行
   • 在此处引用相关代码段from code

   UP_FOR_RETRY

   • 有关更多信息，请阅读Sensors Params section

---

现在直接回答您的问题

第一季度

1. 传感器每4小时戳一次，每次戳一次，都将等待超时时间（12小时）吗？
2. 还是每4小时戳一次，总共3次戳，然后超时？

点2是正确的

第二季度

此外，如果我使用 mode =“ reschedule”？

如前所述，这些参数中的每个参数都是独立的，并且设置if self.reschedule: reschedule_date = timezone.utcnow() + timedelta( seconds=self._get_next_poke_interval(started_at,try_number)) raise AirflowRescheduleException(reschedule_date) else: sleep(self._get_next_poke_interval(started_at,try_number)) try_number += 1 不会以任何方式改变其行为

,

BaseSensorOperator(
  soft_fail=True,poke_interval = 4*60*60,# Poke every 4 hours
  timeout = 12*60*60,# Timeout of 12 hours
  mode = "reschedule"
)

比方说，第一次戳不满足标准。因此，它将在间隔4小时后再次运行。但是由于我们正在使用mode="reschedule"，因此在等待期间将释放工作槽。

这是我的理解。