AWS MWAA 2.6.3：如何通过boto3按run_id标记DAG执行实例为失败？

AWS MWAA的boto3客户端本身没有直接提供修改DAG Run状态的API，但可以通过生成MWAA CLI访问令牌，调用Airflow CLI的dag run fail命令来实现目标操作，具体步骤如下：

步骤说明 #

1. 使用boto3的MWAA客户端生成CLI访问令牌，获取访问MWAA Web服务器的凭证和地址
2. 构造Airflow CLI命令，指定目标DAG ID和run_id
3. 发送HTTP请求执行CLI命令，完成DAG Run状态标记

代码示例 #

import boto3
import requests
from requests.auth import HTTPBasicAuth

# 初始化MWAA客户端
mwaa_client = boto3.client('mwaa')

# 替换为你的MWAA环境名称
environment_name = "your-mwaa-environment-name"
# 替换为目标DAG ID和run_id
dag_id = "your-dag-id"
target_run_id = "target-run-id"

# 1. 创建CLI令牌
response = mwaa_client.create_cli_token(Name=environment_name)
cli_token = response['CliToken']
web_server_hostname = response['WebServerHostname']

# 2. 构造请求参数
cli_command = f"dag run fail --dag-id {dag_id} --run-id {target_run_id}"
url = f"https://{web_server_hostname}/aws_mwaa/cli"

# 3. 发送POST请求执行命令
auth = HTTPBasicAuth('airflow', cli_token)
response = requests.post(
    url,
    auth=auth,
    data={"command": cli_command}
)

# 处理响应
if response.status_code == 200:
    print("DAG Run标记失败成功，响应结果：")
    print(response.text)
else:
    print(f"操作失败，状态码：{response.status_code}，错误信息：{response.text}")

注意事项 #

• 确保执行代码的IAM实体（用户/角色）拥有mwaa:CreateCliToken权限
• Airflow 2.6.3版本支持dag run fail命令，若需验证可在MWAA环境的CLI中执行该命令测试
• 命令中的dag-id和run-id必须准确匹配目标DAG实例，否则会执行失败

内容的提问来源于stack exchange，提问作者kuollam