编程开源技术交流,分享技术与知识

网站首页 > 开源技术 正文

0613-Airflow集成自动生成DAG插件

wxchong 2024-07-03 02:27:43 开源技术 13 ℃ 0 评论

作者:李继武

1.文档编写目的


Airflow的DAG是通过python脚本来定义的,原生的Airflow无法通过UI界面来编辑DAG文件,这里介绍一个插件,通过该插件可在UI界面上通过拖放的方式设计工作流,最后自动生成DAG定义文件。

  • 内容概述

1. Airflow插件集成

2. 使用介绍

3. 总结

  • 安装环境

1. RedHat7.4

2. Python2.7

3. Airflow1.10.1

2.集成DAG生成插件


1. 在github上下载该插件并上传到服务器上并解压,github地址为:

https://github.com/lattebank/airflow-dag-creation-manager-plugin





2. 在AIRFLOW_HOME目录下创建plugins目录,复制插件文件到该目录下,执行以下命令:

mkdir -p /opt/airflow/plugins
cp -r airflow-dag-creation-manager-plugin-master/plugins/* /opt/airflow/plugins/


3. 修改配置文件airflow.cfg,在最后添加如下配置

[dag_creation_manager]
# DEFAULT: basis
dag_creation_manager_line_interpolate = basis 
# Choices for queue and pool
dag_creation_manager_queue_pool = mydefault:mydefault|mydefault 
# MR queue for queue pool
dag_creation_manager_queue_pool_mr_queue = mydefault:mydefault
# Category for display
dag_creation_manager_category = custom
# Task category for display
dag_creation_manager_task_category = custom_task:#ffba40
# Your email address to receive email
# DEFAULT: 
dag_creation_manager_default_email = lijiwu@macro-china.net
dag_creation_manager_need_approver = False
dag_creation_manager_can_approve_self = True
dag_creation_manager_dag_templates_dir = /opt/airflow/plugins/dcmp/dag_templates





4. 关闭Airflow

pkill airflow


5. 因为该插件还集成了安全认证,但使用的flask-login模块与当前的airflow自动下载的模块版本不匹配,先卸载原来的flask-login

pip uninstall flask-login


上传Flask-Login-0.2.11.tar.gz并解压




安装

cd Flask-Login-0.2.11
python setup.py install





该插件启用之后,许多功能会被屏蔽掉,此处不开启,如果需要开启在Airflow.cfg中的[webserver]配置:

authenticate = True
auth_backend = dcmp.auth.backends.password_auth


6. 执行如下命令更新数据库

python /opt/airflow/plugins/dcmp/tools/upgradedb.py


7. 启动airflow

8. 该插件生成的DAG都需要指定一个POOL来执行任务,根据我们在DAG中配置的POOL来创建POOL:




打开UI界面,选择“Admin”下的“Pools”




选择“create”进行创建:



3.使用介绍


1. 创建DAG,选择“Admin”下的“DAG Creation Manager”




2. 点击“Create”




3. 出现如下界面




4. 下拉到底部,填写DAG相关配置,此处配置每分钟执行一次




5. 在DAG图中,选择添加“ADD TASK”,来添加一个节点




6. 在下方填写该TASK的名称及脚本类型与脚本代码等信息,此处脚本内容为向/tmp/airflow.dat文件定时输入“*************************”:




7. 再点击“ADD TASK”,将会在上面的“task1”节点后添加一个task,此处的规则是要在哪个task后添加一个任务,先点击该task,再点击“ADD TASK”:

第二个TASK设为定期向上面的文件/tmp/airflow.dat中输入当前时间:




8. 再添加一个与task1同级的task,向/tmp/airflow.log定期输出当前时间:




9. 修改依赖,将task1和task3都作为task2的依赖:先点击task2,点击Change Upstream,选择task3。




10. 点击保存




11. 回到主界面之后,该DAG不会马上被识别出来,默认情况下Airflow是5分钟扫描一次dag目录,该配置可在airflow.cfg中修改。

识别出来之后打开主界面,点击“暂停按钮”取消暂停开始执行:




启动之后airflow仍会将之前积压的批次执行,终端上查看这两个文件



4.总结


1. 该插件目前只适用于Python2,对于Python3的环境不适合。

Tags:

本文暂时没有评论,来添加一个吧(●'◡'●)

欢迎 发表评论:

最近发表
标签列表