airflow 的使用之 Operators 介绍

主要内容

1. Operators 简介
2. BaseOperator 简介
3. BashOperator
4. PythonOperator
5. SSHOperator
6. HiveOperator
7. 如何自定义Operator

搭建 airflow 的目的还是为了使用，使用离不开各种 Operators，本文主要介绍以下几点

1. Operators 简介

Operators 允许生成特定类型的任务，这些任务在实例化时成为 DAG 中的任务节点。所有的 Operator 均派生自 BaseOperator，并以这种方式继承许多属性和方法。
Operator 主要有三种类型:

执行一项操作或在远程机器上执行一项操作。
将数据从一个系统移动到另一个系统
类似传感器，是一种特定类型 Operator，它将持续运行，直到满足某种条件。例如在 HDFS 或 S3 中等待特定文件到达，在 Hive 中出现特定的分区或一天中的特定时间，继承自 BaseSensorOperator。

2. BaseOperator 简介

所有的 Operator 都是从 BaseOperator 派生而来，并通过继承获得更多功能。这也是引擎的核心，所以有必要花些时间来理解 BaseOperator 的参数，以了解 Operator 基本特性。

先看一下构造函数的原型：

class airflow.models.BaseOperator(task_id, owner='Airflow', email=None, email_on_retry=True, email_on_failure=True, retries=0, retry_delay=datetime.timedelta(0, 300), retry_exponential_backoff=False, max_retry_delay=None, start_date=None, end_date=None, schedule_interval=None, depends_on_past=False, wait_for_downstream=False, dag=None, params=None, default_args=None, adhoc=False, priority_weight=1, weight_rule=u'downstream', queue='default', pool=None, sla=None, execution_timeout=None, on_failure_callback=None, on_success_callback=None, on_retry_callback=None, trigger_rule=u'all_success', resources=None, run_as_user=None, task_concurrency=None, executor_config=None, inlets=None, outlets=None, *args, **kwargs)

这里有很多参数，一些参数通过字面意思已经了解其大至含义，可查阅官方文档查看详细解释，这里不再详述，需要注意的是参数 start_date。start_date 决定了任务第一次运行的时间，最好的实践是设置 start_date 在 schedule_interval 的附近。比如每天跑的任务开始日期设为’2018-09-21 00:00:00’，每小时跑的任务设置为 ‘2018-09-21 05:00:00’，airflow 将 start_date 加上 schedule_interval 作为执行日期。需要注意的是任务的依赖需要及时排除，例如任务 A 依赖任务 B，但由于两者 start_date 不同导致执行日期不同，那么任务 A 的依赖永远不会被满足。如果你需要执行一个日常任务，比如每天下午 2 点开始执行，你可以在 DAG中使用 cron 表达式

schedule_interval="0 14 * * *"

或者考虑使用 TimeSensor 或 TimeDeltaSensor。由于所有的 Operator 都继承自 BaseOperator,因此 BaseOperator的参数也是其他Operator的参数。

3. BashOperator

官方提供的 DAG 示例－tutorial 就是一个典型的 BashOperator，调用 bash 命令或脚本，传递模板参数就可以参考 tutorial

"""
### Tutorial Documentation
Documentation that goes along with the Airflow tutorial located
[here](http://pythonhosted.org/airflow/tutorial.html)
"""
import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import timedelta# these args will get passed on to each operator
# you can override them on a per-task basis during operator initialization
default_args = {'owner': 'airflow','depends_on_past': False,'start_date': airflow.utils.dates.days_ago(2),'email': ['airflow@example.com'],'email_on_failure': False,'email_on_retry': False,'retries': 1,'retry_delay': timedelta(minutes=5),# 'queue': 'bash_queue',# 'pool': 'backfill',# 'priority_weight': 10,# 'end_date': datetime(2016, 1, 1),# 'wait_for_downstream': False,# 'dag': dag,# 'adhoc':False,# 'sla': timedelta(hours=2),# 'execution_timeout': timedelta(seconds=300),# 'on_failure_callback': some_function,# 'on_success_callback': some_other_function,# 'on_retry_callback': another_function,# 'trigger_rule': u'all_success'
}dag = DAG('tutorial',default_args=default_args,description='A simple tutorial DAG',schedule_interval=timedelta(days=1))# t1, t2 and t3 are examples of tasks created by instantiating operators
t1 = BashOperator(task_id='print_date',   #这里也可以是一个 bash 脚本文件bash_command='date',dag=dag)t1.doc_md = """\
#### Task Documentation
You can document your task using the attributes `doc_md` (markdown),
`doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
rendered in the UI's Task Instance Details page.
![img](http://montcs.bloomu.edu/~bobmon/Semesters/2012-01/491/import%20soul.png)
"""dag.doc_md = __doc__t2 = BashOperator(task_id='sleep',depends_on_past=False,bash_command='sleep 5',dag=dag)templated_command = """
{% for i in range(5) %}echo "{{ ds }}"echo "{{ macros.ds_add(ds, 7)}}"echo "{{ params.my_param }}"
{% endfor %}
"""t3 = BashOperator(task_id='templated',depends_on_past=False,bash_command=templated_command,params={'my_param': 'Parameter I passed in'},dag=dag)t2.set_upstream(t1)
t3.set_upstream(t1)

这里 t1 和 t2 都很容易理解，直接调用的是 bash 命令，其实也可以传入带路径的 bash 脚本, t3 使用了 Jinja 模板，"{% %}" 内部是 for 标签，用于循环操作。"{{ }}" 内部是变量，其中 ds 是执行日期，是 airflow 的宏变量，params.my_param 是自定义变量。根据官方提供的模板，稍加修改即可满足我们的日常工作所需。

4. PythonOperator

PythonOperator 可以调用 Python 函数，由于 Python 基本可以调用任何类型的任务，如果实在找不到合适的 Operator，将任务转为 Python 函数，再使用 PythonOperator 也是一种选择。下面是官方文档给出的 PythonOperator 使用的样例。

from __future__ import print_function
from builtins import range
import airflow
from airflow.operators.python_operator import PythonOperator
from airflow.models import DAGimport time
from pprint import pprintargs = {'owner': 'airflow','start_date': airflow.utils.dates.days_ago(2)
}dag = DAG(dag_id='example_python_operator', default_args=args,schedule_interval=None)def my_sleeping_function(random_base):"""This is a function that will run within the DAG execution"""time.sleep(random_base)def print_context(ds, **kwargs):pprint(kwargs)print(ds)return 'Whatever you return gets printed in the logs'run_this = PythonOperator(task_id='print_the_context',provide_context=True,python_callable=print_context,dag=dag)# Generate 10 sleeping tasks, sleeping from 0 to 9 seconds respectively
for i in range(10):task = PythonOperator(task_id='sleep_for_' + str(i),python_callable=my_sleeping_function,op_kwargs={'random_base': float(i) / 10},dag=dag)task.set_upstream(run_this)

通过以上代码我们可以看到，任务 task 及依赖关系都是可以动态生成的，这在实际使用中会减少代码编写数量，逻辑也非常清晰，非常方便使用。PythonOperator 与 BashOperator 基本类似，不同的是 python_callable 传入的是 Python 函数，而后者传入的是 bash 指令或脚本。通过 op_kwargs 可以传入任意多个参数。

5. SSHOperator

在实际的任务调度中，任务大多分布在多台机器上，如何调用远程机器的任务呢，这时可以简单地使用 SSHOperator 来调用远程机器上的脚本任务。SSHOperator 使用 ssh 协议与远程主机通信，需要注意的是 SSHOperator 调用脚本时并不会读取用户的配置文件，最好在脚本中加入以下代码以便脚本被调用时会自动读取当前用户的配置信息。

. ~/.profile
或
. ~/.bashrc

下面是一个 SSHOperator 的任务示例：

task_crm = SSHOperator(ssh_conn_id='ssh_crmetl', # 指定conn_idtask_id='crmetl-filesystem-check',command='/home/crmetl/bin/monitor/filesystem_monitor.sh', # 远程机器上的脚本文件dag=dag
)

这里 ssh_crmetl 是一个连接ID，是在 airflow webserver 界面配置的，配置方法如下：
打开 webserver 点击 Admin 菜单下的 Connections 项，如下图所示：

然后选择 Create 来新建一个 ssh 连接，输入连接 id，ip 地址，用户名密码等信息，如下图所示：

保存之后，即可使用 ssh_crmetl 来调用对应主机上的脚本了。

6. HiveOperator

hive 是基于 Hadoop 的一个数据仓库工具，可以将结构化的数据文件映射为一张数据库表，并提供简单的 sql 查询功能，可以将 sql 语句转换为 MapReduce 任
务进行运行。在 airflow 中调用 hive 任务，首先需要安装依赖

pip install apache-airflow[hive]

下面是使用示例：

t1 = HiveOperator(task_id='simple_query',hql='select * from cities',dag=dag)

常见的 Operator 还有 DockerOperator，OracleOperator，MysqlOperator，DummyOperator，SimpleHttpOperator 等使用方法类似，不再一一介绍。

7. 如何自定义Operator

如果官方的 Operator 仍不满足需求, 那么我们就自己开发一个 Operator。开发 Operator 比较简单，继承 BaseOperator 并实现 execute 方法即可：

from airflow.models import BaseOperatorclass MyOperator(BaseOperator):def __init__(*args, **kwargs):super(MyOperator, self).__init__(*args, **kwargs)def execute(self, context):###do something here

除了 execute 方法外，还可以实现以下方法：
on_kill: 在 task 被 kill 的时候执行。

airflow 是支持Jinjia模板语言的，那么如何在自定义的 Operator 中加入Jinjia模板语言的支持呢？
其实非常简单，只需要在自定义的Operator类中加入属性

template_fields = (attributes_to_be_rendered_with_jinja)

即可，例如官方的 bash_operator中是这样的：

template_fields = ('bash_command', 'env')

这样，在任务执行之前，airflow 会自动渲染 bash_command 或 env 中的属性再执行任务。

总结：airflow 官方已经提供了足够多，足够实用的 Operator，涉及数据库、分布式文件系统、http连接，远程任务等等，可以参考 airflow 的operators 源码，已基本满足日常工作需要，个性的任务可以通过自定义Operator 来实现，更为复杂的业务可以通过 restful apt 的形式提供接口，然后再使用 SimpleHttpOperator 来实现任务的调用。