Limpieza de bases de datos de Aurora PostgreSQL en un entorno de Amazon MWAA

Consideraciones importantes

Comportamiento dependiente de la tabla

Al especificar una tabla con--tables, el comando incluye automáticamente todas las tablas dependientes (secundarias) que tengan relaciones de clave externa con la tabla especificada. Los registros de la tabla secundaria se eliminan primero y, a continuación, los registros de la tabla principal, para cumplir con las restricciones de clave externa. Por ejemplo, especificar --tables dag_run también limpiatask_instance,, task_instance_history xcomtask_state_store, y deadline porque estas tablas hacen referencia dag_run a través de claves externas.

En la tabla siguiente se resumen las cadenas de dependencias.

Las tablas sin dependientes (comolog,, jobimport_error,sla_miss) se limpian de forma aislada cuando se especifican.

--dry-runUtilícelo para ver exactamente qué tablas y cuántas filas se verían afectadas antes de iniciar una limpieza.

Parámetros disponibles para airflow db clean

En la siguiente tabla se describen los parámetros disponibles paraairflow db clean.

Para obtener más información sobre los parámetros disponibles, consulte la referencia a las variables de entorno y CLI en el sitio web de Apache Airflow.

Ejemplos de código

Los siguientes ejemplos muestran cómo invocar a airflow db clean través del punto de enlace de la CLI de Amazon MWAA. Para obtener más información sobre la creación de tokens de CLI, consulte. Creación de un token de la CLI de Apache Airflow

Uso de una secuencia de comandos de 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}")

Ejemplo de ejecución en seco (primer paso recomendado)

Antes de realizar una limpieza real, ejecuta con --dry-run para ver qué se eliminaría:

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
    )