Airflow1.8 分布式工作流平台搭建 - 好文

版本信息：
python 2.7.15 airflow==1.8.0 SQLAlchemy==1.1.18 flower==0.9.2
其中SQLAlchemy的版本如果是1.2.x版本会在创建用户是报错，
[root@localhost ~]# python2 create_user.py [2018-07-18 18:51:09,336]
{__init__.py:57} INFO - Using executor CeleryExecutor Traceback (most recent
call last): File "create_user.py", line 7, in <module> user.password = 'afuser'
File "/soft/anaconda2/lib/python2.7/site-packages/sqlalchemy/ext/hybrid.py",
line 873, in __set__ raise AttributeError("can't set attribute")
AttributeError: can't set attribute
使用pip安装airflow，celery，此外

* 需要使用rabbitmq作为celery的backend需要pip install airflow[rabbitmq]
* 需要登录认证时pip install airflow[password]
* 需要使用CeleryExecutor时install apache-airflow[celery]
说明文档参考Extra Packages
<http://airflow.incubator.apache.org/installation.html#extra-packages>

airflow默认使用sqlite作为Airflow后台数据库，这里我改成mysql：

首先pip install airflow[mysql]

这里先设置airflow的环境变量AIRFLOW_HOME，
vim /etc/profile 加入export AIRFLOW_HOME=/home/crawl/airflow 然后执行 source
/etc/profile 使生效
否则airflow会在当前用户的家目录下生成airflow文件夹

然后执行airflow，就会在$AIRFLOW_HOME下生成airflow文件夹
. ├── airflow    ├── airflow.cfg    ├── logs    └── unittests.cfg
然后我们需要创建dags目录放任务文件，像这样
. ├── airflow    ├── airflow.cfg    ├── airflow-webserver.pid    ├── dags
│   ├── hello.py    │   ├── my_email.py    │   ├── send_email.py    ├── logs
└── unittests.cfg
airflow-webserver.pid是在运行airflow webserver创建的

然后就开始修改airflow.cfg配置文件
[core] # executor = SequentialExecutor #
使用celeryExecutor可以动态的增加worker个数并将任务在远程机器上执行.生产中建议使用celeryExecutor来执行 executor =
CeleryExecutor # sql_alchemy_conn = sqlite:////root/airflow/airflow.db #
使用mysql作为后台数据库 sql_alchemy_conn = mysql://root:@192.168.2.89/airflow
[webserver] # web登录验证，需要加上auth_backend 否则报错 authenticate = True auth_backend =
airflow.contrib.auth.backends.password_auth [celery] # broker_url =
sqla+mysql://root:@192.168.2.89:3306/airflow # broker_url =
amqp://airflow:[email protected]:5672/for_airflow #
可以使用mysql、RabbitMQ和redis作为broker broker_url =
redis://:[email protected]:6379/0 # celery_result_backend =
db+mysql://airflow:airflow@localhost:3306/airflow2 # celery_result_backend =
amqp://airflow:[email protected]:5672/for_airflow #
同broker_url可以使用mysql、RabbitMQ和redis作为broker celery_result_backend =
redis://:[email protected]:6379/0
如果是用RabbitMQ需要安装RabbitMQ，同理使用redis需要安装redis服务器

我这里使用的是redis，安装参考：Redis（一）安装
<https://blog.csdn.net/allenjay11/article/details/80518136>

这里需要修改配置文件redis.conf
将bind属性改为bind 0.0.0.0，确保其他主机能访问。将requirepass修改为requirepass for_airflow，设置密码
确保redis开启，mysql有airflow数据库，然后执行airflow initdb初始化，正常后会在airflow库下创建如下表

然后运行如下代码创建web用户：
import airflow from airflow import models, settings from
airflow.contrib.auth.backends.password_auth import PasswordUser user =
PasswordUser(models.User()) user.username = 'user_name' user.email =
'[email protected]' user.password = 'passwd' session = settings.Session()
session.add(user) session.commit() session.close()
然后开启web服务：airflow webserver

开启调度airflow scheduler

在浏览器上访问该服务器8080端口，

登录后：

上面的dags是默认自带的，我们可以在$AIRFLOW_HOME下的dags添加自己的任务，

配置远程worker，这也是写这个博客的目的，在配置远程worker的时候踩了一些坑，网上也没找到详细的教程，一般都是这样写：

airflow的安装
<https://www.cnblogs.com/skyrim/p/7456163.html#5.5%20%E9%83%A8%E7%BD%B2%E5%A4%9A%E4%B8%AAworker>
5.5 部署多个worker 在需要运行作业的机器上的安装airflow airflow[celery] celery[redis]
模块后,启动airflow worker即可.这样作业就能运行在多个节点上.
A Guide On How To Build An Airflow Server/Cluster
<https://stlong0521.github.io/20161023%20-%20Airflow.html>
Configure your airflow workers; follow most steps for the airflow server,
except that they do not have PostgreSQL and RabbitMQ servers
主要就是

* 主要的包版本需要一致
* redis或RabbitMQ所在服务器端口开放
*
worker和server的airflow文件夹的文件需要一样，当然AIRFLOW_HOME的值可以不一样，但是airflow.cfg文件里的一些文件路径也要修改，要不然会找不到文件。
确保以上几点后，运行 airflow worker，

上图显示transport为配置文件里的broker_url，result为celery_result_backend，而Running on
http://0.0.0.0:8793/则是日志服务端口，可以查看日志

将以上链接补全后访问：

安装在server上flower，运行airflow flower可以监控监控worker状态：

flower默认没有开启登录，可以通过配置文件设置basic_auth 创建账户，更多配置参见配置
<http://flower-docs-cn.readthedocs.io/zh/latest/config.html#address>：
[root@localhost airflow]# vim flowerconfig.py # Broker settings BROKER_URL =
'redis://:[email protected]:6379/0' # RabbitMQ management api broker_api
= 'http://walwal:*****@0.0.0.0:15672/api/' # Enable debug logging logging =
'DEBUG' # 设置登录的用户名和账号 basic_auth = ['walwal:walwalqq',] address = '0.0.0.0' #
port = 8090
使用celery flower --conf=全路径/flowerconfig.py启动，完成。

热门工具换一换