如何自定义Airflow日志并添加元信息后导入ElasticSearch？

一、DAG任务日志的自定义（基于Python Logging Filter） #

Airflow任务日志完全基于Python logging体系，通过自定义Logging Filter可以轻松将dag_id、task_id、owner等元数据注入每条日志：

1. 编写自定义Filter类，从Airflow上下文提取元数据
   任务运行时，Airflow会将当前任务实例绑定到上下文，直接通过Airflow 2.x+的context_get()获取元数据更稳定：

   import logging
   from airflow.context import context_get
   
   class AirflowTaskMetadataFilter(logging.Filter):
       def filter(self, record):
           ctx = context_get()
           if ctx:
               record.dag_id = ctx.dag_id
               record.task_id = ctx.task_id
               record.owner = ctx.dag.owner
               record.execution_date = ctx.execution_date.isoformat()
           else:
               # 非任务上下文场景设默认值
               record.dag_id = "unknown"
               record.task_id = "unknown"
               record.owner = "unknown"
               record.execution_date = "unknown"
           return True
   

2. 修改Airflow日志配置，挂载Filter并更新日志格式
   找到Airflow的logging_config.py（默认路径$AIRFLOW_HOME/config），在配置中添加Filter、绑定到任务日志Handler，并修改日志格式为结构化JSON（方便Fluentd解析）：

   # 配置FILTERS
   FILTERS = {
       # ... 保留原有Filter
       "airflow_task_metadata": {
           "()": "your.custom.module.AirflowTaskMetadataFilter",  # 替换为你的模块路径
       },
   }
   
   # 配置HANDLERS，给task日志添加Filter
   HANDLERS = {
       # ... 其他Handler
       "task": {
           "class": "airflow.utils.log.file_task_handler.FileTaskHandler",
           "formatter": "structured_airflow",
           "filters": ["airflow_task_metadata"],
           # ... 保留原有路径、轮转配置
       },
   }
   
   # 配置结构化FORMATTER，用JSON格式输出元数据
   FORMATTERS = {
       "structured_airflow": {
           "format": '%(asctime)s %(dag_id)s %(task_id)s %(owner)s %(levelname)s %(message)s',
           "class": "pythonjsonlogger.jsonlogger.JsonFormatter"
       },
   }
   

二、调度器日志的自定义 #

调度器日志没有直接的任务上下文，可通过解析日志内容提取DAG元数据：

1. 编写调度器专属Filter
   调度器日志中会包含"Processing dag_id: xxx"这类关键字段，用正则提取即可：

   import logging
   import re
   
   class AirflowSchedulerMetadataFilter(logging.Filter):
       def filter(self, record):
           msg = record.getMessage()
           dag_match = re.search(r'dag_id: (\w[\w\-_]+)', msg)
           record.dag_id = dag_match.group(1) if dag_match else "unknown_scheduler_dag"
           record.task_id = "scheduler"
           record.owner = "airflow_scheduler"
           return True
   

2. 绑定到调度器日志Handler
   在logging_config.py中更新调度器Handler配置：

   HANDLERS = {
       # ... 其他Handler
       "scheduler": {
           "class": "logging.handlers.RotatingFileHandler",
           "formatter": "structured_airflow",
           "filters": ["airflow_scheduler_metadata"],
           # ... 保留原有路径、轮转配置
       },
   }
   
   FILTERS = {
       # ... 保留原有Filter
       "airflow_scheduler_metadata": {
           "()": "your.custom.module.AirflowSchedulerMetadataFilter",
       },
   }
   

三、验证与注意事项 #

• 确保自定义Filter所在的Python模块在Airflow的PYTHONPATH中，可通过设置环境变量export PYTHONPATH=$PYTHONPATH:/path/to/your/module实现
• 重启Airflow服务后，运行测试DAG，查看日志文件是否已包含预期的元数据字段
• 若使用Fluentd采集日志，JSON格式的日志可直接被解析为字段，无需额外正则匹配，直接映射到ElasticSearch的字段即可
• 不同Airflow版本的调度器日志格式可能略有差异，需调整正则表达式适配你的版本

内容的提问来源于stack exchange，提问作者이윤택