Airflow 系列3：Airflow 元数据库与执行器配置

前面已经完成了 Airflow 2.3.4 的基础安装，能够启动 webserver 和 scheduler，也能够通过页面访问 Airflow。

不过，能够启动并不代表这个环境已经适合实际使用。

默认安装出来的 Airflow 通常会使用 SQLite 作为元数据库，并且使用 SequentialExecutor 作为执行器。这种组合适合快速体验 Airflow，但是不适合作为长期使用的调度环境。

本文继续基于 Apache Airflow 2.3.4，主要整理两个配置：

1. 将 Airflow 元数据库从 SQLite 修改为 MySQL。
2. 将执行器从 SequentialExecutor 修改为 LocalExecutor。

这两个配置完成之后，Airflow 才更接近一个单机准生产环境。

一、为什么要修改 Airflow 元数据库

Airflow 中有一个非常重要的组件叫做 Metadata Database，也就是元数据库。

它并不存储业务数据，而是存储 Airflow 自己运行所需要的状态数据，例如：

• DAG 信息
• DAG Run 运行记录
• Task Instance 状态
• 任务重试次数
• 任务开始时间和结束时间
• 用户、角色和权限信息
• Connection、Variable、Pool 等配置

也就是说，Airflow 页面上看到的大量状态信息，最终都依赖元数据库保存。

如果元数据库不稳定，Airflow 的调度状态、页面展示、任务重试和历史记录都会受到影响。

官方文档中也说明，默认 SQLite 主要用于开发测试。如果要真正体验或使用 Airflow，需要考虑 PostgreSQL、MySQL 或 MSSQL 这类数据库后端。

二、默认 SQLite 的问题

刚安装完成 Airflow 后，如果没有修改数据库配置，airflow.cfg 中通常会看到类似配置：

[database]
sql_alchemy_conn = sqlite:////home/atguigu/airflow/airflow.db

SQLite 的优点是简单。

它不需要单独启动数据库服务，也不需要创建用户和授权，只需要一个本地文件就可以保存数据。

但是 SQLite 的限制也很明显：

1. 不适合多进程并发写入。
2. 不适合多个 Airflow 组件同时频繁访问。
3. 只能配合 SequentialExecutor 使用。
4. 不适合生产或准生产环境。

Airflow 的调度过程会频繁读写元数据库。Scheduler 需要写入 DAG Run、Task Instance 状态，Webserver 需要读取状态展示页面，执行器也需要更新任务运行结果。

这些操作放在 SQLite 上，作为学习环境问题不大；但是如果后续需要运行多个 DAG、多个任务，或者希望任务能够并行执行，就应该把元数据库切换到 MySQL 或 PostgreSQL。

本文使用 MySQL。

这张图中需要重点理解的是：元数据库是 Airflow 保存调度状态的核心位置。

默认 SQLite 模式下，状态保存在本地 airflow.db 文件中；切换到 MySQL 后，Webserver、Scheduler、Executor 都会通过同一个 MySQL 元数据库协同工作。

三、创建 MySQL 元数据库

先在 MySQL 中创建 Airflow 使用的数据库。

参考 Word 文档中的命令：

CREATE DATABASE airflow_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

如果是在本机测试环境中，可以直接使用 root 用户连接 MySQL；如果是正式环境，建议单独创建 Airflow 用户。

这里使用单独用户：

$ sudo mysql -e "
CREATE DATABASE IF NOT EXISTS airflow_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
CREATE USER IF NOT EXISTS 'airflow'@'localhost' IDENTIFIED BY '<AIRFLOW_DB_PASSWORD>';
ALTER USER 'airflow'@'localhost' IDENTIFIED BY '<AIRFLOW_DB_PASSWORD>';
GRANT ALL PRIVILEGES ON airflow_db.* TO 'airflow'@'localhost';
FLUSH PRIVILEGES;
"

查看数据库字符集：

$ sudo mysql -e "
SELECT SCHEMA_NAME, DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME
FROM information_schema.SCHEMATA
WHERE SCHEMA_NAME='airflow_db';
"

正常结果类似：

SCHEMA_NAME  DEFAULT_CHARACTER_SET_NAME  DEFAULT_COLLATION_NAME
airflow_db   utf8mb4                     utf8mb4_unicode_ci

这里要注意，Airflow 官方文档要求数据库使用 UTF-8 字符集。MySQL 中通常可以使用 utf8mb4。

四、安装 MySQL 连接依赖

Airflow 连接 MySQL 不是直接通过命令行里的 mysql 客户端，而是通过 Python 数据库驱动连接。

Word 文档中使用的是：

$ pip install mysql-connector-python

对应连接串为：

sql_alchemy_conn = mysql+mysqlconnector://root:123456@hadoop102:3306/airflow_db

本文实际安装 Airflow 2.3.4 时，使用的是官方更推荐的 mysqlclient 驱动，对应连接串是 mysql+mysqldb://。

安装方式如下：

$ conda run -n airflow234 python -m pip install 'mysqlclient==2.1.1'

如果直接使用 pip 环境，也可以执行：

$ pip install 'mysqlclient==2.1.1'

安装完成后，可以验证 MySQLdb 模块是否能够正常导入：

$ conda run -n airflow234 python -c 'import MySQLdb; print(MySQLdb.__file__)'

正常情况下会输出 MySQLdb 模块所在路径。

五、ModuleNotFoundError: No module named 'MySQLdb'

这个错误是在初始化 Airflow 元数据库时遇到的：

ModuleNotFoundError: No module named 'MySQLdb'

当 airflow.cfg 中使用下面这种连接串时：

sql_alchemy_conn = mysql+mysqldb://airflow:******@localhost:3306/airflow_db?charset=utf8mb4

Airflow 会通过 SQLAlchemy 加载 mysqldb 驱动，而 mysqldb 驱动实际依赖 Python 包 mysqlclient 提供的 MySQLdb 模块。

所以报错原因不是 MySQL 服务没有启动，而是当前 Python 环境中没有安装 mysqlclient。

处理方式：

$ conda run -n airflow234 python -m pip install 'mysqlclient==2.1.1'

如果安装 mysqlclient 时编译失败，需要先安装系统依赖：

$ sudo apt-get install -y build-essential pkg-config default-libmysqlclient-dev

然后再重新安装：

$ conda run -n airflow234 python -m pip install 'mysqlclient==2.1.1'

六、修改 airflow.cfg 中的数据库连接

Airflow 的配置文件通常位于 $AIRFLOW_HOME/airflow.cfg。

如果不确定当前 Airflow 读取的是哪个目录，可以先设置：

$ export AIRFLOW_HOME=/home/parallels/airflow234

然后修改配置文件：

$ vim $AIRFLOW_HOME/airflow.cfg

找到 [database] 配置段，将 SQLite 连接修改为 MySQL 连接。

修改前：

[database]
sql_alchemy_conn = sqlite:////home/parallels/airflow234/airflow.db

修改后：

[database]
sql_alchemy_conn = mysql+mysqldb://airflow:<AIRFLOW_DB_PASSWORD>@localhost:3306/airflow_db?charset=utf8mb4

如果使用 Word 文档中的 mysql-connector-python，连接串可以写成：

[database]
sql_alchemy_conn = mysql+mysqlconnector://root:123456@hadoop102:3306/airflow_db

两种方式不要混用。

如果使用 mysql+mysqldb://，就安装 mysqlclient。

如果使用 mysql+mysqlconnector://，就安装 mysql-connector-python。

可以通过命令查看 Airflow 当前读取到的数据库连接：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow config get-value database sql_alchemy_conn

如果输出仍然是 SQLite，说明配置文件路径、AIRFLOW_HOME 或环境变量还没有对应上。

七、重新初始化 Airflow 元数据库

修改数据库连接后，需要重新初始化 Airflow 元数据库。

如果之前启动过 Airflow，建议先停止：

$ ps -ef | egrep 'scheduler|airflow-webserver' | grep -v grep
$ ps -ef | egrep 'scheduler|airflow-webserver' | grep -v grep | awk '{print $2}' | xargs kill -15

然后执行初始化：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow db init

正常输出中可以看到 MySQL 相关信息：

DB: mysql+mysqldb://airflow:***@localhost:3306/airflow_db?charset=utf8mb4
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

初始化完成后，可以查看 MySQL 中 Airflow 创建了多少张表：

$ sudo mysql -N -e "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='airflow_db';"

本次测试环境中返回：

33

这说明 Airflow 已经把元数据表创建到了 MySQL 中。

八、MySQL SSL 报错处理

Word 文档中提到过一个 MySQL SSL 相关报错：

Linux error:1425F102:SSL routines:ssl_choose_client_version:unsupported protocol

这个错误通常和 MySQL 客户端、服务端 SSL 协议协商有关。

可以先查看 MySQL 是否启用了 SSL：

SHOW VARIABLES LIKE '%ssl%';

如果输出中 have_ssl 为 YES，说明 MySQL 支持 SSL。

Word 文档中的处理方式是在 MySQL 配置文件 my.cnf 中关闭 SSL：

# disable_ssl
skip_ssl

修改后需要重启 MySQL：

$ sudo systemctl restart mysql

这里需要说明一下：

如果只是本机学习环境，关闭 SSL 可以快速绕过这个问题。

如果是生产环境，不建议简单关闭 SSL，而是应该根据 MySQL 版本、客户端驱动版本和证书配置，选择正确的 SSL 参数。

本文的本机 Ubuntu + MySQL 8.0.32 环境没有遇到这个 SSL 报错，所以没有修改 skip_ssl。

九、Invalid default value for 'update_at' 问题分析

Word 文档中还提到一个初始化报错：

1067 - Invalid default value for 'update_at'

这个问题通常和 MySQL 的 sql_mode 有关。

在某些 MySQL 版本或严格模式配置下，MySQL 不允许 timestamp 字段使用非法默认值，例如空值或 0000-00-00 00:00:00。如果 Airflow 初始化表结构时碰到这类默认值，就可能报 Invalid default value。

Word 文档中的处理方式是修改 MySQL 的 sql_mode：

set GLOBAL sql_mode = 'STRICT_TRANS_TABLES,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION';

但是只执行这条 SQL 有一个问题：MySQL 重启后配置会失效。

所以更推荐写入 my.cnf：

[mysqld]
sql_mode = STRICT_TRANS_TABLES,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION

然后重启 MySQL：

$ sudo systemctl restart mysql

本文使用的是 MySQL 8.0.32，执行 airflow db init 时没有遇到该错误，所以没有主动修改 sql_mode。

这里建议按实际情况处理：

1. 如果初始化没有报错，不需要为了这个问题提前修改 sql_mode。
2. 如果遇到 Invalid default value，先确认当前 MySQL 版本和 sql_mode。
3. 再根据实际报错决定是否调整 NO_ZERO_DATE 等配置。

查看当前 sql_mode：

SELECT @@GLOBAL.sql_mode;
SELECT @@SESSION.sql_mode;

十、为什么要修改执行器

数据库切换完成后，还需要关注执行器。

Airflow 的 Scheduler 负责判断任务什么时候可以运行，而 Executor 负责决定任务以什么方式运行。

默认安装时经常会看到：

[core]
executor = SequentialExecutor

SequentialExecutor 的特点非常简单：

1. 一次只运行一个 Task Instance。
2. 适合 SQLite 和本地快速测试。
3. 不适合作为实际调度环境。

官方文档中也说明，SequentialExecutor 是首次安装 Airflow 时的默认执行器，它是唯一可以和 SQLite 配合使用的执行器，因为 SQLite 不支持多连接并发；该执行器一次只运行一个任务实例。

这就会带来一个明显问题：

如果一个 DAG 中有多个可以并行运行的任务，或者同时有多个 DAG 被触发，SequentialExecutor 也只能一个一个执行。

调度任务少的时候感受不明显；任务多起来之后，就会出现任务排队、调度阻塞、执行延迟等问题。

十一、修改为 LocalExecutor

本文是单机部署，不使用 Celery，也不使用 Kubernetes，所以选择 LocalExecutor。

修改 $AIRFLOW_HOME/airflow.cfg：

[core]
executor = LocalExecutor

LocalExecutor 会在本机通过进程方式执行任务。

它仍然是单机执行器，但是可以在本机并行运行多个任务，比 SequentialExecutor 更适合单机 Airflow 环境。

需要注意的是，修改为 LocalExecutor 后，元数据库不能继续使用 SQLite，必须使用 MySQL、PostgreSQL 等数据库后端。

所以数据库和执行器的修改顺序建议是：

1. 先准备 MySQL。
2. 修改 sql_alchemy_conn。
3. 修改 executor = LocalExecutor。
4. 执行 airflow db init。
5. 启动 webserver 和 scheduler。

这张图中需要重点理解的是：执行器决定任务实例的执行方式，不决定 DAG 怎么写。

同一份 DAG 代码，在 SequentialExecutor 下会按顺序一次执行一个任务；切换为 LocalExecutor 后，满足依赖条件的多个任务可以在单机多个本地进程中并行执行。

十二、常见 Executor 简单说明

Airflow 支持多种 Executor。

常见几种如下：

Executor	说明	适用场景
SequentialExecutor	顺序执行器，一次只运行一个任务	本地学习、SQLite 测试
LocalExecutor	本地执行器，在单机上并行执行任务	单机部署、准生产测试
CeleryExecutor	基于 Celery 分发任务到多个 Worker	多节点部署、任务量较大
KubernetesExecutor	每个任务以 Kubernetes Pod 方式运行	Kubernetes 环境
CeleryKubernetesExecutor	Celery 和 Kubernetes 混合方式	同时需要常驻 Worker 和弹性 Pod 的场景

本文选择 LocalExecutor，不是因为它最强，而是因为它和当前单机部署目标匹配。

如果后续需要多节点分布式执行任务，可以继续升级为 CeleryExecutor 或 KubernetesExecutor。

十三、修改后的关键配置汇总

最终需要重点确认这几处配置。

[core]
executor = LocalExecutor
load_examples = False

[database]
sql_alchemy_conn = mysql+mysqldb://airflow:<AIRFLOW_DB_PASSWORD>@localhost:3306/airflow_db?charset=utf8mb4

确认配置是否生效：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow config get-value core executor
LocalExecutor

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow config get-value database sql_alchemy_conn
mysql+mysqldb://airflow:******@localhost:3306/airflow_db?charset=utf8mb4

确认数据库初始化是否成功：

$ sudo mysql -N -e "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='airflow_db';"
33

确认 Airflow 健康状态：

$ curl -fsS http://127.0.0.1:8080/health

正常情况下，metadatabase 和 scheduler 都应该是 healthy。

十四、总结

本文主要完成了 Airflow 从“能跑”到“更接近实际使用”的两个关键配置。

第一，元数据库从 SQLite 切换为 MySQL。

SQLite 简单，但是只适合开发测试；MySQL 更适合作为单机或准生产环境的元数据库。

第二，执行器从 SequentialExecutor 切换为 LocalExecutor。

SequentialExecutor 一次只能运行一个任务，适合本地体验；LocalExecutor 可以在单机上并行执行任务，更适合后续运行多个 DAG。

本篇中还整理了几个常见问题：

1. ModuleNotFoundError: No module named 'MySQLdb'
2. MySQL SSL 协议报错
3. Invalid default value for 'update_at'
4. SQLite 与 SequentialExecutor 的限制

这几个问题都是 Airflow 从默认安装走向可用部署时比较容易遇到的点。后续继续编写 DAG 时，很多任务状态、运行记录和日志入口，都要依赖本文配置好的元数据库和执行器。

参考资料

• Apache Airflow 2.3.4 官方文档：https://airflow.apache.org/docs/apache-airflow/2.3.4/howto/set-up-database.html
• Apache Airflow 2.3.4 官方文档：https://airflow.apache.org/docs/apache-airflow/2.3.4/executor/sequential.html
• Apache Airflow 2.3.4 官方文档：https://airflow.apache.org/docs/apache-airflow/2.3.4/executor/local.html