https://coderoad.ru/57930048/Apache-Airflow-DAG-%D0%BD%D0%B5-%D0%B2%D1%8B%D0%B7%D1%8B%D0%B2%D0%B0%D0%B5%D1%82-on_success_callback-%D0%B8-on_failure_callback

Сходу более красивого текста не нашел, но и тут видно.

спасибо большое

Добро пожаловать в самое дружелюбное комьюнити.

Знаю, что пинг не маленький, но где их глянуть? В Browse->Log я так понял не те логи, на сервере в папке log (прописана в Airflow.cfg в поле base_log_folder) есть подпапка scheduler,  в ней есть директории с датами запусков дагов, но они пусты

если у вас в докере эйрфло, то лог контейнера смотреть, если нет — хз

Можно папку логов эйрфлоу вынести через volume, если он у вас в контейнере

Это все хорошо, только папка логов в контейнере пуста, если через вольюм выведу, он мне пустую папку выведет

Привет!
А у кого был опыт запуска airflow при старте системы, без кластера ? На сингл-машине
#debian10

Не могу мануалов найти

в systemd сервис создай и испольуй

Коллеги добрый вечер. Перенес некоторые таски в LatestOnly. Теперь туплю сижу, что воткнуть перед LoadStageDone  что бы хвост всегда отрабатывал

Смысл такой, что некоторые таски полностью перезаливают таблицы и им не надо каждый час отрабатывать. Можно в самом конце. Две верхние таски тащат данные за период. А хвост переносит все это в другую базу

Хвост сейчас не отрабатывает потому что таски из latest не отрабатывают

Бранч?

Никто с таким не сталкивался?

[2021-02-26 11:12:13,553] {scheduler_job.py:193} ERROR - Got an exception! Propagating...
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 184, in _run_file_processor
    result: Tuple[int, int] = dag_file_processor.process_file(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 648, in process_file
    dagbag.sync_to_db()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 552, in sync_to_db
    for attempt in run_with_db_retries(logger=self.log):
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 390, in __iter__
    do = self.iter(retry_state=retry_state)
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 356, in iter
    return fut.result()
  File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 432, in result
    return self.__get_result()
  File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 388, in __get_result
    raise self._exception
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 566, in sync_to_db
    DAG.bulk_write_to_db(self.dags.values(), session=session)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dag.py", line 1904, in bulk_write_to_db
    DagCode.bulk_sync_to_db([dag.fileloc for dag in orm_dags])
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.8/contextlib.py", line 120, in __exit__
    next(self.gen)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 32, in create_session
    session.commit()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1046, in commit
    self.transaction.commit()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 504, in commit
    self._prepare_impl()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 483, in _prepare_impl
    self.session.flush()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2540, in flush
    self._flush(objects)
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2682, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 68, in __exit__
    compat.raise_(
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 2642, in _flush
    flush_context.execute()
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 422, in execute
    rec.execute(self)
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 586, in execute
    persistence.save_obj(
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 205, in save_obj
    for (
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 398, in _organize_states_for_save
    and instance_key in uowtransaction.session.identity_map
  File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/orm/identity.py", line 98, in __contains__
    if key in self._dict:

TypeError: unhashable type: 'list'

дагпроцессор в таблицу dagcode похоже не может данные записать при парсинге файла дага

Всем привет. Возник один вопрос: для моего кейса надо запускать пайплайн после notify в постгрессе, в интернете не нашел такого кейса, не знаю подходят ли здесь по логике сенсоры airflow

сенсор, да

При тестировании sqlsensor сразу же кидается ошибка: «no results to fetch»

Сделай свой LatestOnlyOperator без скипания даунстрима 🤷🏻‍♂️