如何在Airflow启动后仅执行一次初始化代码？

嘿，这个需求太常见了——把初始化代码塞DAG里确实会因为DAG被反复解析、任务反复执行而重复运行，完全没必要。给你几个靠谱的方法，能让你的日志初始化代码只在Airflow启动时跑一次：

方法1：利用Airflow的on_startup配置 #

Airflow允许你在启动服务时指定要执行的命令，你可以把初始化逻辑做成一个单独的脚本，然后在配置中指定：

1. 先写一个初始化脚本init_logger.py：

from utils import logger
logger.initialize()

2. 修改airflow.cfg里的on_startup_command参数：

on_startup_command = python /path/to/init_logger.py

这样每次启动Airflow的webserver、scheduler或者worker时，都会先执行这个脚本，完成日志初始化。

方法2：自定义启动脚本 #

如果你不想改配置文件，可以写一个shell脚本，先执行初始化代码，再启动Airflow服务。比如：

#!/bin/bash
# 先执行日志初始化
python -c "from utils import logger; logger.initialize()"
# 启动Airflow webserver
airflow webserver -p 8080

同理，scheduler的启动脚本也可以这么写。用这个脚本代替直接启动Airflow，就能保证初始化只在服务启动时执行一次。

方法3：通过airflow_local_settings.py加载 #

Airflow启动时会自动加载airflow_local_settings.py文件（如果它在Python路径里），你可以把初始化代码放在这里：

# airflow_local_settings.py
from utils import logger
logger.initialize()

# 保留原有的LocalSettings内容（如果有的话）
from airflow.settings import DEFAULT_LOCAL_SETTINGS
__dict__.update(DEFAULT_LOCAL_SETTINGS.__dict__)

然后把这个文件放到Airflow能找到的地方（比如Airflow安装目录的config文件夹，或者你的项目根目录并确保在Python路径中），这样Airflow进程启动时就会自动执行这段初始化代码。

注意事项 #

• 如果是分布式部署（比如多个worker节点），每个节点的Airflow进程启动时都会执行一次初始化——这其实是合理的，因为每个进程都需要正确配置日志。
• 绝对不要把初始化代码放在DAG文件的顶层，因为Airflow scheduler会定期解析DAG文件（默认每30秒一次），而且每个任务实例运行时也会加载DAG代码，会导致初始化逻辑被频繁执行，浪费资源。

内容的提问来源于stack exchange，提问作者achmanngopf