airflow的学习使用

airflow

Operators:

基本可以理解为一个抽象化的task, Operator加上必要的运行时上下文就是一个task. 有三类Operator:

Sensor(传感监控器), 监控一个事件的发生.
Trigger(或者叫做Remote Excution), 执行某个远端动作, (我在代码中没有找到这个类别)
Data transfer(数据转换器), 完成数据转换

Tasks: task代表DAG中的一个节点, 它其实是一个BaseOperator子类.

Task instances, 即task的运行态实例, 它包含了task的status(成功/失败/重试中/已启动)

Job: Airflow中Job很少提及, 但在数据库中有个job表, 需要说明的是Job和task并不是一回事, Job可以简单理解为Airflow的批次, 更准确的说法是同一批被调用task或dag的统一代号. 有三类Job, 分别SchedulerJob/BackfillJob/LocalTaskJob, 对于SchedulerJob和BackfillJob, job指的是指定dag这次被调用的运行时代号, LocalTaskJob是指定task的运行时代号.

Hooks：

Hook是airflow与外部平台/数据库交互的方式, 一个Hook类就像是一个JDBC driver一样. airflow已经实现了jdbc/ftp/http/webhdfs很多hook. 要访问RDBMS数据库有两类Hook可供选择, 基于原生Python DBAPI的Hook和基于JDBC的Hook

Connections:

我们的Task需要通过Hook访问其他资源, Hook仅仅是一种访问方式, 就像是JDBC driver一样, 要连接DB, 我们还需要DB的IP/Port/User/Pwd等信息. 这些信息不太适合hard code在每个task中, 可以把它们定义成Connection, airflow将这些connection信息存放在后台的connection表中. 我们可以在WebUI的Admin->Connections管理这些连接.

Variables:

Variable 没有task_id/dag_id属性, 往往用来定义一些系统级的常量或变量, 我们可以在WebUI或代码中新建/更新/删除Variable. 也可以在WebUI上维护变量.
Variable 的另一个重要的用途是, 我们为Prod/Dev环境做不同的设置,

XComs:

XCom和Variable类似, 用于Task之间共享一些信息. XCom 包含task_id/dag_id属性, 适合于Task之间传递数据, XCom使用方法比Variables复杂些. 比如有一个dag, 两个task组成(T1->T2), 可以在T1中使用xcom_push()来推送一个kv, 在T2中使用xcom_pull()来获取这个kv.

分支的支持:

airflow有两个基于PythonOperator的Operator来支持dag分支功能.

ShortCircuitOperator, 用来实现流程的判断. Task需要基于ShortCircuitOperator, 如果本Task返回为False的话, 其下游Task将被skip; 如果为True的话, 其下游Task将会被正常执行. 尤其适合用在其下游都是单线节点的场景.

BranchPythonOperator, 用来实现Case分支. Task需要基于BranchPythonOperator, airflow会根据本task的返回值(返回值是某个下游task的id),来确定哪个下游Task将被执行, 其他下游Task将被skip.

Trigger Rules:

可以为dag中的每个task都指定它的触发条件, 这里的触发条件有两个维度, 以T1&T2->T3 这样的dag为例:

一个维度是: 要根据dag上次运行T3的状态确定本次T3是否被调用, 由
DAG的default_args.depends_on_past参数控制, 为True时, 只有上次T3运行成功, 这次T3才会被触发
另一个维度是: 要根据前置T1和T2的状态确定本次T3是否被调用, 由T3.trigger_rule参数控制, 有下面6种情形, 缺省是all_success.

dag提交-python配置任务

DAG 基本参数配置

default_args = {'owner': 'airflow','depends_on_past': False,   # 是否依赖上一个自己的执行状态 'start_date': datetime.datetime(2019, 1, 1),'email': ['wangzhenjun@gmail.com'], # 需要在airflow.cfg中配置下发件邮箱'email_on_failure': False,'email_on_retry': False,'retries': 1,'retry_delay': datetime.timedelta(minutes=5),# 'end_date': datetime(2020, 1, 1),   # 结束时间，注释掉也就会一直执行下去
}

DAG对象

设置dag的执行周期:schedule_interval.该参数可以接收cron 表达式和datetime.timedelta对象，另外airflow还预置了一些调度周期。

preset	Description	cron
None	Don’t schedule, use for exclusively “externally triggered” DAGs
@once	Schedule once and only once
@hourly	Run once an hour at the beginning of the hour	0 * * * *
@daily	Run once a day at midnight	0 0 * * *
@weekly	Run once a week at midnight on Sunday morning	0 0 * * 0
@monthly	Run once a month at midnight of the first day of the month	0 0 1 * *
@yearly	Run once a year at midnight of January 1	0 0 1 1 *

t1 = BashOperator(   #任务类型是bash
task_id='echoDate', #任务id
bash_command='echo date > /home/datefile', #任务命令
dag=dag)

完整样例

# coding: utf-8from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta# 定义默认参数
default_args = {'owner': 'wangzhenjun',  # 拥有者名称'depends_on_past': False,   # 是否依赖上一个自己的执行状态'start_date': datetime(2019, 1, 15, 10, 00),  # 第一次开始执行的时间，为格林威治时间，为了方便测试，一般设置为当前时间减去执行周期'email': ['wangzhenjun01@corp.netease.com'],  # 接收通知的email列表'email_on_failure': True,  # 是否在任务执行失败时接收邮件'email_on_retry': True,  # 是否在任务重试时接收邮件'retries': 3,  # 失败重试次数'retry_delay': timedelta(seconds=5)  # 失败重试间隔
}# 定义DAG
dag = DAG(dag_id='hello_world',  # dag_iddefault_args=default_args,  # 指定默认参数# schedule_interval="00, *, *, *, *"  # 执行周期，依次是分，时，天，月，年，此处表示每个整点执行schedule_interval=timedelta(minutes=1)  # 执行周期，表示每分钟执行一次
)"""
1.通过PythonOperator定义执行python函数的任务
"""
# 定义要执行的Python函数1
def hello_world_1():current_time = str(datetime.today())with open('/root/tmp/hello_world_1.txt', 'a') as f:f.write('%s\n' % current_time)assert 1 == 1  # 可以在函数中使用assert断言来判断执行是否正常，也可以直接抛出异常
# 定义要执行的Python函数2
def hello_world_2():current_time = str(datetime.today())with open('/root/tmp/hello_world_2.txt', 'a') as f:f.write('%s\n' % current_time)# 定义要执行的task 1
t1 = PythonOperator(task_id='hello_world_1',  # task_idpython_callable=hello_world_1,  # 指定要执行的函数dag=dag,  # 指定归属的dagretries=2,  # 重写失败重试次数，如果不写，则默认使用dag类中指定的default_args中的设置
)
# 定义要执行的task 2
t2 = PythonOperator(task_id='hello_world_2',  # task_idpython_callable=hello_world_2,  # 指定要执行的函数dag=dag,  # 指定归属的dag
)t2.set_upstream(t1)  # t2依赖于t1;等价于 t1.set_downstream(t2);同时等价于 dag.set_dependency('hello_world_1', 'hello_world_2')
# 表示t2这个任务只有在t1这个任务执行成功时才执行，
# 或者
t1 >> t2"""
2.通过BashOperator定义执行bash命令的任务
"""
hello_operator = BashOperator(   #通过BashOperator定义执行bash命令的任务task_id='sleep_task',depends_on_past=False,bash_command='echo `date` >> /home/py/test.txt',dag=dag
)
"""
其他任务处理器:
3.EmailOperator : 发送邮件
4.HTTPOperator : 发送 HTTP 请求
5.SqlOperator : 执行 SQL 命令
"""