Amazon MWAA 環境での Aurora PostgreSQL データベースのクリーンアップ

重要な考慮事項

依存テーブルの動作

でテーブルを指定すると--tables、コマンドには、指定されたテーブルと外部キー関係を持つ依存 (子) テーブルが自動的に含まれます。外部キーの制約を満たすために、子テーブルレコードが最初に削除され、次に親テーブルレコードが削除されます。例えば、 を指定するとtask_instance、これらのテーブルは外部キーdag_runを介して参照deadlineされるため、task_state_store、、task_instance_historyxcom、、および --tables dag_runもクリーンアップされます。

次の表は、依存関係チェーンをまとめたものです。

依存関係のないテーブル (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
    )