airflow - 部署新 dags 的过程

Question

我需要一些关于如何在部署时重新启动所有 Airflow 服务而不杀死任务中的工作人员的建议。

我已经为我的 DAG 编写了一个部署过程，它在 virtualenv 中安装 Airflow 和任何其他 pip 依赖项。一旦我的发布目录准备就绪，我:

1. 停止airflow-flower、airflow-worker、airflow-scheduler 和airflow-webserver
2. 更新“当前”simlink 以指向我的新版本
3. 启动airflow-flower、airflow-worker、airflow-scheduler 和airflow-webserver

这个部署过程的问题是工作人员会立即被杀死。我想在脚本中添加某种监控以暂停所有 DAG，等待工作线程空闲，然后重新启动服务，但 Airflow CLI 无法了解哪些 dags 已启用，也无法了解工作线程是否空闲。

我知道许多 Airflow 服务可以自动检测 dags 文件夹中的更改，但我希望每个部署都有自己的 virtualenv。如果我不重新启动所有服务，那么新部署将不会在我的requirements.txt 文件中获取新行。

Answer 1

您有权访问 Airflow DB，因此请考虑开发一个部署脚本来为您执行此过程。

• 更新 DAG 表以暂停所有 DAG
• 读取 TASK_INSTANCE 表以等待所有 RUNNING 状态任务完成
• 重新启动 Airflow 服务。
• 更新 DAG 表以取消暂停 DAG。