Jak zapobiec przepływowi powietrza z zasypywania działek?

Question

Załóżmy, że masz DAG-a z przepływem powietrza, który nie ma sensu zasypywania, co oznacza, że ​​po jego uruchomieniu raz, jego kolejne uruchomienie szybko byłoby zupełnie bezcelowe.

Na przykład, jeśli ładujesz dane z jakiegoś źródła, które jest aktualizowane tylko co godzinę w bazie danych, zasypywanie, które następuje szybko po sobie, będzie po prostu importować te same dane raz za razem.

Jest to szczególnie denerwujące, gdy tworzysz nowe, godzinne zadanie, i uruchamia ono N ilość razy dla każdej godziny, którą przegapił, wykonując nadmiarową pracę, zanim zacznie działać w podanym przedziale.

Jedynym rozwiązaniem można myślę, że to coś specjalnie odradzane w FAQ of the docs

Zalecamy przed użyciem wartości dynamiczne jak datą_początkową, zwłaszcza datetime.now() jak to może być dość kłopotliwe.

Czy istnieje sposób, aby wyłączyć zasypywanie w przypadku DAG, czy powinienem wykonać powyższe czynności?

Answer 1

Uaktualnij do wersji przepływu powietrza 1.8 i użyj catchup_by_default = False w airflow.cfg lub zastosuj catchup = False do każdego ze swoich dagów.

https://github.com/apache/incubator-airflow/blob/master/UPDATING.md#catchup_by_default

Answer 2

To wydaje się być nierozwiązanym problemem z przepływem powietrza. Wiem, że bardzo chciałbym mieć dokładnie tę samą funkcję. Oto tyle, ile dostałem; może być użyteczny dla innych.

Są funkcje interfejsu użytkownika (przynajmniej w 1.7.1.3), które mogą pomóc w rozwiązaniu tego problemu. Jeśli przejdziesz do widoku drzewa i klikniesz na określone zadanie (kwadratowe pola), pojawi się przycisk dialogowy z przyciskiem "Oznacz sukces". Kliknięcie przycisku "przeszłość", a następnie kliknięcie przycisku "Oznacz sukces" spowoduje oznaczenie wszystkich wystąpień tego zadania w DAG jako udanych i nie zostaną one uruchomione. Najwyższy poziom DAG (kółka na górze) może być również oznaczony jako udany w podobny sposób, ale wydaje się, że nie ma sposobu na etykietowanie wielu instancji DAG.

Nie zajrzałem jeszcze dostatecznie głęboko, ale możliwe jest użycie podkomendy "trigger_dag" do oznaczania stanów DAG. zobacz tutaj: https://github.com/apache/incubator-airflow/pull/644/commits/4d30d4d79f1a18b071b585500474248e5f46d67d

CLI wyposażone oznaczyć DAG jest w pracach: http://mail-archives.apache.org/mod_mbox/airflow-commits/201606.mbox/%3CJIRA.12973462.1464369259000.37918.1465189859133@Atlassian.JIRA%3E https://github.com/apache/incubator-airflow/pull/1590

UPDATE (28.09.2016) został dodany nowy operator 'LatestOnlyOperator' (https://github.com/apache/incubator-airflow/pull/1752), który uruchomi tylko najnowszą wersję zadań niższego rzędu. Brzmi bardzo użytecznie i mam nadzieję, że niedługo pojawi się w wydaniach.

AKTUALIZACJA 2: Od czasu przepływu powietrza 1.8 został zwolniony LatestOnlyOperator.