在airflow上部署dag文件的有效方法

以下是我们如何为团队管理它。

首先,在命名约定方面 DAG文件名 匹配 DAG Id 来自DAG本身的内容(包括DAG版本)。这很有用,因为最终在气流UI中看到的是DAG Id,因此您将确切知道每个DAG后面使用了哪个文件。

DAG示例如下:

from airflow import DAG
from datetime import datetime, timedelta

default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': datetime(2017,12,05,23,59),
  'email': ['me@mail.com'],
  'email_on_failure': True
}

dag = DAG(
  'my_nice_dag-v1.0.9', #update version whenever you change something
  default_args=default_args,
  schedule_interval="0,15,30,45 * * * *",
  dagrun_timeout=timedelta(hours=24),
  max_active_runs=1)
  [...]

的名称 DAG文件 将是: my\u nice\u dag-v1.0.9。py公司

• 我们所有的DAG文件都存储在Git存储库中(除其他外)
• 每次在我们的主分支中执行合并请求时,我们的持续集成管道都会启动一个新的构建,并将DAG文件打包到zip中(我们使用Atlassian Bambol,但还有其他解决方案,如Jenkins、Circle CI、Travis…)
• 在Bambol中,我们配置了一个部署脚本(shell),该脚本解压包并将DAG文件放置在 /dags 文件夹
• 我们通常在DEV中部署dag进行测试,然后部署到UAT,最后部署到PROD。由于上面提到的shell脚本,只需单击bambol UI中的按钮即可完成部署。

利益

1. 由于文件名中包含了DAG版本,因此DAG文件夹中不会覆盖DAG文件的早期版本,因此您可以轻松地返回到DAG文件夹
2. 当您的新DAG文件加载到Airflow中时,您可以通过版本号在UI中识别它。
3. 由于DAG文件名=DAG Id,您甚至可以通过添加一些气流命令行来改进部署脚本,以便在部署新DAG后自动打开它们。
4. 因为每个版本的DAGs都在Git中进行了历史化,所以如果需要,我们可以随时返回到以前的版本。

到目前为止,Airflow还没有自己的版本化工作流功能(请参见 this ). 但是,您可以通过在自己的git存储库中管理DAG并将其状态作为子模块获取到airflow repository中来自己管理它。通过这种方式,您始终具有包含具有特定版本的DAG集的单个气流版本。观看更多 here

文档中编写了一个最佳实践:

删除任务

从不从DAG中删除任务。如果删除,历史 任务信息将从Airflow UI中消失。建议 创建新的DAG以防需要删除任务

我相信这就是为什么版本控制主题还不是那么容易解决的原因,我们必须计划一些解决方法。

https://airflow.apache.org/docs/apache-airflow/2.0.0/best-practices.html#deleting-a-task