设置 Postgres celery result_backend 时气流调度程序崩溃答案

【问题标题】：Airflow Scheduler Crashes when setting Postgres celery result_backend设置 Postgres celery result_backend 时气流调度程序崩溃
【发布时间】：2020-03-18 13:26:12
【问题描述】：

我尝试使用 CeleryExecutor 实现 Apache Airflow。对于我使用 Postgres 的数据库，对于我使用 Redis 的 celery 消息队列。使用 LocalExecutor 时一切正常，但是当我在 airflow.cfg 中设置 CeleryExecutor 并希望将 Postgres 数据库设置为 result_backend 时

result_backend = postgresql+psycopg2://airflow_user:*******@localhost/airflow

无论我触发哪个 DAG，运行 Airflow 调度程序时都会出现此错误：

[2020-03-18 14:14:13,341] {scheduler_job.py:1382} ERROR - Exception when executing execute_helper
Traceback (most recent call last):
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
    return obj.__dict__[self.__name__]
KeyError: 'backend'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1380, in _execute
    self._execute_helper()
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1441, in _execute_helper
    if not self._validate_and_run_task_instances(simple_dag_bag=simple_dag_bag):
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/airflow/jobs/scheduler_job.py", line 1503, in _validate_and_run_task_instances
    self.executor.heartbeat()
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/airflow/executors/base_executor.py", line 130, in heartbeat
    self.trigger_tasks(open_slots)
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/airflow/executors/celery_executor.py", line 205, in trigger_tasks
    cached_celery_backend = tasks[0].backend
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/celery/local.py", line 146, in __getattr__
    return getattr(self._get_current_object(), name)
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/celery/app/task.py", line 1037, in backend
    return self.app.backend
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/kombu/utils/objects.py", line 44, in __get__
    value = obj.__dict__[self.__name__] = self.__get(obj)
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/celery/app/base.py", line 1227, in backend
    return self._get_backend()
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/celery/app/base.py", line 944, in _get_backend
    self.loader)
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/celery/app/backends.py", line 74, in by_url
    return by_name(backend, loader), url
  File "<PATH_TO_VIRTUALENV>/lib/python3.6/site-packages/celery/app/backends.py", line 60, in by_name
    backend, 'is a Python module, not a backend class.'))
celery.exceptions.ImproperlyConfigured: Unknown result backend: 'postgresql'.  Did you spell that correctly? ('is a Python module, not a backend class.')

指向数据库的完全相同的参数可以工作

sql_alchemy_conn = postgresql+psycopg2://airflow_user:*******@localhost/airflow

将 Redis 设置为 celery result_backend 是可行的，但我读到这不是推荐的方式。

result_backend = redis://localhost:6379/0

有人看到我做错了吗？

【问题讨论】：

标签： python postgresql redis airflow airflow-scheduler

【解决方案1】：

您需要在数据库连接字符串中添加db+前缀：

f"db+postgresql+psycopg2://{user}:{password}@{host}/{database}"

文档中也提到了这一点：https://docs.celeryproject.org/en/stable/userguide/configuration.html#database-url-examples

【讨论】：

【解决方案2】：

您需要在数据库连接字符串中添加db+前缀：

result_backend = db+postgresql://airflow_user:*******@localhost/airflow

【讨论】：