环境配置

Ubuntu

Python 2.7

安装步骤

设置基础的默认路径：export AIRFLOW_HOME=~/airflow,这是将默认的路径设置成为根路径下的airflow的文件夹下面（也是默认设置），如果你想要将文件保存其他地方，如根目录下的test目录下，那么就设置 export AIRFLOW_HOME=~/test。

开始安装：pip install airflow

初始化配置：pip initdb，这个时候会出现如下圈出的文件

接下来我们需要再这个文件夹下面创建一个dags文件夹：cd ~/airflow  mkdir dags

在dags目录创建一个test.py文件：touch test.py，并且加入如下的代码：

启动web 服务：airflow webserver –p 3004, 启动服务后我们可以看到下面这个界面

然后我们开始启动调度：airflow scheduler

现在基本的airflow已经部署成功

配置文件

我们可以在airflow文件夹下面可以看到airflow.cfg的文件

我们可以打开发现内容如下：

我们可以从文件中可以看到，config文件主要包含着[core], [cli], [api], [lineage], [atlas], [operators], [hive],

[webserver], [email], [smtp], [celery], [celery_broker_transport_options],

[dask], [shceduler], [ldap], [mesos], [github_enterprise], [admin], [elasticsearch],

[kubernetes], [kubernetes_secrets], [kubernetes_node_selectors]这些模块

在这么多模块中，先介绍下几个重要的配置：

[core] 下面的executor配置：airflow执行任务的方式，在配置中一共有5个选项（SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor,

KubernetesExecutor）

表示单进程顺序执行，通常用来测试用的

表示多进程本地执行，它用Python的多进程库从而达到多进程跑任务的效果

表示使用celery作为执行器，只要配置了celery，就可以分布式的部署跑任务，一般用于生产环境

后面两个没有找到相关的资料，打扰了~~

关于celery安装

使用pip安装：pip install airflow[celery]

由于celery需要设置broker和result队列来保存消息（celery支持多种broker），在这里我们需要安装redis，redis安装教程：http://www.runoob.com/redis/redis-install.html

同时，对airflow.cfg文件进行配置：

Executor = CeleryExecutor

Broler_rul = 你的redis地址

Celery_resul_backend = 你的redis地址

关于数据库的设置

当我们在使用airflow initdb的时候，我们会默认的生成一个sqllite文件

如果你想切换成mysql或者其他数据库：

1. 安装你需要切换的数据库所需要的包，这里选择的是mysql：pip install airflow[mysql]

2. 修改配置文件：sql_alchemy_comm = 你的mysql数据库连接

关于登录模块

1. 首先，我们需要安装登录模块所需要的安装包：pip install airflow[password]

2. 修改配置文件airflow.cfg的[webserver]模块：

Authenticate = True

Auth_backend = airflow.contrib.auth.backends.password_auth

最后通过python运行下面的代码

创建多task任务

一般情况下，我们创建task任务的时候需要一个个的去定义，然后通过设置set_downstream或者set_upstream的方式去创建工作流，但是这样工作量却显得太大（万一task很多的时候），如下：

我们可以通过如下代码去设置：

通过locals()生成变量名。

给任务函数参数设置

在使用PythonOperator的时候，如何给python_callable添加参数，我们进入PythonOperator函数定义可以看到：

所以我们只要通过设置参数op_args(list) 或者op_kwargs(dict)就可以设置回调函数的参数：