气流调度程序故障

Question

我关注了 this 教程尝试使用我自己的 DAG 在 localhost 上构建气流集群。当我在配置文件中设置executor = CeleryExecutor 后运行airflow scheduler 时，我收到以下回溯：

Traceback（最近一次调用最后一次）：

文件“/home/yurii/Tools/anaconda3/bin/airflow”，第 28 行，在 args.func(args)

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/airflow/bin/cli.py”，第 839 行，在调度程序 job.run() 中

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/airflow/jobs.py”，第 200 行，运行中 self._execute()

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/airflow/jobs.py”，第 1309 行，在 _execute self._execute_helper(processor_manager)

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/airflow/jobs.py”，第 1441 行，在 _execute_helper self.executor.heartbeat()

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/airflow/executors/base_executor.py”，第 124 行，心跳 self.execute_async(key, command=command, queue=queue)

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/airflow/executors/celery_executor.py”，第 80 行，在 execute_async args=[command], queue=queue)

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/celery/app/task.py”，第 573 行，在 apply_async **dict(self._get_exec_options(), **options)

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/celery/app/base.py”，第 354 行，在 send_task reply_to=reply_to 或 self.oid，**选项

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/celery/app/amqp.py”，第 310 行，在 publish_task **kwargs

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/kombu/messaging.py”，第 172 行，在发布中 routing_key，强制，立即，交换，声明）

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/kombu/connection.py”，第 449 行，在 _ensured 返回乐趣(*args, **kwargs)

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/kombu/messaging.py”，第 188 行，在 _publish 强制=强制，立即=立即，

文件“/home/yurii/Tools/anaconda3/lib/python3.6/site-packages/librabbitmq/init.py”，第 122 行，在 basic_publish 强制或假，立即或假，

TypeError：需要一个整数（类型为 NoneType）

一些附加信息：

• 我正在使用 Airflow 1.8.0 以及 Celery 3.1.25 和 RabbitMQ 3.5.7 作为代理和后端，但也尝试了 Airflow 1.9.0 和 Celery 4.2。
• 带有顺序执行器的气流可以正常工作。
• `airflow test "dag_name" "task_name" "exec_date" 运行成功。

我是 Airflow/Celery/RabbitMQ/SQL 的新手，因此我们将不胜感激！

Answer 1

添加到以前的答案。使用 py-amqp 涉及从 broker_url = amqp://XXXXX 更改为 broker_url = pyamqp://XXXXX 或 pip uninstall librabbitmq.

此外，您可能需要将airflow.cfg 中的celery_result_backend 变量更改为result_backend。在最近的版本中，airflow.cfg 中 [celery] 节点中的变量的 celery_ 前缀已被删除。

Answer 2

您似乎使用 librabbitmq 作为 amqp 代理，这不是 celery 核心团队推荐的。使用 py-amqp 作为 rabbitmq 代理，你应该可以摆脱这个错误。