🐵安装AirFlow - 图1

1. 安装MySQL

首先确保已经安装好 MySQL，最少是 5.7 版本以上，个人建议 MySQL 8.0。

2. 配置相关数据库

使用 MySQL root 账户创建 airflow 数据库、用户以及相关权限，后续我们将配置 MySQL 作为 Backend 数据库。当然，如果你只是想尝试一下，可以直接跳到第 3 步。

  1. > CREATE DATABASE airflow;
  2. > ALTER DATABASE airflow CHARACTER SET utf8;
  3. > CREATE USER 'airflow'@'localhost' IDENTIFIED BY 'airflow1234';
  4. > GRANT ALL PRIVILEGES ON airflow.* TO 'airflow'@'localhost';
  5. > FLUSH PRIVILEGES;
  6. > SET GLOBAL explicit_defaults_for_timestamp=1;

3. 安装Airflow

关于 Airflow 安装，其实官网上已经非常详细了，直接使用 pip 就可以进行安装。值得一提的是，如果你后续想要将Airflow、Zeppelin 和 Superset 放在一个环境中使用的话，个人推荐还是使用 Python 3.7 环境。

  1. $ export AIRFLOW_HOME=~/bigdata/airflow
  2. $ AIRFLOW_VERSION=2.3.0
  3. $ PYTHON_VERSION=3.9
  4. $ CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
  5. # For example: https://raw.githubusercontent.com/apache/airflow/constraints-2.0.2/constraints-3.6.txt
  6. $ pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

4. 安装第三方插件

Airflow 有很多非常好用的第三方插件，提供了功能非常丰富的 Hooks 和 Operators，目前官方(2.2 版本) 所列出来的第三方插件如下：
image.png

在终端中执行以下代码来安装常用的第三方插件，其中 pykerberos 、pyodbc、mysqlclient 需要先使用 conda 进行安装：

  1. $ conda install pykerberos pyodbc mysqlclient -y
  2. $ pip install 'apache-airflow[async, celery, cgroups, cncf.kubernetes, dask, github_enterprise, google_auth]'
  3. $ pip install 'apache-airflow[kerberos, ldap, leveldb, pandas, password, rabbitmq, sentry, statsd, virtualenv]'
  4. $ pip install 'apache-airflow[docker, jenkins, mongo, neo4j, redis, postgres, microsoft.mssql, odbc, mysql, hdfs]'
  5. $ pip install 'apache-airflow[ftp, grpc, http, imap, jdbc, papermill, sftp, sqlite, ssh]'
  6. $ pip install 'apache-airflow[dingding, jira, tableau]'

5. 命令行补全

Airflow 提供了命令行 Tab 补全，通过以下方式进行安装和激活：

  1. $ pip install argcomplete
  2. $ activate-global-python-argcomplete

添加下面内容到 .zshrc，然后执行 source ~/.zshrc：

  1. export AIRFLOW_HOME=$HOME/bigdata/airflow
  2. export SLUGIFY_USES_TEXT_UNIDECODE=yes
  3. export PYTHONPATH="/home/yumingmin/bigdata/airflow/plugins"
  5. autoload bashcompinit
  6. bashcompinit
  7. eval "$(register-python-argcomplete airflow)"

6. 配置Airflow

6.1 AIR_HOME变量

在 ~/.zshrc、~/.bash_profile 中添加 AIRFLOW_HOME 变量：

  1. export AIRFLOW_HOME=$HOME/bigdata/airflow
  2. export SLUGIFY_USES_TEXT_UNIDECODE=yes

6.2 修改airflow.cfg

先执行 airflow db init 来在制定目下生成相关文件（该方式使用的数据库是 sqlite），我们在相关文件上进行修改就比较简单了。

主要关注以下几个地方需要进行修改：

  1. [core]
  2. default_timezone = Asia/Shanghai
  3. executor = CeleryExecutor
  4. sql_alchemy_conn = mysql+pymysql://airflow:airflow1234@localhost:3306/airflow?charset=utf8&unix_socket=/var/run/mysqld/mysqld.sock
  5. sql_alchemy_pool_size = 0
  6. sql_alchemy_max_overflow = -1
  7. parallelism = 64
  8. dag_concurrency = 32
  9. load_examples = False
  10. load_default_connections = False
  13. [webserver]
  14. base_url = http://localhost:9000
  15. default_ui_timezone = Asia/Shanghai
  16. web_server_port = 9000
  17. reload_on_plugin_change = True
  18. # 下面2行是没有的，需要添加
  19. authenticate = True
  20. auth_backend = airflow.contrib.auth.backends.password_auth
  22. [smtp]
  23. smtp_host = airflow.com
  24. smtp_starttls = True
  25. smtp_ssl = False
  26. smtp_user = airflow
  27. smtp_password = airflow1234
  28. smtp_port = 25
  29. smtp_mail_from = airflow@airflow.com
  31. [celery]
  32. worker_concurrency = 32
  33. broker_url = redis://localhost:6379/0
  34. flower_url_prefix = /flower
  35. result_backend = db+pymysql://airflow:airflow1234@localhost:3306/airflow?charset=utf8&unix_socket=/var/run/mysqld/mysqld.sock

7. 创建用户

  1. $ airflow users create --username admin --firstname Mingmin --lastname Yu --role Admin --email yu_mingm623@163.com

8. 一些常见问题总结

8.1 MySQLdb库为安装

在使用 airflow db init 命令来初始化时，如果你连接的数据库是 MySQL，经常会遇到提示 MySQLdb 库还未安装。

解决办法：可以先尝试安装以下 mysqlclient 库，看看能不能解决：

  1. $ pip install mysqlclient

8.2 _mysql is not defined

解决办法：如果上述方法并不能解决问题，这个时候我们可以使用 pymysql 库来替代，这在 Python 3.x 是非常常见的做法。在 ~/bigdata/miniconda3/lib/python3.6/site-packages/airflow/__init__.py 头部添加以下代码：

  1. import pymysql
  2. # pymysql.version_info = (1, 4, 13, "final", 0)  # 如果提示 MySQLdb 版本太低，可以手动指定版本
  3. pymysql.install_as_MySQLdb()

8.3 DateTime2literal问题

修改完之后，再次运行 airflow db init 还是有报错，但是这次报错如下：

  1. File "~/bigdata/miniconda3/lib/python3.6/site-packages/airflow/settings.py", line 333, in configure_adapters
  2.     MySQLdb.converters.conversions[Pendulum] = MySQLdb.converters.DateTime2literal
  3. AttributeError: module 'pymysql.converters' has no attribute 'DateTime2literal'

解决办法：打开 ~/bigdata/miniconda3/lib/python3.6/site-packages/airflow/settings.py 文件，找到 configure_adapters 函数，注释掉以下内容：

  1. try:
  2.     import MySQLdb.converters
  4.     MySQLdb.converters.conversions[Pendulum] = MySQLdb.converters.DateTime2literal
  5. except ImportError:
  6.     pass

8.4 Specified key was too long; max key length is 3072 bytes

解决办法：修改 MySQL 中 airflow 数据库的编码

  1. > ALTER DATABASE airflow CHARACTER SET utf8;

9. 启动

  1. $ airflow webserver
  2. $ airflow scheduler
  3. # 如果安装 apache-airflow[celery] 的话
  4. $ airflow celery flower
  5. $ airflow celery worker

10. 参考文档