Airflow 系列7：运维与问题排查

前面几篇文章已经把 Airflow 的基本概念、单机部署、元数据库、DAG 编写方式都整理了一遍。

到这里，一个单机 Airflow 环境基本已经能跑起来：

1. Webserver 可以访问。
2. Scheduler 可以正常调度。
3. 元数据库已经切换为 MySQL。
4. Executor 已经切换为 LocalExecutor。
5. DAG 文件可以被识别。
6. Task 可以手动触发并查看日志。

但是 Airflow 真正使用起来，最常见的问题并不是“怎么安装”，而是运行过程中怎么排查问题。

例如：

1. DAG 文件放进去了，但是页面上看不到。
2. Scheduler 页面提示没有心跳。
3. Task 一直处于 queued。
4. Task 一直处于 running。
5. 任务失败了，但是不知道失败原因。
6. 删除 DAG 之后，页面上过一会又出现。
7. 修改了 DAG 代码，但是页面上没有变化。
8. 日志页面打不开，或者日志里没有看到预期内容。

这一篇就专门整理 Airflow 2.3.4 的常见运维和问题排查方式。

本文环境仍然沿用前面的单机部署环境：

配置项	当前值
Airflow 版本	2.3.4
AIRFLOW_HOME	/home/parallels/airflow234
Conda 环境	airflow234
元数据库	MySQL airflow_db
Executor	LocalExecutor
Webserver 地址	http://10.211.55.4:8080

一、先明确 Airflow 排查顺序

Airflow 出问题时，不建议一上来就重启。

更好的排查顺序是：

1. 先看进程是否还在。
2. 再看 Webserver 和 Scheduler 是否健康。
3. 再看 DAG 是否被解析。
4. 再看 DAG Run 是否被创建。
5. 再看 Task Instance 当前状态。
6. 最后看任务日志和元数据库记录。

可以把排查链路理解成：

Airflow 进程
  -> Scheduler 心跳
  -> DAG 文件解析
  -> DAG Run 创建
  -> Task Instance 调度
  -> Executor 执行
  -> 任务日志
  -> 元数据库状态

这里面比较关键的是：

排查对象	重点
Webserver	页面是否可访问
Scheduler	是否有心跳，是否在解析 DAG
Metadata Database	DAG Run、Task Instance 状态是否正常
Executor	任务是否被提交执行
Log	任务具体失败原因

这张图中需要重点理解的是：Airflow 排查不是只看页面。

Web UI 是入口，元数据库是状态真相，任务日志是失败原因。把页面、进程、健康接口、元数据库和日志结合起来看，排查效率会高很多。

二、常用环境变量和命令前缀

前面安装时已经说明过，操作 Airflow 前先设置环境变量。

$ export AIRFLOW_HOME=/home/parallels/airflow234

如果使用 Conda 环境：

$ source /home/parallels/miniforge3/etc/profile.d/conda.sh
$ conda activate airflow234

也可以不进入环境，直接使用：

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

后面命令为了保持清楚，仍然使用完整写法：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow <subcommand>

这样可以避免当前 shell 环境不对，导致命令操作了另一个 AIRFLOW_HOME。

三、检查 Airflow 进程

Airflow 单机环境中，至少要有两个核心进程：

1. airflow webserver
2. airflow scheduler

可以使用：

$ ps -ef | grep airflow | grep -E 'webserver|scheduler' | grep -v grep

正常情况下，应该能看到类似进程：

airflow webserver -p 8080 -D
airflow scheduler -D
gunicorn: master [airflow-webserver]
gunicorn: worker [airflow-webserver]

如果只看到 Webserver，没有 Scheduler，那么页面可以打开，但是 DAG 解析和任务调度都会有问题。

如果两个都没有，说明 Airflow 服务已经停止。

四、启动和停止 Airflow

Word 文档中也整理过启动和停止脚本。

在当前环境中，可以这样启动：

$ export AIRFLOW_HOME=/home/parallels/airflow234
$ source /home/parallels/miniforge3/etc/profile.d/conda.sh
$ conda activate airflow234
$ airflow webserver -p 8080 -D
$ airflow scheduler -D

也可以写成一行：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow webserver -p 8080 -D
$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow scheduler -D

停止时，不建议直接使用很宽泛的命令：

$ pkill -f airflow

因为这可能误杀其他命令。

更稳妥的方式是先查进程，再按 PID 停止：

$ ps -ef | grep airflow | grep -E 'webserver|scheduler' | grep -v grep

然后：

$ kill -15 <pid>

如果进程长时间不退出，再考虑：

$ kill -9 <pid>

一般先用 kill -15，让进程正常退出。

五、检查 Webserver 是否可访问

浏览器访问：

http://10.211.55.4:8080

如果页面打不开，先在服务器本机测试端口：

$ curl -I http://127.0.0.1:8080

如果本机也不通，检查 Webserver 进程：

$ ps -ef | grep airflow | grep webserver | grep -v grep

如果本机通，但是宿主机访问不通，检查：

1. 虚拟机 IP 是否变化。
2. Webserver 是否监听 0.0.0.0。
3. 防火墙是否放开 8080。
4. 虚拟机网络模式是否可以从宿主机访问。

可以查看端口监听：

$ ss -lntp | grep 8080

如果只监听 127.0.0.1:8080，外部机器就访问不了。

六、检查健康状态

Airflow 提供健康检查接口。

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

正常情况下会看到类似结果：

{
  "metadatabase": {
    "status": "healthy"
  },
  "scheduler": {
    "status": "healthy",
    "latest_scheduler_heartbeat": "2023-06-21T14:20:37.814069+00:00"
  }
}

这里要重点看两个字段：

字段	说明
metadatabase.status	元数据库是否可用
scheduler.status	Scheduler 是否有正常心跳

如果 metadatabase 不健康，通常先检查 MySQL。

如果 scheduler 不健康，通常先检查 Scheduler 进程和 job 表心跳。

注意，健康接口由 Webserver 提供。

所以如果 Webserver 自己都没有启动，健康接口也无法访问。

七、检查 Scheduler 心跳

Scheduler 是 Airflow 调度的核心。

页面上如果出现类似提示：

The scheduler does not appear to be running.
The DAGs list may not update, and new tasks will not be scheduled.

优先检查 Scheduler 进程：

$ ps -ef | grep airflow | grep scheduler | grep -v grep

然后检查健康接口：

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

也可以直接查元数据库中的 job 表。

SELECT id,
       state,
       job_type,
       start_date,
       latest_heartbeat,
       executor_class,
       hostname
FROM job
WHERE job_type = 'SchedulerJob'
ORDER BY latest_heartbeat DESC;

如果 latest_heartbeat 长时间没有更新，说明 Scheduler 没有正常工作。

当前环境可以用 MySQL 命令直接查询：

$ mysql -uairflow -pairflow_pass_234 -D airflow_db -e "
SELECT id,state,job_type,start_date,latest_heartbeat,executor_class,hostname
FROM job
WHERE job_type='SchedulerJob'
ORDER BY latest_heartbeat DESC;"

如果 Scheduler 进程存在，但是心跳不更新，可以继续查看 Scheduler 日志。

日志一般在：

$AIRFLOW_HOME/logs/scheduler/

当前环境也就是：

/home/parallels/airflow234/logs/scheduler/

八、DAG 页面看不到怎么办

这是最常见的问题。

排查顺序如下。

8.1 确认 DAG 文件目录

先确认 Airflow 当前扫描哪个目录：

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

正常应该类似：

/home/parallels/airflow234/dags

然后确认文件是否真的在这个目录：

$ ls -lh /home/parallels/airflow234/dags

8.2 确认 Python 文件没有语法错误

直接执行 DAG 文件：

$ python /home/parallels/airflow234/dags/spark_layer_demo.py

如果这里报错，Scheduler 解析时也会失败。

例如：

ModuleNotFoundError
SyntaxError
ImportError

都可能导致 DAG 不显示。

8.3 查看导入错误

使用 Airflow 命令：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-import-errors

也可以查询 import_error 表：

SELECT timestamp,
       filename,
       stacktrace
FROM import_error
ORDER BY timestamp DESC;

如果有记录，先解决导入错误。

8.4 查看 DAG 是否写入元数据库

SELECT dag_id,
       is_paused,
       is_active,
       last_parsed_time,
       has_import_errors,
       fileloc
FROM dag
WHERE dag_id = 'spark_layer_demo';

如果 dag 表中没有记录，说明 Scheduler 还没有成功解析这个 DAG。

如果 has_import_errors = 1，继续查 import_error。

如果 is_paused = 1，说明 DAG 被暂停了。

8.5 删除 DAG 不等于删除文件

Word 文档中有一句很关键：

删除 DAG 任务不会删除底层文件，过一会还会自动加载回来。

这个现象在 Airflow 中很正常。

页面上删除 DAG，主要删除的是元数据库中的部分记录。

如果 DAG 文件还在 dags_folder 目录中，Scheduler 下一轮扫描时还会重新加载。

所以真正不想让某个 DAG 出现，需要删除或移走底层 Python 文件。

九、修改了 DAG 代码但页面没变化

这种情况一般从三个地方排查。

第一，确认修改的是 Airflow 正在扫描的文件。

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

第二，查看 Code View 中展示的代码是否已经变了。

如果页面 Code View 仍然是旧代码，说明 Scheduler 还没有解析到新文件，或者修改的不是同一个文件。

第三，查看 dag 表中的解析时间。

SELECT dag_id,
       last_parsed_time,
       last_pickled,
       fileloc
FROM dag
WHERE dag_id = 'spark_layer_demo';

如果 last_parsed_time 没有变化，可以等待下一轮解析，或者检查 Scheduler 是否有心跳。

也可以查看 Scheduler 日志，确认是否有解析错误。

十、DAG 不自动运行怎么办

DAG 能看到，不代表一定会自动创建 DAG Run。

需要检查几个点。

10.1 DAG 是否暂停

页面上的开关如果是关闭状态，DAG 不会自动调度。

也可以查表：

SELECT dag_id,
       is_paused,
       next_dagrun,
       next_dagrun_create_after
FROM dag
WHERE dag_id = 'spark_layer_demo';

如果 is_paused = 1，说明 DAG 被暂停。

命令行可以开启：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags unpause spark_layer_demo

10.2 schedule_interval 是否为空

如果 DAG 中写的是：

schedule_interval=None

这个 DAG 不会自动调度，只能手动触发。

10.3 start_date 和当前时间是否合理

如果 start_date 设置在未来，Airflow 不会马上创建 DAG Run。

例如当前时间是 2023-07-12，但是 DAG 写：

start_date=datetime(2023, 8, 1)

那么当前不会自动调度。

10.4 catchup 是否导致历史任务过多

如果：

start_date=datetime(2023, 1, 1)
schedule_interval="@daily"
catchup=True

Airflow 可能会尝试补大量历史周期。

学习环境通常建议：

catchup=False

10.5 查看 DAG Run

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-runs -d spark_layer_demo

或者查表：

SELECT dag_id,
       run_id,
       run_type,
       state,
       execution_date,
       start_date,
       end_date
FROM dag_run
WHERE dag_id = 'spark_layer_demo'
ORDER BY execution_date DESC;

如果没有 dag_run，说明 DAG 还没有被调度或手动触发。

十一、任务一直 queued 怎么办

queued 表示任务已经进入队列，但是还没有真正执行。

常见原因有：

1. Scheduler 有问题。
2. Executor 没有正常执行任务。
3. 并发参数限制。
4. Pool 没有可用 Slot。
5. 上一次异常退出导致状态残留。

先查当前 queued 任务：

SELECT dag_id,
       task_id,
       run_id,
       state,
       queued_dttm,
       queue,
       pool,
       pool_slots,
       priority_weight,
       try_number
FROM task_instance
WHERE state = 'queued'
ORDER BY queued_dttm;

再检查 Scheduler：

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

以及 job 表：

SELECT id,
       state,
       job_type,
       latest_heartbeat,
       executor_class
FROM job
WHERE job_type = 'SchedulerJob'
ORDER BY latest_heartbeat DESC;

如果 Scheduler 正常，再检查 Pool：

SELECT pool,
       slots,
       description
FROM slot_pool;

如果任务使用了某个 Pool，而这个 Pool 的 Slot 不够，任务可能一直排队。

还可以查看当前任务的并发配置：

SELECT dag_id,
       max_active_tasks,
       max_active_runs,
       has_task_concurrency_limits
FROM dag
WHERE dag_id = 'spark_layer_demo';

需要注意：不要一看到 queued 就直接手动改数据库状态。

先确认调度器、执行器和并发限制。

十二、任务一直 running 怎么办

running 表示任务已经开始执行，但是还没有结束。

先查正在运行的任务：

SELECT dag_id,
       task_id,
       run_id,
       state,
       start_date,
       end_date,
       duration,
       hostname,
       unixname,
       job_id,
       try_number
FROM task_instance
WHERE state = 'running'
ORDER BY start_date;

如果任务确实还在运行，应该能在对应机器上看到进程。

可以根据任务内容检查：

$ ps -ef | grep spark_submit
$ ps -ef | grep python
$ ps -ef | grep bash

如果任务状态是 running，但是机器上已经没有对应进程，可能是异常退出后状态没有及时更新。

这时优先看任务日志和 Scheduler 日志。

如果确认是测试环境中的脏状态，可以考虑清理任务状态，而不是直接改表。

Airflow 命令中可以使用：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks clear spark_layer_demo \
  --start-date 2023-07-12 \
  --end-date 2023-07-12 \
  --task-regex dwd

清理后可以重新触发。

生产环境不要随便清理状态，必须先确认任务是否真的没有在运行。

十三、任务失败怎么排查

任务失败时，不要只看页面上的红色状态。

建议按下面顺序查。

13.1 先看任务实例

SELECT dag_id,
       task_id,
       run_id,
       state,
       try_number,
       max_tries,
       start_date,
       end_date,
       duration,
       hostname,
       unixname,
       operator
FROM task_instance
WHERE state = 'failed'
ORDER BY end_date DESC;

这里重点看：

字段	说明
task_id	哪个任务失败
run_id	哪一次 DAG Run 失败
try_number	当前尝试次数
max_tries	最大尝试次数
operator	使用的 Operator 类型
hostname	任务运行在哪台机器

13.2 查看失败记录

SELECT dag_id,
       task_id,
       run_id,
       start_date,
       end_date,
       duration
FROM task_fail
ORDER BY end_date DESC;

task_fail 可以用来观察任务失败历史。

但是具体失败原因还是要看任务日志。

13.3 查看任务日志

Web 页面中可以在 Grid View 点击具体 Task Instance，然后进入日志页面。

命令行也可以查看：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks logs spark_layer_demo dwd <run_id>

如果不确定 run_id，先查：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-runs -d spark_layer_demo

本地日志目录通常在：

$AIRFLOW_HOME/logs/

当前环境中是：

/home/parallels/airflow234/logs/

Airflow 的任务日志通常会按照 DAG ID、Run ID、Task ID、尝试次数组织目录。

13.4 常见失败原因

现象	常见原因
ModuleNotFoundError	当前 Airflow 环境没有安装依赖
Permission denied	脚本或目录权限不足
No such file or directory	命令路径、脚本路径写错
command not found	环境变量不完整，命令不在 PATH 中
Spark 提交失败	Spark 环境、YARN、HDFS 或依赖包问题
MySQL 连接失败	账号、密码、Host、权限或驱动问题
任务超时	execution_timeout 或外部任务卡住

对于 BashOperator，一定要关注 Bash 退出码。

如果命令最终退出码不是 0，任务会失败。

十四、任务重试怎么排查

任务失败后，如果配置了重试，可能会进入：

up_for_retry

查询：

SELECT dag_id,
       task_id,
       run_id,
       state,
       try_number,
       max_tries,
       start_date,
       end_date,
       next_method,
       next_kwargs
FROM task_instance
WHERE state IN ('failed', 'up_for_retry')
ORDER BY start_date DESC;

如果任务处于 up_for_retry，说明它还没有真正结束，而是在等待下一次重试时间。

这时应该看：

1. retries 配置了几次。
2. retry_delay 配置了多长。
3. 当前 try_number 是多少。
4. 失败原因是否已经解决。

例如：

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

表示最多重试 3 次，每次间隔 5 分钟。

如果根因没有解决，重试只会重复失败。

十五、手动触发和补跑

手动触发 DAG：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags trigger spark_layer_demo

触发后查看运行记录：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-runs -d spark_layer_demo

测试单个任务：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks test spark_layer_demo dwd 2023-07-12

tasks test 适合本地调试任务逻辑。

它和正式调度不完全一样，不应该用它替代正常 DAG Run。

如果是生产任务补跑，要先确认：

1. 补跑哪个业务日期。
2. 上游数据是否还在。
3. 下游是否可以覆盖。
4. 是否会重复写入数据。
5. 是否会触发重复通知。

十六、清理和重跑任务

有时候任务失败后，需要清理状态重新跑。

可以使用 airflow tasks clear。

例如清理 spark_layer_demo 中 dwd 在某一天的状态：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks clear spark_layer_demo \
  --start-date 2023-07-12 \
  --end-date 2023-07-12 \
  --task-regex dwd

如果要清理整个 DAG 的某个日期，可以不加 --task-regex。

但是要谨慎。

清理状态会让 Airflow 认为任务需要重新运行。

如果任务本身不是幂等的，可能会造成重复写入。

例如：

1. 重复写 MySQL。
2. 重复写 Hive 分区。
3. 重复发送邮件。
4. 重复调用外部接口。

所以清理前要先确认任务是否支持重跑。

十七、日志页面打不开怎么办

任务日志页面打不开，一般从三个方向排查。

17.1 确认任务是否真的运行过

如果任务还没有运行，日志自然不存在。

先查：

SELECT dag_id,
       task_id,
       run_id,
       state,
       try_number,
       start_date,
       end_date
FROM task_instance
WHERE dag_id = 'spark_layer_demo'
ORDER BY start_date DESC;

17.2 确认本地日志目录

当前环境日志目录：

/home/parallels/airflow234/logs/

可以查找某个任务日志：

$ find /home/parallels/airflow234/logs -path '*spark_layer_demo*' -type f | head

如果本地文件存在，但是页面打不开，可能是 Webserver 读取日志路径或权限有问题。

17.3 确认日志权限

如果之前用 root 启动过 Airflow，再用普通用户启动，可能出现日志文件权限不一致。

可以检查：

$ ls -lh /home/parallels/airflow234/logs

如果部分文件属于 root，普通用户可能无法读取。

可以按实际情况修复权限：

$ sudo chown -R parallels:parallels /home/parallels/airflow234/logs

这里需要注意，不要在不了解目录归属的情况下对系统目录乱用 chown -R。

十八、元数据库连接异常

如果健康接口中 metadatabase 不健康，或者命令执行时报数据库连接错误，先检查配置。

查看当前连接串：

$ grep '^sql_alchemy_conn' /home/parallels/airflow234/airflow.cfg

当前环境类似：

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

确认 MySQL 服务：

$ systemctl status mysql --no-pager

确认端口：

$ ss -lntp | grep 3306

确认账号能连接：

$ mysql -uairflow -pairflow_pass_234 -D airflow_db -e "SELECT 1;"

如果是远程连接，还需要检查：

1. MySQL bind-address。
2. 用户 Host 是否允许远程。
3. 防火墙是否放开 3306。
4. 客户端驱动是否匹配。

十九、邮件报警配置

Word 文档中也提到了邮件服务器配置。

Airflow 中邮件相关配置通常在 airflow.cfg 的 [smtp] 段。

例如：

[smtp]
smtp_host = smtp.qq.com
smtp_starttls = True
smtp_ssl = False
smtp_user = xxx@qq.com
smtp_password = 邮箱授权码
smtp_port = 587
smtp_mail_from = xxx@qq.com

DAG 中可以配置：

default_args = {
    "email": ["xxx@qq.com"],
    "email_on_failure": True,
    "email_on_retry": False,
}

需要注意：

1. QQ 邮箱需要开启 SMTP 服务。
2. 密码通常不是登录密码，而是授权码。
3. 测试环境可以先关闭邮件，避免频繁发送。
4. 正式环境建议使用统一告警系统，而不是每个 DAG 各写一套通知逻辑。

Word 文档中的示例使用了 QQ 邮箱 SMTP，端口是 587，这里容易把 smtp 写成 stmps。实际配置段和参数名仍然是 [smtp]、smtp_host、smtp_port 这些字段。

如果邮件没有发出去，先看 Scheduler 或任务日志中是否有 SMTP 报错。

19.1 在 DAG 中增加邮件通知任务

Word 文档中还给了一个 EmailOperator 示例：在 dwd -> dws -> ads 执行完成后，再接一个邮件通知任务。

在 Airflow 2.3.4 中，可以使用下面的导包方式：

from airflow.operators.email import EmailOperator

示例：

email = EmailOperator(
    task_id="email",
    to="example@example.com",
    subject="airflow task finished",
    html_content="<h1>spark layer demo finished</h1>",
)

dwd >> dws >> ads >> email

这段代码表达的是：

dwd -> dws -> ads -> email

也就是三个数据处理任务执行完成后，再发送邮件通知。

如果希望上游任务失败时也执行通知任务，需要结合 trigger_rule 调整触发规则，例如：

email = EmailOperator(
    task_id="email",
    to="example@example.com",
    subject="airflow task finished",
    html_content="<h1>spark layer demo finished</h1>",
    trigger_rule="all_done",
)

这样无论上游成功还是失败，只要上游都结束，邮件任务就会被调度。

是否使用 all_done 要看业务需要。如果只希望成功后发通知，保持默认的 all_success 即可。

二十、常用排查 SQL 汇总

20.1 查看 DAG 状态

SELECT dag_id,
       is_paused,
       is_active,
       last_parsed_time,
       has_import_errors,
       next_dagrun,
       next_dagrun_create_after,
       fileloc
FROM dag
ORDER BY dag_id;

20.2 查看某个 DAG 的 DAG Run

SELECT dag_id,
       run_id,
       run_type,
       state,
       execution_date,
       start_date,
       end_date
FROM dag_run
WHERE dag_id = 'spark_layer_demo'
ORDER BY execution_date DESC;

20.3 查看某个 DAG 的任务状态

SELECT dag_id,
       task_id,
       run_id,
       state,
       try_number,
       max_tries,
       start_date,
       end_date,
       duration,
       operator
FROM task_instance
WHERE dag_id = 'spark_layer_demo'
ORDER BY start_date DESC;

20.4 查看 queued 任务

SELECT dag_id,
       task_id,
       run_id,
       queued_dttm,
       queue,
       pool,
       pool_slots
FROM task_instance
WHERE state = 'queued'
ORDER BY queued_dttm;

20.5 查看 failed 和 up_for_retry 任务

SELECT dag_id,
       task_id,
       run_id,
       state,
       try_number,
       max_tries,
       start_date,
       end_date
FROM task_instance
WHERE state IN ('failed', 'up_for_retry')
ORDER BY start_date DESC;

20.6 查看 Scheduler 心跳

SELECT id,
       state,
       job_type,
       start_date,
       latest_heartbeat,
       executor_class,
       hostname
FROM job
WHERE job_type = 'SchedulerJob'
ORDER BY latest_heartbeat DESC;

20.7 查看 DAG 导入错误

SELECT timestamp,
       filename,
       stacktrace
FROM import_error
ORDER BY timestamp DESC;

二十一、常用命令汇总

查看 Airflow 版本：

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

查看配置：

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

查看 DAG：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list

查看导入错误：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-import-errors

查看任务：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks list spark_layer_demo
$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks list spark_layer_demo --tree

手动触发：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags trigger spark_layer_demo

查看运行记录：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow dags list-runs -d spark_layer_demo

测试任务：

$ AIRFLOW_HOME=/home/parallels/airflow234 conda run -n airflow234 airflow tasks test spark_layer_demo dwd 2023-07-12

查看健康状态：

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

二十二、总结

Airflow 运维排查最重要的不是记住某一个命令，而是形成固定顺序。

我通常会按下面顺序处理：

1. 页面打不开，先查 Webserver 和端口。
2. 页面提示 Scheduler 没心跳，先查 Scheduler 进程、健康接口和 job 表。
3. DAG 看不到，先查 dags_folder、Python 语法、list-import-errors 和 import_error 表。
4. DAG 不自动运行，先查 is_paused、schedule_interval、start_date、catchup 和 dag_run。
5. 任务一直 queued，先查 Scheduler、Executor、Pool 和并发限制。
6. 任务一直 running，先确认实际进程是否还在，再看日志和任务状态。
7. 任务失败，先查 task_instance，再看 task_fail 和任务日志。
8. 日志打不开，先确认任务是否运行过，再查日志目录和权限。

前面第四篇已经整理过元数据库表结构，这一篇更多是把这些表用到具体排查场景中。

对于单机 Airflow 环境，只要能熟练使用下面几类信息，日常问题基本都能定位：

进程状态
健康接口
Airflow CLI
任务日志
元数据库 SQL

到这里，Airflow 从安装、配置、DAG 编写到运维排查，整个单机学习链路就基本整理完整了。

参考资料

1. Apache Airflow 2.3.4 CLI Reference
2. Apache Airflow 2.3.4 Logging for Tasks
3. Apache Airflow 2.3.4 Checking Airflow Health Status