Apache Airflow调度中心发布任务步骤

本文主要介绍如何使用airflow发布和调度任务。

Airflow是一个以编程方式创作，安排和监控工作流程的平台。对比crontab来看，它是一个可以定时调度任务的系统，只不过，airflow的调度更容易管理。

airflow支持任务依赖pipeline，这是crontab以及quartz所不支持的。
airflow调度系统和业务系统解耦。业务单独编写流程，支持任务热加载。
airflow支持crontab定时格式
airflow通过Python来定义task，可以实现复杂的逻辑，支持分支条件等
airflow有一套完整的UI和管理系统
airflow有强大的插件扩展方式，各种插件很丰富，很容易二次开发，添加新功能
airflow是分布式设计，支持水平扩容
airflow支持task实例，并支持数据业务日期bizdate, 也叫 execution_date.
airflow支持任务补录backfill
airflow支持任务之间数据传递(这个任务依赖于上个任务的变量)
airflow支持序列执行(这个周期的任务依赖于上一个周期的执行结果是否成功)
Airflow支持MQ完成scheduler到worker的消息传递，底层由celery+Rabbitmq实现。scheduler将需要运行的任务发送到MQ中，发送到MQ中包含任务对应的队列信息。worker从MQ获取消息时，仅获取相应队列任务，拉取到对应worker上执行。

airflow可以调度远程(SSH)机器上的bash脚本或发送http请求远程机器提供的任务接口，不仅如此airflow还有可视化的任务调度平台，可以方便直观的查看任务执行情况，并可以随意支配任务执行频次，指定任务间依赖关系等。下面就来具体介绍一下如何使用airflow来发布一个任务

1.首先需要创建airflow平台并创建用户

平台访问地址：http://airflow服务IP:airflow端口/admin/

平台搭建方法在另一篇文章中有详细描述：Apache Airflow搭建及使用指南

2.写调度脚本

调度脚本必须使用python语言来写。这里有两个例子：blogHttpDag（执行http接口形式的任务）和blogSshDag（执行bash脚本），可以参考。

还可以参考airflow的使用及注意事项中关于脚本的编写

具体步骤：（1）点击“DAGs”进入dag列表中，如下图

（2）点击其中一个例子的名称进入该dag详情页，点击红框的“Code”就展示了这个dag的脚本，根据这个脚本可以仿写你自己的调用脚本。

3.上传调度脚本

写好的python脚本需要上传到服务器的$AIRFLOW_HOME/dags/下，然后刷新DAGs页面，就能看到刚才上传的调用脚本出现在了列表中。

4.新建connection信息

脚本中涉及到的远程连接方式及权限需要告知管理员，由他增加到连接管理中，这里要注意conn id由任务创建者自定义，但需保证conn id在connections中的唯一性。

下图是连接列表，这里可以管理airflow与各机器的通信信息，用于远程调用。

下图是某一个连接信息的编辑页面，可以看到，这是一个ssh连接，远程机器IP、登录用户、密码、端口号都需要填写。这里密码虽然设置过但不会显示

5.触发任务调度

这时就可以根据需要，如果想要任务立刻进入调度队列，被调度器调度，那么就开启该dag左边的on按钮，如果想要立刻执行而不是等到设置的间隔时间来触发，可以点击该dag右侧的触发按钮。

on按钮和触发按钮分别是下面截图中的绿色和蓝色框所示。

DAGs页面不会自动刷新,因此需要手动刷新来查看任务执行情况，点击Links里的刷新按钮。

6.查看执行情况

如果出现红色的圈，代表任务或dag有错误，可以点击红圈查看错误详情。

Recent Tasks
这里包含9个圆圈，每个圆圈代表task的执行状态和次数
圈1 success：现实成功的task数，基本上就是该tag包含多少个task，这里基本上就显示几。
圈2 running：正在运行的task数
圈3 failed：失败的task数
圈4 unstream_failed:
圈5 skipped:跳过的task数
圈6 up_for_retry:执行失败的task，重新执行的task数
圈7 queued：队列，等待执行的task数
圈8 ：
圈9 scheduled:刚开始调度dag时，这一次执行总共调度了dag下面多少个task数，并且随着task的执行成功，数值逐渐减少。

Last Run
dag最后执行的时间点

DAG Runs
这里显示dag的执行信息，包括3个圆圈，每个圆圈代表dag的执行状态和次数
圈1 success：总共执行成功的dag数，执行次数
圈2 runing：正在执行dag数
圈3 faild：执行失败的dag数

Code View
查看任务执行代码

Logs
查看执行日志，比如失败原因

Refresh
刷新dag任务

Delete Dag

删除该dag任务，删除的是这个dag实例以及它的日志，之后再刷新页面该dag又会出现在list中。

若想真正删除该dag，需要将其从服务器上的dags路径下手动删除。

注意事项：

如果是远程脚本调用方式，需要提前做以下几件准备工作：

（1）如果是堡垒机，并且要使用ssh远程脚本调用方式，则需要堡垒机上的该台服务器开通非22端口用于监听airflow的ssh连接。

具体开通端口方法，可参考文章：https://blog.csdn.net/lihuifen2011/article/details/105064617

（2）执行任务的服务器需要确保有专门的linux用户可以执行任务，特别是堡垒机服务器，必须将任务涉及的脚本及资源文件的属主改成非堡垒机用户，且该用户要有相应的任务执行权限等。

（3）需要开通网络访问权限。airflow到该台服务器的ssh端口需要开通权限,防火墙需要放开airflow机器的ssh访问。

如果是http远程调用方式，需要注意开通对应的端口的网络访问权限，防火墙需要放开airflow机器的http访问。