对气流的BaseSensorOperator参数感到困惑：超时、POKE

编程入门行业动态更新时间:2024-10-27 06:22:27

本文介绍了对气流的BaseSensorOperator参数感到困惑：超时、POKE_INTERVAL和MODE的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！问题描述

我有一些关于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和timeout两个参数之后，我发现任务失败而不是跳过。

那么这里发生了什么？

传感器每4小时戳一次，每次戳都会等待超时持续时间(12小时)？

还是每4小时戳一次，总共戳3次，然后超时？

另外，如果我使用mode="；rechedule"；？

，这些参数会发生什么情况

以下是BaseSensorOperator的文档

class BaseSensorOperator(BaseOperator, SkipMixin): """ Sensor operators are derived from this class and inherit these attributes. Sensor operators keep executing at a time interval and succeed when a criteria is met and fail if and when they time out. :param soft_fail: Set to true to mark the task as SKIPPED on failure :type soft_fail: bool :param poke_interval: Time in seconds that the job should wait in between each tries :type poke_interval: int :param timeout: Time, in seconds before the task times out and fails. :type timeout: int :param mode: How the sensor operates. Options are: ``{ poke | reschedule }``, default is ``poke``. When set to ``poke`` the sensor is taking up a worker slot for its whole execution time and sleeps between pokes. Use this mode if the expected runtime of the sensor is short or if a short poke interval is requried. When set to ``reschedule`` the sensor task frees the worker slot when the criteria is not yet met and it's rescheduled at a later time. Use this mode if the expected time until the criteria is met is. The poke inteval should be more than one minute to prevent too much load on the scheduler. :type mode: str """ 推荐答案

定义术语

poke_interval：b/w连续‘戳’的持续时间(评估被‘感知’的必要条件)

timeout：只是无限期戳是不可接受的(例如，如果您的错误代码在某一天戳到29，无论什么时候是2个月，它都会一直戳4年)。因此，我们定义了停止戳并终止(传感器标记为FAILED或SKIPPED)的最长时间段)

soft_fail：正常情况下(当soft_fail=False时)，超时后传感器标记为FAILED。当soft_fail=True时，传感器将在超时后标记为SKIPPED

mode：这是一个稍微复杂的

任何任务(包括传感器)在运行时都会占用某个池(default池或显式指定的pool)中的slot；实质上是占用一些资源。
对于传感器，这是
- 浪费：因为我们只是在等待(不做实际工作
- 危险：如果您的工作流中有太多几乎同时进入感知的传感器，它们可能会在相当长的时间内冻结大量资源。事实上，有太多的ExternalTaskSensor是notorious，无法将整个工作流(DAG)放入deadlocks
要克服此问题，Airflow v1.10.2 introducedmode传感器中
- mode='poke'(默认)表示我们上面讨论的现有行为
- mode='reschedule'表示在拨打尝试而不是going to sleep之后，传感器将表现为失败(在当前尝试中)，其状态将从RUNNING更改为UP_FOR_RETRY。这样，它将释放插槽，允许其他任务在等待另一次插入尝试时继续进行
在此引用相关代码段from code

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