如何在 Airflow 中重启 Celery Worker

Question

我目前 运行 遇到一个问题，我的 Celery worker 有时会死机，需要重新启动。我有一个 Airflow 实例，我想从中使用 BashOperator 或 SSHOperator 配置和安排重启，但我不确定如何继续，因为它在 Docker 容器中.

从本地主机我只是 "docker restart worker"，但由于 Bash 在气流容器内运行，我需要找到一种方法通过气流 CLI 或 SSH 从容器到主机重新启动，这听起来很糟糕。有没有一种优雅的方法来解决这个问题？

编辑： 我看到 Flower 有一个 REST API，我可以从中重新启动池，但我必须在我的 celery 配置中启用它。看起来已经 airflow is setting the config 了，有什么方法可以覆盖它吗？

Answer 1

好的，这就是我最后做的：

t1 = BashOperator(
    task_id="celery_restart",
    bash_command="celery multi restart 1",
    dag=dag
)

并没有真正阻止我的任务挂起，但它确实存在 :)。