Apache Airflow Celery Executor: Импорт локального пользовательского пакета python - PullRequest
Новая
       116

Apache Airflow Celery Executor: Импорт локального пользовательского пакета python

0 голосов
/ 30 апреля 2020

Я использую готовую многоуровневую архитектуру Bitnami Airflow, найденную здесь:

https://azure.microsoft.com/en-us/blog/bitnami-apache-airflow-multi-tier-now-available-in-azure-marketplace/

Вот что они должны сказать о том, как распространяются DAGS между узлами:

Все узлы имеют общий том для синхронизации файлов DAG.

Файлы DAG хранятся в каталоге узла. Этот каталог является внешним томом, смонтированным в одном и том же месте на всех узлах (рабочих, планировщик и веб-сервер). Поскольку это общий том, файлы автоматически синхронизируются между серверами. Добавьте, измените или удалите файлы DAG с этого общего тома, и вся система Airflow будет обновлена.

Итак, я настроил каталог dag следующим образом: 

/opt/bitnami/airflow/dags
|
├── dag.py
└── package
      ├── __init__.py
      └── module.py

Вот как выглядит импорт в моей DAG: 

from package.module import something

Но я получаю эту ошибку: 

Broken DAG: [/opt/bitnami/airflow/dags/dag.py] No module named 'package'

Я читал другие темы и пробовал эти решения, но ничего похоже на работу. Изменяет ли тот факт, что я использую исполнителя сельдерея? Я новичок здесь, поэтому, пожалуйста, если вам нужно больше деталей, дайте мне знать.

1 Ответ

0 голосов
/ 30 апреля 2020

Ну, я нашел решение.

Сначала я обнаружил диспетчер плагинов воздушного потока. Похоже, это было в основном для создания ваших собственных пользовательских операторов, хуков и т. Д. c, но я решил, что смогу использовать его для своего сценария.

Итак, я проверил предоставленный bitnami файл airflow.cfg и увидел параметр: 

plugins_folder=/opt/bitnami/airflow/plugins

Этот каталог не существовал на машине, поэтому я создал его и уронил туда свой пакет (без внесения в него изменений).

Затем я перезапустил узлы веб-сервера и узлов планировщика, чтобы получить это изменение.

Теперь я могу запустить импорт в моем сценарии DAG, как и раньше, без проблем: 

from package.module import something

Следующими шагами для меня будет синхронизация этой папки плагинов с репозиторием, содержащим наши пакеты, так же, как я синхронизировал найденную папку DAGS здесь:

https://docs.bitnami.com/azure-templates/infrastructure/apache-airflow/configuration/sync-dags/

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

...