Nettoyage de la base de données Aurora PostgreSQL dans un environnement Amazon MWAA

Considérations importantes

Comportement des tables dépendantes

Lorsque vous spécifiez une table avec--tables, la commande inclut automatiquement toutes les tables dépendantes (enfants) qui ont des relations de clé étrangère avec la table spécifiée. Les enregistrements de la table enfant sont d'abord supprimés, puis les enregistrements de la table parent, pour satisfaire aux contraintes de clé étrangère. Par exemple, la spécification --tables dag_run permet également de nettoyertask_instance, task_instance_history xcomtask_state_store, et deadline parce que ces tables font référence dag_run à des clés étrangères.

Le tableau suivant récapitule les chaînes de dépendance.

Les tables sans dépendances (telles quelog, jobimport_error,sla_miss) sont nettoyées de manière isolée lorsque cela est spécifié.

--dry-runÀ utiliser pour voir exactement quelles tables et combien de lignes seraient affectées avant de procéder à un nettoyage.

Paramètres disponibles pour airflow db clean

Le tableau suivant décrit les paramètres disponibles pourairflow db clean.

Pour plus d'informations sur les paramètres disponibles, consultez la référence des variables CLI et env sur le site Web d'Apache Airflow.

Exemples de code

Les exemples suivants montrent comment appeler airflow db clean via le point de terminaison de la CLI Amazon MWAA. Pour plus d'informations sur la création de jetons CLI, consultezCréation d'un jeton CLI Apache Airflow.

À l'aide d'un script 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}")

Exemple de fonctionnement à sec (première étape recommandée)

Avant d'effectuer un véritable nettoyage, exécutez with --dry-run pour voir ce qui serait supprimé :

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
    )