如何修改Airflow/Composer的日志级别？

这问题我太熟了——Airflow默认的info级日志确实容易把日志文件撑得满满当当，完全不符合实际排查需求。下面给你几种不同场景下的调整方案，按需选就行：

直接调整Airflow的核心配置文件airflow.cfg，找到[logging]区块，修改以下参数：

[logging]
# 全局Airflow日志级别，可选DEBUG/INFO/WARNING/ERROR/CRITICAL
logging_level = WARNING
# 如果用了Flask App Builder（Airflow 2.x默认），还要调整web UI的日志级别
fab_logging_level = WARNING

修改后需要重启Airflow的所有服务（webserver、scheduler、worker）才会生效。

不想全局降低日志级别？可以只给特定DAG或任务设置更严格的日志等级，避免影响其他业务：

from airflow import DAG
from airflow.operators.python import PythonOperator
import logging
from datetime import datetime

# 获取Airflow任务专用的日志器
task_logger = logging.getLogger("airflow.task")

def my_task_logic():
    # 任务内部也可以单独调整日志器级别
    local_logger = logging.getLogger(__name__)
    local_logger.warning("这条警告会被正常输出")
    local_logger.info("这条info日志会被过滤掉")

with DAG(
    dag_id="low_noise_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
) as dag:
    # 给当前DAG的所有任务统一设置日志级别为WARNING
    task_logger.setLevel(logging.WARNING)
    
    demo_task = PythonOperator(
        task_id="filtered_log_task",
        python_callable=my_task_logic
    )

如果是BashOperator这类非Python算子，同样可以通过调整airflow.task日志器的级别来控制输出。

有时候是Airflow的某个组件（比如DAG处理器、数据库Hook）或者第三方依赖日志刷屏，这时候可以单独给这些日志器设级别，在airflow.cfg的[logging]区块添加：

[logging]
# 限制Postgres Hook的日志级别
logger_postgres.level = WARNING
# 限制Kubernetes Executor的日志输出
logger_kubernetes.level = WARNING
# 保留DAG处理的info日志（如果需要监控DAG加载情况）
logger_dag_processing.level = INFO

Airflow内置了很多带logger_前缀的配置项，对应不同组件的日志器，你可以在配置文件里找到相关注释。

如果上面的方式不够灵活，比如要自定义日志格式、按级别拆分日志文件，可以用自定义的Python日志配置。首先在airflow.cfg里指定配置类：

[logging]
logging_config_class = my_custom_logging.LOGGING_CONFIG

然后创建my_custom_logging.py文件（要放在Airflow的PYTHONPATH路径下），编写自定义配置：

import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "formatter": "standard",
            "level": "WARNING"
        },
        "file": {
            "class": "logging.handlers.RotatingFileHandler",
            "formatter": "standard",
            "level": "WARNING",
            "filename": "/opt/airflow/logs/airflow_main.log",
            "maxBytes": 10*1024*1024,  # 10MB
            "backupCount": 5
        }
    },
    "root": {
        "handlers": ["console", "file"],
        "level": "WARNING"
    },
    # 单独给特定日志器设置规则
    "loggers": {
        "airflow.task": {
            "level": "WARNING",
            "handlers": ["console", "file"],
            "propagate": False
        },
        "airflow.dag_processing": {
            "level": "INFO",
            "handlers": ["console", "file"],
            "propagate": False
        }
    }
}

这种方式能完全掌控日志的输出逻辑，适合复杂的日志管理需求。

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