고급 로깅 설정

고급 로깅 설정 (Advanced logging configuration)

이 페이지는 airflow.cfg로는 다룰 수 없는 고급 로깅 설정 방법을 설명해요. 기본 Python logger 설정을 덮어써서 커스텀 로깅 구성을 만들고, 특정 Operator·Hook·Task에 로깅 핸들러를 적용할 수 있어요. logging_config_class 옵션으로 커스텀 로깅 dict 설정을 지정하고, 원격 로깅과 함께 사용하는 방법도 다뤄요.

출처: 문서

본문

모든 설정 옵션이 airflow.cfg 파일에서 제공되는 것은 아니에요. 설정 파일은 Task의 로깅을 구성하는 방법을 설명하는데, Task가 생성하는 로그는 기본적으로 별도 파일에 기록될 뿐 아니라 webserver를 통해서도 접근 가능해야 하기 때문이에요.

기본적으로 표준 Airflow 컴포넌트 로그는 $AIRFLOW_HOME/logs 디렉터리에 기록되지만, 커스텀 로깅 구성 객체를 제공해 Python logger 설정을 덮어써 원하는 대로 커스터마이즈할 수도 있어요. 특정 Operator와 Task에 대한 로깅 구성을 만들어 사용할 수도 있어요.

일부 설정 옵션은 로깅 설정 클래스가 덮어써져야 해요. Airflow의 기본 구성을 복사해 필요에 맞게 수정하면 됩니다.

기본 구성은 airflow_local_settings.py 템플릿에서 볼 수 있고, 거기서 사용되는 logger와 handler를 확인할 수 있어요.

로컬 설정을 구성하는 방법에 대한 자세한 내용은 로컬 설정 구성을 참고해요.

airflow.cfg를 통해 구성할 수 있는 커스텀 logger·handler 외에도 Airflow의 로깅 방식은 일반적인 Python 로깅 규칙을 따르는데, Python 객체는 <package>.<module_name> 명명 규칙을 따르는 logger에 로그를 남겨요.

표준 Python 로깅 클래스(Logger, Handler, Formatter)에 대해 더 자세히 알아보려면 Python logging 문서를 참고해요.

커스텀 로깅 구성 만들기

로깅 클래스 구성은 airflow.cfg 파일의 logging_config_class 옵션을 통해 할 수 있어요. 옵션 이름과 달리 그 값은 logging.config.dictConfig()를 충족하는 dict에 대한 dotted import 경로예요 — Python 클래스가 아니에요. _class 접미사는 과거의 이유로 하위 호환성을 위해 유지돼요. 파일이 표준 import 위치라면 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)
파일 끝에 기본 dict 구성을 수정하는 코드를 추가해요.
$AIRFLOW_HOME/airflow.cfg를 다음으로 업데이트해요:
[logging] logging_config_class = log_config.LOGGING_CONFIG
원격 로깅을 활성화한 채 구성을 그냥 확장·업데이트할 계획이라면 logging_config_class를 원격 로깅과 함께 사용할 수도 있어요. 그러면 deep-copy된 dict에 원격 로깅 구성이 자동 생성되어 담기고, 원격 로깅 설정이 추가된 뒤에 사용자의 수정이 적용돼요:
[logging] remote_logging = True logging_config_class = log_config.LOGGING_CONFIG
애플리케이션을 재시작해요.

Python과 Airflow가 모듈을 어떻게 관리하는지에 대한 자세한 내용은 Modules Management를 참고해요.

Note

컴포넌트의 표준 로그와 "task" 로그를 처리하는 방식을 모두 덮어쓸 수 있어요.

커스텀 로깅 구성과 원격 로깅

[logging] remote_logging = True이고 logging_config_class가 나만의 모듈을 가리키면, 그 모듈에 두 개의 모듈 레벨 속성을 정의해요:

REMOTE_TASK_LOG — Task 로그를 업로드하고 UI를 위해 다시 읽어들이는 RemoteLogIO(또는 RemoteLogStreamIO) 인스턴스.
DEFAULT_REMOTE_CONN_ID — [logging] remote_log_conn_id가 설정되지 않았을 때 사용하는 기본 Airflow 연결 id.

REMOTE_TASK_LOG가 없으면 Airflow가 시작 시 WARNING 하나를 내보내고, UI가 원격 백엔드에서 Task 로그를 다시 읽을 수 없게 돼요.

# ~/airflow/config/log_config.py
from airflow.logging.remote import RemoteLogIO

class MyRemoteLogIO:
    @property
    def processors(self):
        return ()

    def upload(self, path, ti): ...  # upload local log file at ``path`` to your backend

    def read(self, relative_path, ti): ...  # return (source_info, log_messages) for the UI

REMOTE_TASK_LOG: RemoteLogIO | None = MyRemoteLogIO()
DEFAULT_REMOTE_CONN_ID: str | None = "my_remote_conn"

Note

REMOTE_TASK_LOG는 airflow.config_templates.airflow_local_settings에서 재-export하기보다는 나만의 모듈에 정의해요. 그 파일은 deprecated될 예정이에요.

Operator, Hook, Task를 위한 커스텀 logger

커스텀 로깅 핸들러를 만들어 특정 Operator, Hook, Task에 적용할 수 있어요. 기본적으로 Operator와 Hook logger는 airflow.task logger의 자식이에요. 각각 airflow.task.operators.<package>.<module_name>과 airflow.task.hooks.<package>.<module_name> 명명 규칙을 따르죠. 커스텀 로깅 클래스를 만든 뒤 이들에 특정 logger를 할당할 수 있어요.

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,
            },
        }
    },
)

DAG의 Task에 logger_name 속성으로 커스텀 이름을 설정할 수도 있어요. 여러 Task가 같은 Operator를 사용하지만 그중 일부에 대해 로깅을 비활성화하고 싶을 때 유용해요.

커스텀 logger 이름 예제:

# 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,
            },
        }
    },
)

Task의 로그 크기를 제한하고 싶다면 handlers.task.max_bytes 파라미터를 추가할 수 있어요.

Task 크기를 제한하는 예제:

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),
    {
        "handlers": {
            "task": {"max_bytes": 104857600, "backup_count": 1}  # 100MB and keep 1 history rotate log.
        }
    },
)

고급 로깅 설정