Limpeza do banco de dados do Aurora PostgreSQL em um ambiente do Amazon MWAA

Considerações importantes

Comportamento dependente da tabela

Quando você especifica uma tabela com--tables, o comando inclui automaticamente qualquer tabela dependente (filha) que tenha relações de chave estrangeira com a tabela especificada. Os registros da tabela secundária são excluídos primeiro, depois os registros da tabela principal, para satisfazer as restrições de chave estrangeira. Por exemplo, especificar --tables dag_run também limpatask_instance,, task_instance_history xcomtask_state_store, e deadline porque essas tabelas fazem referência por dag_run meio de chaves estrangeiras.

A tabela a seguir resume as cadeias de dependência.

Tabelas sem dependentes (comolog,, jobimport_error,sla_miss) são limpas isoladamente quando especificadas.

Use --dry-run para ver exatamente quais tabelas e quantas linhas seriam afetadas antes de se comprometer com uma limpeza.

Parâmetros disponíveis para airflow db clean

A tabela a seguir descreve os parâmetros disponíveis paraairflow db clean.

Para obter mais informações sobre os parâmetros disponíveis, consulte a referência de variáveis CLI e env no site do Apache Airflow.

Exemplos de código

Os exemplos a seguir mostram como invocar airflow db clean por meio do endpoint da CLI do Amazon MWAA. Para obter mais informações sobre a criação de tokens CLI, consulteComo criar um token da CLI do Apache Airflow.

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

Exemplo de operação a seco (primeira etapa recomendada)

Antes de realizar uma limpeza real, execute com --dry-run para ver o que seria excluído:

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
    )