Pulizia del database Aurora PostgreSQL in un ambiente Amazon MWAA

Considerazioni importanti

Comportamento dipendente della tabella

Quando si specifica una tabella con--tables, il comando include automaticamente tutte le tabelle dipendenti (secondarie) che hanno relazioni di chiave esterna con la tabella specificata. I record della tabella secondaria vengono eliminati prima, quindi i record della tabella principale, per soddisfare i vincoli di chiave esterna. Ad esempio, specificando si eliminano --tables dag_run anche i task_instance valori, task_instance_history xcomtask_state_store, e deadline perché queste tabelle fanno riferimento dag_run tramite chiavi esterne.

La tabella seguente riassume le catene di dipendenze.

Le tabelle senza elementi dipendenti (ad esempiolog,, jobimport_error,sla_miss) vengono pulite isolatamente quando specificato.

Usalo --dry-run per vedere esattamente quali tabelle e quante righe sarebbero interessate prima di eseguire una pulizia.

Parametri disponibili per airflow db clean

La tabella seguente descrive i parametri disponibili perairflow db clean.

Per ulteriori informazioni sui parametri disponibili, consultate il riferimento alle variabili CLI e env sul sito Web di Apache Airflow.

Esempi di codice

Gli esempi seguenti mostrano come richiamare airflow db clean tramite l'endpoint CLI di Amazon MWAA. Per ulteriori informazioni sulla creazione di token CLI, consulta. Creazione di un token CLI Apache Airflow

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

Esempio di funzionamento a secco (primo passaggio consigliato)

Prima di eseguire una pulizia effettiva, esegui --dry-run per vedere cosa verrebbe eliminato:

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
    )