[Airflow] airflow_db_cleanup Dag 사용하기

Apache Airflow

[Airflow] airflow_db_cleanup Dag 사용하기

비번변경 2024. 4. 19. 01:17

개요

2023.08.08 - [Airflow] DB Clean에서 Airflow Meta DB를 관리하는 두 가지 방법을 소개했었다.

airflow_db_cleanup Dag 사용
airflow db clean 명령 사용

기존 글에서는 airflow 명령어를 사용한 방법에 초점을 맞췄는데, 이번 글에서는 airflow_db_cleanup Dag를 사용한 방법에 초점을 맞춰서 정리해보려고 한다.

airflow_db_cleanup

airflow_db_cleanup Dag는 https://github.com/teamclairvoyant/airflow-maintenance-dags 에서 개발되었다. 하지만 저장소 이력을 보면 전혀 관리가 되어 있지 않은 상태로 보이는데, 좀 더 찾아보니 GCP의 https://github.com/GoogleCloudPlatform/python-docs-samples에서 해당 저장소의 db_cleanup 코드를 포크하여 관리하고 있는 것을 확인할 수 있었다.

해당 저장소에서 관리되고 있는 소스를 사용해 Airflow Meta DB를 관리해보자.

사용 방법

1. dag_folder에 airflow_db_cleanup.py 복사

wget https://raw.githubusercontent.com/GoogleCloudPlatform/python-docs-samples/main/composer/workflows/airflow_db_cleanup.py

2. 필요 전역 변수 설정

기본적으로 아래의 변수를 지정하도록 안내되어 있다. 코드의 주석을 참고하여 달리 필요한 설정도 편집한다.

# 실행 주기
SCHEDULE_INTERVAL = "@daily"
# DAG_OWNER 지정
DAG_OWNER_NAME = "operations"
# 실행 실패 시 메일 경고를 전송한 이메일 주소 목록
ALERT_EMAIL_ADDRESSES = []
# Airflow 버전
AIRFLOW_VERSION = airflow_version.split(".")
# DB 삭제 여부
ENABLE_DELETE = True

2. DATABASE_OBJECTS 편집

airflow_db_cleanup로 처리할 항목은 DATABASE_OBJECTS 변수로 지정되어 있다. 필요 시 처리할 항목을 추가하거나 주석으로 막아 제외할 수 있다. 항목을 추가할 때는 아래의 설정을 지정하도록 한다.

{
    # airflow Meta DB 내 테이블에 해당하는 airflow.models에서 임포트되는 Model
    "airflow_db_model": DagRun,
    # 데이터 삭제 시 최대 날짜를 계산하는데 사용할 모델 또는 테이블 내 컬럼
    "age_check_column": DagRun.execution_date,
    # 마지막 실행 인스턴스 보존 여부 결정
    "keep_last": True,
    # 데이터 정리 중 삭제하지 않도록 할 필터 목록
    "keep_last_filters": [DagRun.external_trigger.is_(False)],
    # 데이터베이스 항목에 대해 그룹 연산을 수행할 기준 열
    "keep_last_group_by": DagRun.dag_id,
    }

어지간하면 직접 추가할 일은 없을 것 같다.

3. Admin -> Variables에 변수 airflow_db_cleanup__max_db_entry_age_in_days 추가

Airflow Meta DB의 보관주기를 지정하는 변수를 추가한다.