airflow 笔记 - 像我这种两三⑩岁的人

link之家

链接快照平台

输入网页链接，自动生成快照
标签化管理网页链接

相关文章推荐

踏实的水煮鱼 · js ...· 4 月前 ·

年轻有为的回锅肉 · Python多任务—协程(asyncio详解 ...· 11 月前 ·

不拘小节的冲锋衣 · chrome 浏览器的使用技巧 - 知乎· 1 年前 ·

率性的山楂 · 拆分函数Split函数VBS VBA ...· 1 年前 ·

淡定的斑马 · 4. 最优化建模· 1 年前 ·

首先是一个比较好的英文网站，可能要fq：http://site.clairvoyantsoft.com/installing-and-configuring-apache-airflow/

========================================================

py3venv

在 mkdir ~/airflow/py3venv

执行：python3 -m venv ~/airflow/py3venv

进入python3 环境：source /home/airflow/py3venv/bin/activate

以后的所有操作都进到 py3venv 的python3 环境操作
=========================================================

看看是否有gcc,没有的话需要进行安装：

sudo yum install gcc （后续安装airflow如果不成功，可以再次执行，它会更新包）

=========================================================

(py3venv) [airflow@iZ114t000jwZ airflow]$ airflow initdb
[2018-08-22 14:30:30,472] {configuration.py:206} WARNING - section/key [celery/celery_ssl_active] not found in config
[2018-08-22 14:30:30,473] {default_celery.py:41} WARNING - Celery Executor will run without SSL
[2018-08-22 14:30:30,474] {__init__.py:45} INFO - Using executor CeleryExecutor
DB: mysql://prodroot:***@rds8r7t55g74gz9a61d0.mysql.rds.aliyuncs.com/airflow?charset=utf8
[2018-08-22 14:30:30,724] {db.py:312} INFO - Creating tables
INFO [alembic.runtime.migration] Context impl MySQLImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
[2018-08-22 14:30:31,107] {models.py:189} INFO - Filling up the DagBag from /home/airflow/proj/airflow/test/dags

=========================================================

启动服务脚本

nohup airflow webserver >>/home/airflow/airflow/logs/webserver.log &
nohup airflow worker >> /home/airflow/airflow/logs/worker.log &
nohup airflow scheduler >> /home/airflow/airflow/logs/scheduler.log &
nohup airflow flower >> /home/airflow/airflow/logs/flower.log &

/home/airflow/proj/airflow/test/airflow-webserver.pid

里面的数字就是webserver的process id， kill -9 {id} 就能停掉 webserver 服务了，重启用上面第一个

重启服务器，airflow会以 /home/airflow/airflow/airflow.cfg 为准

=========================================================

修改数据库配置

可以在~/airflow 目录下，修改airflow.cfg配置文件，数据库连接改为本地mysql的URL

先在mysql数据库里新建airflow数据库

然后修改airflow.cfg配置文件：

sql_alchemy_conn = sqlite:////root/airflow/airflow.db 改为：

sql_alchemy_conn = mysql://root:root@192.168.202.128:3306/airflow

接着重新初始化数据库：

airflow initdb

如果成功，则在airflow数据库表中可以看到很多表

如出现以下报错，则执行： yum install MySQL-python （注意此命令为centos7系统适合）

return __import__('MySQLdb') ImportError: No modulenamedMySQLdb

=======================================================

airflow1.10 中， airflow initdb 报错：

在Flask中连接MySQL时出现ModuleNotFoundError: No module named 'MySQLdb'错误，

只要在配置SQLALCHEMY_DATABASE_URI时，加上一个pymysql就可以了：
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:dzd123@localhost/你的数据库名'
，当然，前提是要已经安装了pymysql，如果没有装pymysql，是会报ModuleNotFoundError: No module named 'pymysql'错误的，

安装命令也很简单：pip install pymysql

=========================================================

问题1：解决MySQLdb ImportError: libmysqlclient.so.18错误

主要解决方法就是创建一个 18的软连接

https://blog.csdn.net/haoyuxuan/article/details/45334087

=========================================================

问题2：File "/home/airflow/py3env/lib/python3.6/site-packages/airflow/models.py", line 639, in set_extra

先安装包：pip install apache-airflow[crypto]

执行：python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

将得到的字符串替换 airflow.cfg 中fernet_key 的值

这段字符串加密算法用到

=========================================================

# 在 python 中执行添加账户：

airflow 1.9 默认安装的 SQLAlchemy version 1.2.2 ，下面这段python会报错：

File "/usr/local/lib/python3.6/site-packages/sqlalchemy/ext/hybrid.py", line 873, in set
raise AttributeError("can't set attribute")
AttributeError: can't set attribute

解决方法，执行：pip install 'sqlalchemy<1.2'

这样才能得到加密的用户记录：

import airflow
from airflow import models, settings
from airflow.contrib.auth.backends.password_auth import PasswordUser
user = PasswordUser(models.User())
user.username = 'airflow'
user.email = 'nick111@163.com'
user.password = 'airflow*****'
session = settings.Session()
session.add(user)
session.commit()
session.close()

=======================================================

redis 安装参考：https://blog.csdn.net/jiangdong2007/article/details/80890236

airflow 1.9.0要使用celery4.x， redis版本4才能与 celery4 兼容

------------

安装 Airflow 1.10.1 时的一个错误：

AttributeError: 'float' object has no attribute 'items'

参考：https://github.com/celery/celery/issues/5175

pip uninstall redis

pip install redis==2.10.6

=========================================================

rabbitmq 安装

主要参考：https://www.linuxidc.com/Linux/2016-03/129557.htm

碰到的坑：

原因就是 erlang 和 rabbitmq各有一个 cookie

~/.erlang.cookie

/var/lib/rabbitmq/.erlang.cookie

将其中一个修改，改成相同的。

ps -ef | grep rabbitmq

将进程杀掉

然后重启服务器就解决了。

=========================================================

Airflow中使用 sshoperator

http://yangcongchufang.com/airflow/airflow-ssh-operator.html

问题：{ssh_operator.py:118} WARNING - /home/airflow/tmp/a.sh: line 3: hive: command not found

a.sh 内容：

echo "aaaa" > /home/airflow/tmp/aaa1.txt
hive -f /home/airflow/tmp/a.hql
echo "aaaa" > /home/airflow/tmp/aaa2.txt

明明调用成功了 a.sh， aaa1.txt aaa2.txt 有成功生成，但是找不到 hive 命令，而在emr服务器上直接执行 sh a.sh 却可以成功打印出数据库名称列表。

解决方法：

执行： export PATH=$PATH:$HIVE_HOME/bin

在 a.sh第一行中加入：

source /etc/profile

=========================================================

如出现以下报错，则执行：yum install MySQL-python （注意此命令为centos7系统适合）

return __import__('MySQLdb') ImportError: No modulenamedMySQLdb

==========================================================

之前用的一直是airflow1.9 ，最近安装了一遍 airflow1.10。

在airflow1.9中，subdag_operator 默认使用的executor = GetDefaultExecutor()，

而1.10中默认的是executor=SequentialExecutor()。

所以导致的结果就是在 1.9 中的subdag中可以并行执行task，而 1.10中，如果不

指定 excutor，只会逐个执行。指定方法：

from airflow.executors.celery_executor import CeleryExecutor
mysubdag = SubDagOperator(
    executor=CeleryExecutor()
关于这个变动，官方文档的 tips 中有做说明： https://airflow.apache.org/concepts.html#subdags
其实是Airflow1.9中其实是有个bug，subdag中无法使用 pool 中的 slot 数控制并行任务数。
1.10中没有做这个bug的修复，而是直接给个默认值：SequentialExecutor。这是个坑~~~-_-||