Планировщик воздушного потока выдает ошибку - у объекта «DisabledBackend» нет атрибута «_get_task_meta_for»

Question

Я пытаюсь установить воздушный поток (распределенный режим) в WSL, я получил настройку веб-сервера Airflow, Планировщика воздушного потока, Airflow Worker, Celery (3.1) и RabbitMQ. Во время работы планировщика воздушного потока он выдает эту ошибку (см. Ниже), даже если задан бэкэнд.

ERROR

Traceback (последний вызов был последним): Файл "/usr/local/lib/python3.6/dist-packages/airflow/executors/celery_executor.py", строка 92, синхронизирован state = task.state Файл "/usr/local/lib/python3.6/dist-packages/celery/result.py", строка 398, в состоянии вернуть self._get_task_meta () ['status'] Файл "/usr/local/lib/python3.6/dist-packages/celery/result.py", строка 341, в _get_task_meta вернуть self._maybe_set_cache (self.backend.get_task_meta (self.id)) Файл "/usr/local/lib/python3.6/dist-packages/celery/backends/base.py", строка 288, в get_task_meta meta = self._get_task_meta_for (task_id) AttributeError: у объекта 'DisabledBackend' нет атрибута '_get_task_meta_for'

https://issues.apache.org/jira/browse/AIRFLOW-1840

Это именно та ошибка, которую я получаю, но не могу найти решение.

Результат Backend-

result_backend = db + postgresql: // postgres: **** @ localhost: 5432 / postgres

broker_url = amqp: // rabbitmq_user_name: rabbitmq_password @ localhost / rabbitmq_virtual_host_name

Помогите пожалуйста, просмотрел почти все документы, но не смог найти решение

Answer 1

Я столкнулся с той же проблемой на версии сельдерея - 3.1.26.post2 (с rabitmq, postgresql и airflow), причина этой проблемы - словарь, используемый в celery base.py file at(lib/python3.5/site-packages/celery/app/base.py) не захватывает бэкэнд сельдерея на ключе CELERY_RESULT_BACKEND, а захватывает на ключе result_backend.

Таким образом, решение здесь - перейти к _get_config function available in base.py file at(lib/python3.5/site-packages/celery/app/base.py), в конце функции перед возвратом словаря s добавить следующий код.

s['CELERY_RESULT_BACKEND'] = s['result_backend'] #code to be added

return s

Это решило проблему.