Amazon MWAA 환경에서 Aurora PostgreSQL 데이터베이스 정리

중요 고려 사항

종속 테이블 동작

로 테이블을 지정하면 명령--tables에 지정된 테이블과 외래 키 관계가 있는 종속(하위) 테이블이 자동으로 포함됩니다. 하위 테이블 레코드는 먼저 삭제된 다음 상위 테이블 레코드를 삭제하여 외래 키 제약 조건을 충족합니다. 예를 들어를 지정하면 task_instance, task_instance_history, task_state_store, 및 xcom--tables dag_run도 정리됩니다. 이러한 테이블은 외래 키를 dag_run 통해 참조deadline하기 때문입니다.

다음 표에는 종속성 체인이 요약되어 있습니다.

종속 항목이 없는 테이블(예: log, job, import_error, sla_miss)은 지정된 경우 격리된 상태로 정리됩니다.

--dry-run를 사용하여 정리를 커밋하기 전에 영향을 받을 테이블과 행 수를 정확히 확인합니다.

에 사용 가능한 파라미터 airflow db clean

다음 표에서는에 사용할 수 있는 파라미터를 설명합니다airflow db clean.

사용 가능한 파라미터에 대한 자세한 내용은 Apache Airflow 웹 사이트의 CLI 및 env 변수 참조를 참조하세요.

코드 샘플

다음 예제에서는 Amazon MWAA CLI 엔드포인트를 airflow db clean 통해를 호출하는 방법을 보여줍니다. CLI 토큰 생성에 대한 자세한 내용은 섹션을 참조하세요Apache Airflow CLI 토큰 생성.

Python 스크립트 사용:

import boto3
import base64
import requests

# Replace with your environment name and AWS Region
mwaa_env_name = "YOUR_ENVIRONMENT_NAME"
region = "YOUR_REGION"

# Configure cleanup scope
clean_before_timestamp = "2025-06-01T00:00:00+00:00"
tables = "dag_run,task_instance,log,job,xcom"

# Build the Airflow CLI command
# -y flag is required to skip interactive confirmation prompt
airflow_cmd = f"db clean --clean-before-timestamp {clean_before_timestamp} --tables {tables} -y"

# Create a CLI token
client = boto3.client("mwaa", region_name=region)
cli_token_response = client.create_cli_token(Name=mwaa_env_name)

cli_token = cli_token_response["CliToken"]
web_server_hostname = cli_token_response["WebServerHostname"]

# Invoke the Airflow CLI through the MWAA endpoint
url = f"https://{web_server_hostname}/aws_mwaa/cli"
response = requests.post(
    url,
    headers={
        "Authorization": f"Bearer {cli_token}",
        "Content-Type": "text/plain",
    },
    data=airflow_cmd,
)

# Parse and display the results
stdout_message = base64.b64decode(response.json()["stdout"]).decode("utf-8")
stderr_message = base64.b64decode(response.json()["stderr"]).decode("utf-8")

print(f"Status code: {response.status_code}")
print(f"stdout:\n{stdout_message}")
print(f"stderr:\n{stderr_message}")

드라이 런 예제(권장 첫 번째 단계)

실제 정리를 수행하기 전에를와 함께 실행--dry-run하여 삭제할 내용을 확인합니다.

AIRFLOW_CMD="db clean --clean-before-timestamp 2025-06-01T00:00:00+00:00 --tables dag_run,task_instance --dry-run -y"

from airflow import DAG
from airflow.models.param import Param
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago

from datetime import datetime, timedelta

# Note: Database commands might time out if running longer than 5 minutes. If this occurs, please increase the MAX_AGE_IN_DAYS (or change 
# timestamp parameter to an earlier date) for initial runs, then reduce on subsequent runs until the desired retention is met.

MAX_AGE_IN_DAYS = 30

# To clean specific tables, please provide a comma-separated list per 
# https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#clean
# A value of None will clean all tables

TABLES_TO_CLEAN = None

with DAG(
    dag_id="clean_db_dag", 
    schedule_interval=None, 
    catchup=False, 
    start_date=days_ago(1),
    params={
        "timestamp": Param(
            default=(datetime.now()-timedelta(days=MAX_AGE_IN_DAYS)).strftime("%Y-%m-%d %H:%M:%S"),
            type="string",
            minLength=1,
            maxLength=255,
        ),     
    }   
) as dag:
    if TABLES_TO_CLEAN:
        bash_command="airflow db clean --clean-before-timestamp '{{ params.timestamp }}' --tables '"+TABLES_TO_CLEAN+"' --skip-archive --yes"
    else:
        bash_command="airflow db clean --clean-before-timestamp '{{ params.timestamp }}' --skip-archive --yes"

    cli_command = BashOperator(
        task_id="bash_command",
        bash_command=bash_command
    )