Планировщик не обновляет файлы пакета

Question

Я разрабатываю DAG на Cloud Composer; мой код разделен на основной файл Python и один пакет с подпапками, это выглядит так:

my_dag1.py
package1/__init__.py
package1/functions.py
package1/package2/__init__.py
package1/package2/more_functions.py

---

Я обновил одну из функций на package1/functions.py, чтобы получить дополнительный аргумент (и обновить ссылку в my_dag1.py). Код будет работать правильно в моей локальной среде, и я не получаю никаких ошибок при запуске

gcloud beta composer environments run my-airflow-environment list_dags --location europe-west1

Но веб-интерфейс вызвал ошибку Python

TypeError: my_function () получила неожиданный аргумент ключевого слова 'New_argument'

---

• Я попытался переименовать функцию, и ошибка изменилась на NameError: name 'my_function' is not defined

• Я попытался изменить имя DAG и загрузить файлы в папку dag в заархивированном и разархивированном виде, но ничего не получилось.

Ошибка исчезла только после того, как я переименовал папку пакета.

---

Я подозреваю, что проблема связана с выбором планировщика my_dag1.py, но не package1/functions.py. Ошибка появилась из ниоткуда, поскольку я делал подобные обновления в предыдущие недели.

Есть идеи, как решить эту проблему без рефакторинга всей структуры кода?

---

EDIT-1

Вот ссылка на соответствующее обсуждение в группах Google

Answer 1

Я столкнулся с подобной проблемой. ошибка «Сломанная группа доступности базы данных» не исчезнет в веб-интерфейсе. Я предполагаю, что это ошибка кеша на веб-сервере AirFlow.

• Фон.

  1. Я создал пользовательский оператор с функциями Airflow .
  2. После того, как я импортирую настроенного оператора, веб-интерфейс воздушного потока показывает, что ошибка Broken DAG говорит о том, что он не может найти настроенного оператора.

• Почему я считаю, что это ошибка в веб-сервере Airflow?

  1. Я могу вручную запустить DAG с помощью команды airflow test, поэтому импорт должен быть корректным.
  2. Даже если я удалю соответствующий файл DAG из папки /dags/ воздушного потока, ошибка все еще будет.

• Вот что я сделал, чтобы решить эту проблему.

  1. перезапустить веб-сервис airflow. (иногда вы можете решить проблему только этим).
  2. убедитесь, что DAG не запущен , перезапустите службу планировщика воздушного потока.
  3. убедитесь, что DAG не запущен , перезапустите рабочий поток воздуха

Надеюсь, это может помочь кому-то иметь такую ​​же проблему.