Confuso sobre os parâmetros BaseSensorOperator do Airflow: tempo limite, poke_interval e modo

Question

Estou um pouco confuso sobre a maneira como BaseSensorOperatoros parâmetros funcionam: timeout& poke_interval. Considere este uso do sensor:

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

A documentação menciona que o tempo limite atua para definir a tarefa para 'falhar' depois que terminar. Mas estou usando um soft_fail=True, não acho que ele retenha o mesmo comportamento, porque descobri que a tarefa falhou em vez de pular depois de usar os parâmetros soft_faile timeout.

Então, o que acontece aqui?

1. O sensor cutuca a cada 4 horas e, a cada cutucada, vai esperar a duração do timeout (12 horas)?
2. Ou ele cutuca a cada 4 horas, para um total de 3 cutucadas, e depois expira?
3. Além disso, o que acontece com esses parâmetros se eu usar o modo = "reescalonar"?

Aqui está a documentação do 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

Definindo os termos

1. poke_interval: a duração b / w sucessivas 'cutucadas' (avaliação da condição necessária que está sendo 'sentida')

2. timeout: Apenas cutucar indefinidamente é inadmissível (se, por exemplo, seu código bugado for cutucar no dia 29 quando o mês for 2, ele continuará cutucando por até 4 anos). Portanto, definimos um período máximo além do qual paramos de cutucar e encerramos (o sensor é marcado como FAILEDou SKIPPED)

3. soft_fail: Normalmente (quando soft_fail=False), o sensor é marcado como FAILEDapós o tempo limite. Quando soft_fail=True, o sensor será marcado como SKIPPEDapós o tempo limite

4. mode: Este é um pouco complexo

   • Qualquer tarefa (incluindo sensor) quando executado, devora slotem algum pool (seja defaultpool ou explicitamente especificado pool); essencialmente significando que consome alguns recursos.
   • Para sensores, isso é
     • desperdício : como um slot é consumido mesmo quando estamos apenas esperando (sem fazer nenhum trabalho real
     • perigoso : se o seu fluxo de trabalho tiver muitos sensores que entram em detecção ao mesmo tempo, eles podem congelar muitos recursos por um bom tempo. Na verdade, ter muitos ExternalTaskSensors é notório por colocar fluxos de trabalho inteiros (DAGs) em deadlocks
   • Para superar esse problema, o Airflow v1.10.2 introduziu mode s em sensores
     • mode='poke' (padrão) significa o comportamento existente que discutimos acima
     • mode='reschedule'significa que após uma tentativa de cutucar , em vez de adormecer , o sensor se comportará como se tivesse falhado (na tentativa atual) e seu status mudará de RUNNINGpara UP_FOR_RETRY. Dessa forma, ele irá liberar seu slot, permitindo que outras tarefas progridam enquanto espera por outra tentativa de cutucada
   • Citando o snippet relevante do código aqui

   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
   

   • Para mais informações, leia a seção Parâmetros dos Sensores

---

E agora respondendo suas perguntas diretamente

T1

1. O sensor cutuca a cada 4 horas e, a cada cutucada, vai esperar a duração do timeout (12 horas)?
2. Ou ele cutuca a cada 4 horas, para um total de 3 cutucadas, e depois expira?

ponto 2. está correto

Q2

Além disso, o que acontece com esses parâmetros se eu usar o modo = "reescalonar"?

Conforme explicado anteriormente, cada um desses parâmetros são independentes e a configuração mode='reschedule'não altera o seu comportamento de forma alguma

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"
)

Digamos que os critérios não sejam atendidos na primeira tentativa. Portanto, ele será executado novamente após 4 horas de intervalo. Mas o slot de trabalho será liberado durante a espera, pois estamos usando o mode="reschedule".

Isso é o que eu entendi.