Confus à propos des paramètres BaseSensorOperator d'Airflow: timeout, poke_interval et mode

Question

J'ai un peu de confusion sur le fonctionnement BaseSensorOperatordes paramètres: timeout& poke_interval. Considérez cette utilisation du capteur:

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

La documentation mentionne les actes de délai d'expiration pour définir la tâche sur «échouer» après son épuisement. Mais j'utilise a soft_fail=True, je ne pense pas qu'il conserve le même comportement, car j'ai trouvé que la tâche a échoué au lieu de sauter après avoir utilisé les deux paramètres soft_failet timeout.

Alors que se passe-t-il ici?

1. Le capteur pique toutes les 4 heures, et à chaque poussée, attendra la durée du délai (12 heures)?
2. Ou est-ce qu'il pique toutes les 4 heures, pour un total de 3 coups, puis expire?
3. Aussi, que se passe-t-il avec ces paramètres si j'utilise le mode = "reschedule"?

Voici la documentation de 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
    """

Answer 1

Définition des termes

1. poke_interval: la durée b / w 'pokes' successifs (évaluation de la condition nécessaire qui est 'sentie')

2. timeout: Le simple fait de piquer indéfiniment est inadmissible (si, par exemple, votre code de buggy pique le jour pour devenir 29 chaque fois que le mois est 2, il continuera à piquer pendant jusqu'à 4 ans). On définit donc une période maximale au-delà de laquelle on arrête de piquer et on termine (le capteur est marqué soit FAILEDou SKIPPED)

3. soft_fail: Normalement (quand soft_fail=False), le capteur est marqué comme FAILEDaprès la temporisation. Quand soft_fail=True, le capteur sera plutôt marqué comme SKIPPEDaprès l'expiration du délai

4. mode: Ceci est un peu complexe

   • Toute tâche (y compris le capteur) lorsqu'elle s'exécute, mange un slotdans un pool ( defaultpool ou explicitement spécifié pool); ce qui signifie essentiellement qu'il prend des ressources.
   • Pour les capteurs, c'est
     • gaspillage : car un emplacement est consommé même lorsque nous attendons juste (ne faisant aucun travail réel
     • dangereux : si votre flux de travail comporte trop de capteurs qui entrent en détection à peu près au même moment, ils peuvent geler beaucoup de ressources pendant un certain temps. En fait, trop de personnes ayant des ExternalTaskSensors sont réputées pour mettre des flux de travail entiers (DAG) dans des blocages
   • Pour surmonter ce problème, Airflow v1.10.2 a introduit mode s dans les capteurs
     • mode='poke' (par défaut) signifie le comportement existant dont nous avons discuté ci-dessus
     • mode='reschedule'signifie qu'après une tentative de poussée , plutôt que de s'endormir , le capteur se comportera comme s'il avait échoué (lors de la tentative en cours) et son état passera de RUNNINGà UP_FOR_RETRY. De cette façon, il libèrera son emplacement, permettant à d'autres tâches de progresser en attendant une autre tentative de poussée
   • Citant l'extrait de code pertinent ici

   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
   

   • Pour plus d'informations, lisez la section Paramètres des capteurs

---

Et maintenant répondre directement à vos questions

Q1

1. Le capteur pique toutes les 4 heures, et à chaque poussée, attendra la durée du délai (12 heures)?
2. Ou est-ce qu'il pique toutes les 4 heures, pour un total de 3 coups, puis expire?

le point 2 est correct

Q2

Aussi, que se passe-t-il avec ces paramètres si j'utilise le mode = "reschedule"?

Comme expliqué précédemment, chacun de ces paramètres est indépendant et le réglage mode='reschedule'ne modifie en aucune façon leur comportement

Answer 2

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

Disons que les critères ne sont pas remplis au premier coup. Donc, il fonctionnera à nouveau après 4 heures d'intervalle. Mais le slot worker sera libéré pendant l'attente puisque nous utilisons le mode="reschedule".

C'est ce que j'ai compris.