灵活可扩展的工作流管理平台Airflow
2016-11-30 19:16
441 查看
1. 引言
Airflow是Airbnb开源的一个用Python写就的工作流管理平台(workflow management platform)。在前一篇文章中,介绍了如何用Crontab管理数据流,但是缺点也是显而易见。针对于Crontab的缺点,灵活可扩展的Airflow具有以下特点:工作流依赖关系的可视化;
日志追踪;
(Python脚本)易于扩展
对比Java系的Oozie,Airflow奉行“Configuration as code”哲学,对于描述工作流、判断触发条件等全部采用Python,使得你编写工作流就像在写脚本一样;能debug工作流(test backfill命令),更好地判别是否有错误;能更快捷地在线上做功能扩展。Airflow充分利用Python的灵巧轻便,相比之下Oozie则显得笨重厚拙太多(其实我没在黑Java~~)。《What makes Airflow great?》介绍了更多关于Airflow的优良特性;其他有关于安装、介绍的文档在这里、还有这里。
下表给出Airflow(基于1.7版本)与Oozie(基于4.0版本)对比情况:
功能 | Airflow | Oozie |
---|---|---|
工作流描述 | Python | xml |
数据触发 | Sensor | datasets, input-events |
工作流节点 | operator | action |
完整工作流 | DAG | workflow |
定期调度 | DAG schedule_interval | coordinator frequency |
任务依赖 | >>, << | <ok to> |
内置函数、变量 | template macros | EL function, EL constants |
2. 实战
Airflow常见命令如下:initdb,初始化元数据DB,元数据包括了DAG本身的信息、运行信息等;
resetdb,清空元数据DB;
list_dags,列出所有DAG;
list_tasks,列出某DAG的所有task;
test,测试某task的运行状况;
backfill,测试某DAG在设定的日期区间的运行状况;
webserver,开启webserver服务;
scheduler,用于监控与触发DAG。
下面将给出如何用Airflow完成data pipeline任务。
首先简要地介绍下背景:定时(每周)检查Hive表的partition的任务是否有生成,若有则触发Hive任务写Elasticsearch;然后等Hive任务完后,执行Python脚本查询Elasticsearch发送报表。但是,Airflow对Python3支持有问题(依赖包为Python2编写);因此不得不自己写
HivePartitionSensor:
# -*- coding: utf-8 -*- # @Time : 2016/11/29 # @Author : rain from airflow.operators import BaseSensorOperator from airflow.utils.decorators import apply_defaults from impala.dbapi import connect import logging class HivePartitionSensor(BaseSensorOperator): """ Waits for a partition to show up in Hive. :param host, port: the host and port of hiveserver2 :param table: The name of the table to wait for, supports the dot notation (my_database.my_table) :type table: string :param partition: The partition clause to wait for. This is passed as is to the metastore Thrift client,and apparently supports SQL like notation as in ``ds='2016-12-01'``. :type partition: string """ template_fields = ('table', 'partition',) ui_color = '#2b2d42' @apply_defaults def __init__( self, conn_host, conn_port, table, partition="ds='{{ ds }}'", poke_interval=60 * 3, *args, **kwargs): super(HivePartitionSensor, self).__init__( poke_interval=poke_interval, *args, **kwargs) if not partition: partition = "ds='{{ ds }}'" self.table = table self.partition = partition self.conn_host = conn_host self.conn_port = conn_port self.conn = connect(host=self.conn_host, port=self.conn_port, auth_mechanism='PLAIN') def poke(self, context): logging.info( 'Poking for table {self.table}, ' 'partition {self.partition}'.format(**locals())) cursor = self.conn.cursor() cursor.execute("show partitions {}".format(self.table)) partitions = cursor.fetchall() partitions = [i[0] for i in partitions] if self.partition in partitions: return True else: return False
Python3连接Hive server2的采用的是impyla模块,
HivePartitionSensor用于判断Hive表的partition是否存在。写自定义的operator,有点像写Hive、Pig的UDF;写好的operator需要放在目录
~/airflow/dags,以便于DAG调用。那么,完整的工作流DAG如下:
# tag cover analysis, based on Airflow v1.7.1.3 from airflow.operators import BashOperator from operatorUD.HivePartitionSensor import HivePartitionSensor from airflow.models import DAG from datetime import datetime, timedelta from impala.dbapi import connect conn = connect(host='192.168.72.18', port=10000, auth_mechanism='PLAIN') def latest_hive_partition(table): cursor = conn.cursor() cursor.execute("show partitions {}".format(table)) partitions = cursor.fetchall() partitions = [i[0] for i in partitions] return partitions[-1].split("=")[1] log_partition_value = """{{ macros.ds_add(ds, -2)}}""" tag_partition_value = latest_hive_partition('tag.dmp') args = { 'owner': 'jyzheng', 'depends_on_past': False, 'start_date': datetime.strptime('2016-12-06', '%Y-%m-%d') } # execute every Tuesday dag = DAG( dag_id='tag_cover', default_args=args, schedule_interval='@weekly', dagrun_timeout=timedelta(minutes=10)) ad_sensor = HivePartitionSensor( task_id='ad_sensor', conn_host='192.168.72.18', conn_port=10000, table='ad.ad_log', partition="day_time={}".format(log_partition_value), dag=dag ) ad_hive_task = BashOperator( task_id='ad_hive_task', bash_command='hive -f /path/to/cron/cover/ad_tag.hql --hivevar LOG_PARTITION={} ' '--hivevar TAG_PARTITION={}'.format(log_partition_value, tag_partition_value), dag=dag ) ad2_hive_task = BashOperator( task_id='ad2_hive_task', bash_command='hive -f /path/to/cron/cover/ad2_tag.hql --hivevar LOG_PARTITION={} ' '--hivevar TAG_PARTITION={}'.format(log_partition_value, tag_partition_value), dag=dag ) report_task = BashOperator( task_id='report_task', bash_command='sleep 5m; python3 /path/to/cron/report/tag_cover.py {}'.format(log_partition_value), dag=dag ) ad_sensor >> ad_hive_task >> report_task ad_sensor >> ad2_hive_task >> report_task
相关文章推荐
- 企业级管理软件快速开发平台-极致业务基础平台-可视化工作流设计
- 适易查询分析、项目管理、工作流演示平台
- 【airflow实战系列】 基于 python 的调度和监控工作流的平台
- 基于工作流的网络运维流程管理平台的建设和应用
- 简单通用的缺陷跟踪管理SAAS云平台,含工作流待办任务,转交、挂起、回退等操作,邮件提醒,能导入导出excel
- 智和网管平台-智能化实现灵活的网络管理
- WEB应用 信息管理系统 数据分析展示系统 OA办公工作流 快速构建与开发平台
- 03-jbpm工作流管理方法扩展
- 【且听我说“镶嵌数据集建库”】1、灵活的元数据管理和扩展
- 【airflow实战系列】 基于 python 的调度和监控工作流的平台
- 利用Office 365 定制企业合同管理平台之设定工作流提醒!
- 企业项目管理(EPM)平台的扩展
- 工作流审批平台-辅助功能-角色管理
- 最近根据上海东兰公司和银狐的资料,设计了一个工作流管理平台
- Airflow:Python 工作流管理利器
- 智和网管平台-智能化实现灵活的网络管理
- Airflow 1.8 工作流平台搭建
- 宏天J2EE协同办公OA+进销存+客户管理+财务管理+工作流平台+快速开发平台 对企事单位开放源码
- 使用Airbnb的Airflow来管理数据工作流
- SmartSoft特色功能之二__安全、灵活、易于扩展的文档管理功能