if date == f_quat_date:
  some_code
else:
  pass

типо такого

это да, но там ещё 3 месяца обработать надо, если первый день, а не просто один.

а в чем проблема?

тип январь, февраль и март ждём, в начале апреля запустили даг, который обработает январь, февраль и март

не понимаю в чем проблема

у меня сейчас есть отдельный, который раз в месяц запускается, хочу код переиспользовать)

сейчас он при работе использует execution_date из темплейта

можно создание тасков вынести в отдельный модуль и создавать даги с разным расписанием

Я правильно понимаю что в execution_delta  указывается  ориентировочное время выполнения дага от которого идет зависимость?

да, там время запуска плановое, для бэкфила используется, когда даг догоняет последние пару лет

У меня сейчас есть ExternalTaskSensor и там указан отрицательный execution_delta, хочу понять как это работает

При этом в этой цепочке тасков нет бэкфила

Видимо что-то из будущего ждёт

For yesterday, use [positive!] datetime.timedelta(days=1).

Всем привет) ребят подскажите,что не так с воркером, запускаю его через супервизор , он ранится. Но все равно в ошибках показывает Can't connect to local MySQL server through socket '/var/lib/mysql/mysql.sock' (2)

Очевидно, не может подключиться к базе

Сэнк,сорян за глупый вопрос, только узнаю аирфлоу

всем привет, есть вопрос по xCom...
SSHOperator-ом я получаю имя файла и хочу его передать в livyOperator в args.
Проблема в том, что {{task_instance.xcom_pull(task_ids='first')}} у меня подставляется как - b'text.txt' .
Можно ли как то избавиться от этой 'b' перед именем файла? Или как то подругому это реализовать?

second =  LivyOperator
 args=["--new_file=hdfs:///xxxx/{{task_instance.xcom_pull(task_ids='first')}}"]

Replace?..