高级日志记录配置¶

并非所有配置选项都可从 airflow.cfg 文件获得。配置文件描述了如何为任务配置日志记录，因为任务生成的日志不仅默认情况下记录在单独的文件中，而且还必须可以通过 Web 服务器访问。

默认情况下，标准气流组件日志将写入 $AIRFLOW_HOME/logs 目录，但您还可以通过覆盖可通过提供自定义日志记录配置对象进行配置的 Python 记录器配置对其进行自定义和配置。您还可以创建和使用特定运算符和任务的日志记录配置。

某些配置选项要求覆盖日志记录配置类。您可以通过复制 Airflow 的默认配置并对其进行修改以满足您的需求来执行此操作。

可以在 airflow_local_settings.py 模板中看到默认配置，您可以在其中看到使用的记录器和处理程序。

有关如何配置本地设置的详细信息，请参阅配置本地设置。

除了可以通过 airflow.cfg 在此处配置的自定义记录器和处理程序之外，Airflow 中的日志记录方法遵循通常的 Python 日志记录约定，即 Python 对象记录到遵循 <package>.<module_name> 命名约定的记录器。

您可以在 Python 日志记录文档中阅读有关标准 Python 日志记录类（记录器、处理程序、格式化程序）的更多信息。

创建自定义日志记录类¶

可以通过 airflow.cfg 文件中的 logging_config_class 选项配置日志记录类。此配置应指定与 logging.config.dictConfig() 兼容的配置的导入路径。如果你的文件是标准导入位置，则应设置 PYTHONPATH 环境变量。

按照以下步骤启用自定义日志记录配置类

首先将环境变量设置为已知目录，例如 ~/airflow/
export PYTHONPATH=~/airflow/
创建一个目录来存储配置文件，例如 ~/airflow/config

创建名为 ~/airflow/config/log_config.py 的文件，并添加以下内容

from copy import deepcopy
from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG

LOGGING_CONFIG = deepcopy(DEFAULT_LOGGING_CONFIG)

在文件的末尾，添加代码以修改默认字典配置。

更新 $AIRFLOW_HOME/airflow.cfg 以包含

[logging]
logging_config_class = log_config.LOGGING_CONFIG

如果你计划仅扩展/更新启用了远程日志记录的配置，你还可以将 logging_config_class 与远程日志记录一起使用。然后，深度复制的字典将包含为你生成的远程日志记录配置，并且你的修改将在添加远程日志记录配置后应用

[logging]
remote_logging = True
logging_config_class = log_config.LOGGING_CONFIG

重新启动应用程序。

有关 Python 和 Airflow 如何管理模块的详细信息，请参阅模块管理。

注意

你可以覆盖组件的标准日志和“任务”日志的处理方式。

针对算子、钩子和任务的自定义记录器¶

你可以创建自定义日志记录处理程序，并将其应用于特定的算子、钩子和任务。默认情况下，算子和钩子记录器是 airflow.task 记录器的子项：它们分别遵循命名约定 airflow.task.operators.<package>.<module_name> 和 airflow.task.hooks.<package>.<module_name>。在创建自定义日志记录类后，你可以为它们分配特定的记录器。

针对 SQLExecuteQueryOperator 和 HttpHook 的自定义日志记录示例

from copy import deepcopy
from pydantic.utils import deep_update
from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG

LOGGING_CONFIG = deep_update(
    deepcopy(DEFAULT_LOGGING_CONFIG),
    {
        "loggers": {
            "airflow.task.operators.airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator": {
                "handlers": ["task"],
                "level": "DEBUG",
                "propagate": True,
            },
            "airflow.task.hooks.airflow.providers.http.hooks.http.HttpHook": {
                "handlers": ["task"],
                "level": "WARNING",
                "propagate": False,
            },
        }
    },
)

您还可以使用 logger_name 属性为 Dag 的任务设置自定义名称。如果多个任务使用相同的运算符，但您想禁用某些任务的日志记录，这会很有用。

自定义记录器名称示例

# In your Dag file
SQLExecuteQueryOperator(..., logger_name="sql.big_query")

# In your custom `log_config.py`
LOGGING_CONFIG = deep_update(
    deepcopy(DEFAULT_LOGGING_CONFIG),
    {
        "loggers": {
            "airflow.task.operators.sql.big_query": {
                "handlers": ["task"],
                "level": "WARNING",
                "propagate": True,
            },
        }
    },
)