在 Amazon MWAA 环境中清理 Aurora PostgreSQL 数据库

重要注意事项

依赖表行为

当您使用指定表时--tables，该命令会自动包括与指定表具有外键关系的所有依赖（子）表。为了满足外键约束，首先删除子表记录，然后删除父表记录。例如，指定--tables dag_run还会清除task_instance、、task_instance_historyxcom、和task_state_store，deadline因为这些表dag_run通过外键引用。

下表汇总了依赖链。

没有依赖项（例如log、、jobimport_error、sla_miss）的表在指定时将单独清除。

--dry-run在提交清理之前，使用它可以确切查看哪些表和有多少行会受到影响。

的可用参数 airflow db clean

下表描述了的可用参数airflow db clean。

有关可用参数的更多信息，请参阅 Apache Airflow 网站上的 CLI 和环境变量参考。

代码示例

以下示例显示如何airflow db clean通过亚马逊 MWAA CLI 终端节点进行调用。有关创建 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
    )