【airflow实战系列】 基于 python 的调度和监控工做流的平台

简介

airflow 是一个使用python语言编写的data pipeline调度和监控工做流的平台。Airflow被Airbnb内部用来建立、监控和调整数据管道。任何工做流均可以在这个使用Python来编写的平台上运行。

Airflow是一种容许工做流开发人员轻松建立、维护和周期性地调度运行工做流（即有向无环图或成为DAGs）的工具。在Airbnb中，这些工做流包括了如数据存储、增加分析、Email发送、A/B测试等等这些跨越多部门的用例。

这个平台拥有和 Hive、Presto、MySQL、HDFS、Postgres和S3交互的能力，而且提供了钩子使得系统拥有很好地扩展性。除了一个命令行界面，该工具还提供了一个基于Web的用户界面让您能够可视化管道的依赖关系、监控进度、触发任务等。

传统Workflow一般使用TextFiles(json,xml/etc)来定义DAG,而后Scheduler解析这些DAG文件造成具体的TaskObject执行；Airflow没这么干，它直接用Python写DAGdefinition,一会儿突破了文本文件表达能力的局限，定义DAG变得简单。

Airflow的架构

在一个可扩展的生产环境中，Airflow含有如下组件：

• 一个元数据库（MySQL或Postgres）

• 一组Airflow工做节点

• 一个调节器（Redis或RabbitMQ）

• 一个Airflow Web服务器

全部这些组件能够在一个机器上随意扩展运行。若是使用LocalExcuter来适度的安装则能够得到至关多的额外性能。

优势

• python脚本实现DAG，很是容易扩展

• 工做流依赖可视化

• no XML

• 可测试

• 可做为crontab的替代

• 可实现复杂的依赖规则

• Pools

• CLI和Web UI

功能简介

常见命令

• initdb，初始化元数据DB，元数据包括了DAG自己的信息、运行信息等；

• resetdb，清空元数据DB；

• list_dags，列出全部DAG；

• list_tasks，列出某DAG的全部task；

• test，测试某task的运行情况；

• backfill，测试某DAG在设定的日期区间的运行情况；

• webserver，开启webserver服务；

• scheduler，用于监控与触发DAG。

ETL

ETL，是英文 Extract-Transform-Load 的缩写，用来描述将数据历来源端通过抽取（extract）、转换（transform）、加载（load）至目的端的过程。ETL一词较经常使用在数据仓库，但其对象并不限于数据仓库。

Airflow设计时，只是为了很好的处理ETL任务而已，可是其精良的设计，正好能够用来解决任务的各类依赖问题。

任务依赖

一般，在一个运维系统，数据分析系统，或测试系统等大型系统中，咱们会有各类各样的依赖需求。好比：

• 时间依赖：任务须要等待某一个时间点触发

• 外部系统依赖：任务依赖Mysql中的数据，HDFS中的数据等等，这些不一样的外部系统须要调用接口去访问

• 机器依赖：任务的执行只能在特定的某一台机器的环境中，可能这台机器内存比较大，也可能只有那台机器上有特殊的库文件

• 任务间依赖：任务A须要在任务B完成后启动，两个任务互相间会产生影响

• 资源依赖：任务消耗资源很是多，使用同一个资源的任务须要被限制，好比跑个数据转换任务要10个G，机器一共就30个G，最多只能跑两个，我但愿相似的任务排个队

• 权限依赖：某种任务只能由某个权限的用户启动

也许你们会以为这些是在任务程序中的逻辑须要处理的部分，可是我认为，这些逻辑能够抽象为任务控制逻辑的部分，和实际任务执行逻辑解耦合。

如何理解Crontab

如今让咱们来看下最经常使用的依赖管理系统，Crontab

在各类系统中，总有些定时任务须要处理，每当在这个时候，咱们第一个想到的老是crontab。

确实，crontab能够很好的处理定时执行任务的需求，可是对于crontab来讲，执行任务，只是调用一个程序如此简单，而程序中的各类逻辑都不属于crontab的管辖范围（很好的遵循了KISS）

因此咱们能够抽象的认为：

crontab是一种依赖管理系统，并且只管理时间上的依赖。

Airflow的处理依赖的方式

Airflow的核心概念，是DAG(有向无环图)，DAG由一个或多个TASK组成，而这个DAG正是解决了上文所说的任务间依赖。Task A 执行完成后才能执行 Task B，多个Task之间的依赖关系能够很好的用DAG表示完善

Airflow完整的支持crontab表达式，也支持直接使用python的datatime表述时间，还能够用datatime的delta表述时间差。这样能够解决任务的时间依赖问题。

Airflow在CeleryExecuter下可使用不一样的用户启动Worker，不一样的Worker监听不一样的Queue，这样能够解决用户权限依赖问题。Worker也能够启动在多个不一样的机器上，解决机器依赖的问题。

Airflow能够为任意一个Task指定一个抽象的Pool，每一个Pool能够指定一个Slot数。每当一个Task启动时，就占用一个Slot，当Slot数占满时，其他的任务就处于等待状态。这样就解决了资源依赖问题。

Airflow中有Hook机制(其实我以为不该该叫Hook)，做用时创建一个与外部数据系统之间的链接，好比Mysql，HDFS，本地文件系统(文件系统也被认为是外部系统)等，经过拓展Hook可以接入任意的外部系统的接口进行链接，这样就解决的外部系统依赖问题。

参考

http://wingerted.com/2017/02/20/introduce-to-airflow/

https://www.youtube.com/watch?v=cHATHSB_450

https://www.youtube.com/watch?v=Pr0FrvIIfTU

 

此文已由做者受权腾讯云技术社区发布，转载请注明文章出处