Aurora PostgreSQL-Datenbankbereinigung in einer Amazon MWAA-Umgebung

Wichtige Überlegungen

Verhalten abhängiger Tabellen

Wenn Sie eine Tabelle mit angeben--tables, schließt der Befehl automatisch alle abhängigen (untergeordneten) Tabellen ein, die Fremdschlüsselbeziehungen zu der angegebenen Tabelle haben. Die Datensätze der untergeordneten Tabelle werden zuerst gelöscht, dann die Datensätze der übergeordneten Tabelle, um Fremdschlüsseleinschränkungen zu erfüllen. Beispielsweise bereinigt die Angabe --tables dag_run auchtask_instance,, und task_instance_history xcomtask_state_store, und deadline weil diese Tabellen dag_run über Fremdschlüssel referenzieren.

In der folgenden Tabelle sind die Abhängigkeitsketten zusammengefasst.

Tabellen ohne abhängige Objekte (wielog,,job,sla_miss) werden isoliert bereinigtimport_error, wenn sie angegeben werden.

Verwenden Sie diese Option, --dry-run um genau zu sehen, welche Tabellen und wie viele Zeilen betroffen wären, bevor Sie zu einer Bereinigung übergehen.

Verfügbare Parameter für airflow db clean

In der folgenden Tabelle werden die verfügbaren Parameter für beschriebenairflow db clean.

Weitere Informationen zu den verfügbaren Parametern finden Sie in der Referenz zu CLI- und env-Variablen auf der Apache Airflow-Website.

Codebeispiele

Die folgenden Beispiele zeigen, wie der Aufruf airflow db clean über den Amazon MWAA-CLI-Endpunkt erfolgt. Weitere Informationen zum Erstellen von CLI-Token finden Sie unter. Erstellen eines Apache Airflow CLI-Tokens

Mit einem Python-Skript:

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

Beispiel für einen Probelauf (empfohlener erster Schritt)

Bevor Sie eine eigentliche Bereinigung durchführen, führen Sie Folgendes aus, --dry-run um zu sehen, was gelöscht werden würde:

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
    )