Airflow 1.10.9 Docker环境下自定义日志加DAG ID不生效咨询

原配置失效的核心原因 #

• 语法混用错误：配置中的{{ }}属于Jinja模板语法，而LOG_FORMAT参数是直接传入Python标准logging模块的格式串，不经过Jinja渲染环节，这类无效语法会直接导致格式解析失败，Airflow会自动回退到默认日志格式。
• 占位符规则不匹配：Python logging模块仅支持LogRecord顶层属性的%(key)s形式取值，不支持%(ti.dag_id)s这类嵌套对象的点路径引用，即便格式解析成功，也无法直接从TaskInstance实例中取出dag_id值。
• 字段缺失问题：Airflow 1.10.9版本默认不会将dag_id作为独立字段挂载到LogRecord对象上，就算格式串写对了，也没有对应字段可以输出。

分步实现流程 #

1. 自定义日志过滤器注入dag_id字段 #

首先创建airflow_local_settings.py文件，用于实现自定义日志过滤逻辑，将dag_id从任务上下文提取为LogRecord的顶层字段：

from logging import Filter

class DAGContextFilter(Filter):
    def filter(self, record):
        # 从日志关联的任务实例中提取dag_id，非任务上下文的全局日志默认标记为global
        task_instance = getattr(record, "task_instance", None)
        record.dag_id = task_instance.dag_id if task_instance else "global/scheduler"
        return True

# 给所有日志处理器注册过滤器
from airflow import settings
for handler in settings.LOGGING_CONFIG["handlers"].values():
    handler.setdefault("filters", []).append("dag_id_filter")
settings.LOGGING_CONFIG["filters"]["dag_id_filter"] = {
    "()": "airflow_local_settings.DAGContextFilter"
}

2. 配置正确的日志格式串 #

修正原配置的格式错误，去掉无效的Jinja大括号，使用顶层的dag_id占位符，最终格式串为：
[%(asctime)s] %(filename)s:%(lineno)d %(levelname)s - %(dag_id)s - %(message)s

3. Docker 环境的环境变量配置 #

基于Airflow 1.10.9的环境变量映射规则（双下划线分隔配置节与配置项），配置以下环境变量即可，无需修改容器内的本地配置文件：

• 配置日志格式：
  AIRFLOW__CORE__LOG_FORMAT=[%(asctime)s] %(filename)s:%(lineno)d %(levelname)s - %(dag_id)s - %(message)s
• 配置Python加载路径，让Airflow能识别自定义的airflow_local_settings.py：
  1. 将上述写好的airflow_local_settings.py文件挂载到容器内的/opt/airflow/config/目录
  2. 新增环境变量 PYTHONPATH=/opt/airflow/config

验证效果 #

配置完成重启所有Airflow组件（scheduler、worker、webserver）后，所有日志行都会自动携带对应上下文的dag_id，非任务运行产生的全局调度日志会统一标记为global/scheduler。

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