本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 访问 Apache Airflow Amazon MWAA 提供了多种访问 Apache Airflow 环境的方法：Apache Airflow 用户界面 (UI) 控制台、Apache Airflow CLI 和 Apache Airflow REST API。你可以使用 Amazon MWAA 控制台在 Apache Airflow 用户界面中访问和调用 DAG，也可以使用亚马逊 APIs MWAA 获取令牌并调用 DAG。本节介绍访问 Apache Airflow UI 所需的权限、如何生成令牌以直接在命令 shell 中调用 Amazon MWAA API，以及 Apache Airflow CLI 中支持的命令。 **Topics** + [ ## 先决条件 ](#access-airflow-ui-prereqs) + [ ## 打开 Apache Airflow UI ](#access-airflow-ui-onconsole) + [ ## 登录 Apache Airflow ](#airflow-access-and-login) + [ # 创建 Apache Airflow Web 服务器访问令牌 ](call-mwaa-apis-web.md) + [ # 为 Apache Airflow Web 服务器设置自定义域 ](configuring-custom-domain.md) + [ # 创建 Apache Airflow CLI 令牌 ](call-mwaa-apis-cli.md) + [ # 使用 Apache Airflow REST API ](access-mwaa-apache-airflow-rest-api.md) + [ # Apache Airflow CLI 命令参考 ](airflow-cli-command-reference.md) ## 先决条件 下一节介绍使用本节中的命令和脚本所需的初步步骤。 ### 访问 + AWS 账户 在 AWS Identity and Access Management (IAM) 中访问中的亚马逊 MWAA 权限策略。[Apache Airflow 用户界面访问策略：亚马逊 MWAAWeb ServerAccess](access-policies.md#web-ui-access) + AWS 账户 在 AWS Identity and Access Management (IAM) 中访问亚马逊 MWAA 权限策略。[完整的 API 和控制台访问政策：Amazon MWAAFull ApiAccess](access-policies.md#full-access-policy) ### AWS CLI AWS Command Line Interface (AWS CLI) 是一个开源工具，您可以使用命令行 shell 中的命令与 AWS 服务进行交互。要完成本节中的步骤，您需要以下满足以下条件： + [AWS CLI — 安装版本 2](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html)。 + [AWS CLI — 使用快速配置`aws configure`](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html)。 ## 打开 Apache Airflow UI 下图显示了 Amazon MWAA 控制台上指向 Apache Airflow UI 的链接。 ![\[本图显示了 Amazon MWAA 控制台上指向 Apache Airflow UI 的链接。\]](http://docs.aws.amazon.com/zh_cn/mwaa/latest/userguide/images/mwaa-console-aa-ui.png) ## 登录 Apache Airflow 你需要你的 AWS 账户 内部 AWS Identity and Access Management (IAM) [Apache Airflow 用户界面访问策略：亚马逊 MWAAWeb ServerAccess](access-policies.md#web-ui-access) 权限才能访问你的 Apache Airflow 用户界面。 **要访问 Apache Airflow UI，请执行以下操作** 1. 在 Amazon MWAA 控制台上打开[环境页面](https://console.aws.amazon.com/mwaa/home#/environments)。 1. 选择环境。 1. 选择**打开 Airflow UI**。 # 创建 Apache Airflow Web 服务器访问令牌 您可以使用本页中的命令创建 Web 服务器访问令牌。访问令牌让您能够访问 Amazon MWAA 环境。例如，您可以获取令牌，然后使用 Amazon MWAA API 以编程方式部署 DAG。下一节包括使用 AWS CLI、bash 脚本、POST API 请求或 Python 脚本创建 Apache Airflow Web 登录令牌的步骤。响应中返回的令牌在 60 秒内有效。 **重要** 自 2025 年 8 月 19 日起，Amazon MWAA 增加了对 IPv6 端点的支持，现在支持 IPv4 和 IPv6 端点。从该日期起，所有新创建的环境都将使用 `.on.aws` 域作为 Airflow 用户界面 (UI)。对于这些新创建的环境，客户必须将其 Airflow UI 从 `.on.aws` 域迁移到 `.amazonaws.com` 域。用于 Web 服务器和数据库的虚拟私有云 (VPC) 端点服务将保持其当前 `.amazonaws.com` 域，无需进行任何更改。 **Contents** + [ ## 先决条件 ](#call-mwaa-apis-web-prereqs) + [ ### 访问 ](#access-airflow-ui-prereqs-access) + [ ### AWS CLI ](#access-airflow-ui-prereqs-cli) + [ ## 使用 AWS CLI ](#create-web-login-token-cli) + [ ## 使用 bash 脚本 ](#create-web-login-token-bash) + [ ## 使用 Python 脚本 ](#create-web-login-token-python) + [ ## 接下来做什么？ ](#mwaa-webcli-next-up) ## 先决条件 下一节介绍了使用本页上的命令和脚本所需的初步步骤。 ### 访问 + 在 AWS Identity and Access Management (IAM) 中访问 [Apache Airflow 用户界面访问策略：亚马逊 MWAAWeb ServerAccess](access-policies.md#web-ui-access) 中的 Amazon MWAA 权限策略的 AWS 账户。 + 在 AWS Identity and Access Management (IAM) 中访问 Amazon MWAA 权限策略 [完整的 API 和控制台访问政策：Amazon MWAAFull ApiAccess](access-policies.md#full-access-policy) 的 AWS 账户。 ### AWS CLI AWS Command Line Interface (AWS CLI) 是一种开源工具，您可以用来在命令行 Shell 中使用命令与 AWS 服务进行交互。要完成本节中的步骤，您需要以下满足以下条件： + [AWS CLI – 安装版本 2](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html)。 + [AWS CLI – 使用 `aws configure` 进行快速配置](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html)。 ## 使用 AWS CLI 以下示例使用 AWS CLI 中的 [create-web-login-token](https://docs.aws.amazon.com/cli/latest/reference/mwaa/create-web-login-token.html) 命令创建 Apache Airflow Web 登录令牌。 ``` aws mwaa create-web-login-token --name YOUR_ENVIRONMENT_NAME ``` ## 使用 bash 脚本 以下示例使用 bash 脚本调用 AWS CLI 中的 [create-web-login-token](https://docs.aws.amazon.com/cli/latest/reference/mwaa/create-web-login-token.html) 命令来创建 Apache Airflow Web 登录令牌。 1. 复制以下代码示例的内容，并在本地另存为 `get-web-token.sh`。 ``` #!/bin/bash HOST=YOUR_HOST_NAME YOUR_URL=https://$HOST/aws_mwaa/aws-console-sso?login=true# WEB_TOKEN=$(aws mwaa create-web-login-token --name YOUR_ENVIRONMENT_NAME --query WebToken --output text) echo $YOUR_URL$WEB_TOKEN ``` 1. 用*红色*占位符代替 `YOUR_HOST_NAME` 和 `YOUR_ENVIRONMENT_NAME`。例如，公有网络的主机名可能如下所示（没有 *https://）*： ``` 123456a0-0101-2020-9e11-1b159eec9000.c2.us-east-1.airflow.amazonaws.com ``` 1. （可选）macOS 和 Linux 用户可能需要运行以下命令以确保脚本可执行。 ``` chmod +x get-web-token.sh ``` 1. 运行以下脚本可获取 Web 登录令牌。 ``` ./get-web-token.sh ``` 您的命令提示符将显示： ``` https://123456a0-0101-2020-9e11-1b159eec9000.c2.us-east-1.airflow.amazonaws.com/aws_mwaa/aws-console-sso?login=true#{your-web-login-token} ``` ## 使用 Python 脚本 以下示例使用 Python 脚本中的 [boto3 create\$1web\$1login\$1token](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/mwaa.html#MWAA.Client.create_web_login_token) 方法来创建 Apache Airflow Web 登录令牌。您可以在 Amazon MWAA 之外运行此脚本。您只需安装 boto3 库。您可能需要创建一个虚拟环境来安装该库。该环境假设您已经为账户[配置了 AWS 身份验证凭证](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html#configuration)。 1. 复制以下代码示例的内容，并在本地另存为 `create-web-login-token.py`。 ``` import boto3 mwaa = boto3.client('mwaa') response = mwaa.create_web_login_token( Name="YOUR_ENVIRONMENT_NAME" ) webServerHostName = response["WebServerHostname"] webToken = response["WebToken"] airflowUIUrl = 'https://{0}/aws_mwaa/aws-console-sso?login=true#{1}'.format(webServerHostName, webToken) print("Here is your Airflow UI URL: ") print(airflowUIUrl) ``` 1. 用*红色*占位符代替 `YOUR_ENVIRONMENT_NAME`。 1. 运行以下脚本可获取 Web 登录令牌。 ``` python3 create-web-login-token.py ``` ## 接下来做什么？ + 在 [CreateWebLoginToken](https://docs.aws.amazon.com/mwaa/latest/API/API_CreateWebLoginToken.html) 上浏览用于创建 Web 登录令牌的 Amazon MWAA API 操作。 # 为 Apache Airflow Web 服务器设置自定义域 使用 Amazon Managed Workflows for Apache Airflow (Amazon MWAA) 时，您可以为托管式 Apache Airflow Web 服务器设置自定义域。通过使用自定义域，您可以使用 Apache Airflow 用户界面、Apache Airflow CLI 或 Apache Airflow Web 服务器访问环境的 Amazon MWAA 托管式 Apache Airflow Web 服务器。 **注意** 您只能在无互联网访问权限的私有 Web 服务器上使用自定义域。 **Amazon MWAA 上的自定义域应用场景** 1. 在 AWS 上的云应用程序中共享 Web 服务器域：通过使用自定义域，您可以定义用于访问 Web 服务器的用户友好型 URL，而不是生成的服务域名。您可以存储此自定义域，并将其作为应用程序中的环境变量共享。 1. 访问私有 Web 服务器：如果您想为无互联网访问权限的 VPC 中的 Web 服务器配置访问权限，使用自定义域可以简化 URL 重定向工作流。 **Topics** + [ ## 配置自定义域 ](#create-environment-with-custom-domain) + [ ## 设置联网基础设施 ](#set-up-networking-for-custom-domain) ## 配置自定义域 要配置自定义域功能，您需要在创建或更新 Amazon MWAA 环境时通过 `webserver.base_url` Apache Airflow 配置来提供自定义域值。以下约束适用于自定义域名： + 该值应是不含任何协议或路径的完全限定域名 (FQDN)。例如 `your-custom-domain.com`。 + Amazon MWAA 不允许在 URL 中添加路径。例如，`your-custom-domain.com/dags/` 不是有效的自定义域名。 + URL 长度最多为 255 个 ASCII 字符。 + 如果您提供空字符串，则默认情况下将使用 Amazon MWAA 生成的 Web 服务器 URL 创建环境。 根据以下示例使用 AWS CLI 创建带有自定义 Web 服务器域名的环境。 ``` aws mwaa create-environment \ --name my-mwaa-env \ --source-bucket-arn arn:aws:s3:::amzn-s3-demo-bucket \ --airflow-configuration-options '{"webserver.base_url":"my-custom-domain.com"}' \ --network-configuration '{"SubnetIds":["subnet-0123456789abcdef","subnet-fedcba9876543210"]}' \ --execution-role-arn arn:aws:iam::123456789012:role/my-execution-role ``` 创建或更新环境后，您需要在 AWS 账户 中设置联网基础设施，以便通过自定义域访问私有 Web 服务器。 要恢复使用默认由服务生成的 URL，请更新您的私有环境并移除 `webserver.base_url` 配置选项。 ## 设置联网基础设施 执行以下步骤设置所需的联网基础设施，以便与 AWS 账户 中的自定义域一起使用。 1. 获取 Amazon VPC 端点网络接口（ENI）的 IP 地址。要执行此操作，请首先使用 [https://awscli.amazonaws.com/v2/documentation/api/2.9.6/reference/mwaa/get-environment.html](https://awscli.amazonaws.com/v2/documentation/api/2.9.6/reference/mwaa/get-environment.html) 查找适合环境的 `WebserverVpcEndpointService`。 ``` aws mwaa get-environment --name your-environment-name ``` 如果成功，您将收到与以下内容类似的输出。 ``` { "Environment": { "AirflowConfigurationOptions": {}, "AirflowVersion": "latest-version", "Arn": "environment-arn", "CreatedAt": "2024-06-01T01:00:00-00:00", "DagS3Path": "dags", . . . "WebserverVpcEndpointService": "web-server-vpc-endpoint-service", "WeeklyMaintenanceWindowStart": "TUE:21:30" } } ``` 记下 `WebserverVpcEndpointService` 值并将其用于以下 Amazon EC2 `describe-vpc-endpoints` 命令中的 `web-server-vpc-endpoint-service`。以下命令中的 `--filters Name=service-name,Values=web-server-vpc-endpoint-service-id`。 1. 检索 Amazon VPC 端点详细信息。此命令用于获取与特定服务名称匹配的 Amazon VPC 端点详细信息，并以文本格式返回端点 ID 和关联的网络接口 ID。 ``` aws ec2 describe-vpc-endpoints \ --filters Name=service-name,Values=web-server-vpc-endpoint-service \ --query 'VpcEndpoints[*].{EndpointId:VpcEndpointId,NetworkInterfaceIds:NetworkInterfaceIds}' \ --output text ``` 1. 获取网络接口详细信息。此命令用于检索与上一步中所确定 Amazon VPC 端点关联的每个网络接口的私有 IP 地址。 ``` for eni_id in $( aws ec2 describe-vpc-endpoints \ --filters Name=service-name,Values=service-id \ --query 'VpcEndpoints[*].NetworkInterfaceIds' \ --output text ); do aws ec2 describe-network-interfaces \ --network-interface-ids $eni_id \ --query 'NetworkInterfaces[*].PrivateIpAddresses[*].PrivateIpAddress' \ --output text done ``` 1. 使用 `create-target-group` 创建新的目标组。您将使用此目标组来注册 Web 服务器 Amazon VPC 端点的 IP 地址。 ``` aws elbv2 create-target-group \ --name new-target-group-namne \ --protocol HTTPS \ --port 443 \ --vpc-id web-server-vpc-id \ --target-type ip \ --health-check-protocol HTTPS \ --health-check-port 443 \ --health-check-path / \ --health-check-enabled \ --matcher 'HttpCode="200,302"' ``` 使用 `register-targets` 命令注册 IP 地址。 ``` aws elbv2 register-targets \ --target-group-arn target-group-arn \ --targets Id=ip-address-1 Id=ip-address-2 ``` 1. 请求 ACM 证书。如果您使用的是现有证书，则跳过此步骤。 ``` aws acm request-certificate \ --domain-name my-custom-domain.com \ --validation-method DNS ``` 1. 配置 Application Load Balancer。首先创建负载均衡器，然后为该负载均衡器创建侦听器。指定在上一步中创建的 ACM 证书。 ``` aws elbv2 create-load-balancer \ --name my-mwaa-lb \ --type application \ --subnets subnet-id-1 subnet-id-2 ``` ``` aws elbv2 create-listener \ --load-balancer-arn load-balancer-arn \ --protocol HTTPS \ --port 443 \ --ssl-policy ELBSecurityPolicy-2016-08 \ --certificates CertificateArn=acm-certificate-arn \ --default-actions Type=forward,TargetGroupArn=target-group-arn ``` 如果您在私有子网中使用网络负载均衡器，请设置[堡垒主机](tutorials-private-network-bastion.md)或 [Site-to-Site VPN 隧道](tutorials-private-network-vpn-client.md)来访问 Web 服务器。 1. 使用 Route 53 为该域创建托管区。 ``` aws route53 create-hosted-zone --name my-custom-domain.com \ --caller-reference 1 ``` 为该域创建一条 A 记录。要使用 AWS CLI 执行此操作，请使用 `list-hosted-zones-by-name` 获取托管区 ID，然后使用 `change-resource-record-sets` 应用记录。 ``` HOSTED_ZONE_ID=$(aws route53 list-hosted-zones-by-name \ --dns-name my-custom-domain.com \ --query 'HostedZones[0].Id' --output text) ``` ``` aws route53 change-resource-record-sets \ --hosted-zone-id $HOSTED_ZONE_ID \ --change-batch '{ "Changes": [ { "Action": "CREATE", "ResourceRecordSet": { "Name": "my-custom-domain.com", "Type": "A", "AliasTarget": { "HostedZoneId": "load-balancer-hosted-zone-id>", "DNSName": "load-balancer-dns-name", "EvaluateTargetHealth": true } } } ] }' ``` 1. 更新 Web 服务器 Amazon VPC 端点的安全组规则，使其遵循最低权限原则，仅允许来自应用程序负载均衡器所在公有子网的 HTTPS 流量。将以下 JSON 保存到本地。例如，`sg-ingress-ip-permissions.json`。 ``` [ { "IpProtocol": "tcp", "FromPort": 443, "ToPort": 443, "UserIdGroupPairs": [ { "GroupId": "load-balancer-security-group-id" } ], "IpRanges": [ { "CidrIp": "public-subnet-1-cidr" }, { "CidrIp": "public-subnet-2-cidr" } ] } ] ``` 运行以下 Amazon EC2 命令更新入口安全组规则。指定 `--ip-permissions` 的 JSON 文件。 ``` aws ec2 authorize-security-group-ingress \ --group-id \ --ip-permissions file://sg-ingress-ip-permissions.json ``` 运行以下 Amazon EC2 命令更新出口规则。 ``` aws ec2 authorize-security-group-egress \ --group-id webserver-vpc-endpoint-security-group-id \ --protocol tcp \ --port 443 \ --source-group load-balancer-security-group-id ``` 打开 Amazon MWAA 控制台并导航到 Apache Airflow UI。如果是在私有子网中设置网络负载均衡器，而不是此处使用的应用程序负载均衡器，则必须使用以下选项之一访问 Web 服务器。 + [教程：使用 Linux 堡垒主机配置私有网络访问权限](tutorials-private-network-bastion.md) + [教程：使用 AWS Client VPN 配置私有网络访问权限](tutorials-private-network-vpn-client.md) # 创建 Apache Airflow CLI 令牌 **提示** REST API 比 CLI 更现代，专为与外部系统的编程集成而设计。REST 是与 Apache Airflow 交互的首选方式。 您可以使用本页上的命令生成 CLI 令牌，然后直接在命令 shell 中调用 Amazon MWAA API。例如，您可以获取令牌，然后使用 Amazon MWAA API 以编程方式部署 DAG。下一节包括使用 AWS CLI、curl 脚本、Python 脚本或 bash 脚本创建 Apache Airflow CLI 令牌的步骤。响应中返回的令牌在 60 秒内有效。 AWS CLI 令牌旨在替代同步 shell 操作，而不是异步 API 命令。因此，可用的并发是有限的。为确保 Web 服务器对用户保持响应能力，建议在前一个请求成功完成之前不要打开新的 AWS CLI 请求。 **Contents** + [ ## 先决条件 ](#call-mwaa-apis-cli-prereqs) + [ ### 访问 ](#access-airflow-ui-prereqs-access) + [ ### AWS CLI ](#access-airflow-ui-prereqs-cli) + [ ## 使用 AWS CLI ](#create-cli-token-cli) + [ ## 使用 curl 脚本 ](#create-cli-token-curl) + [ ## 使用 bash 脚本 ](#create-cli-token-bash) + [ ## 使用 Python 脚本 ](#create-cli-token-python) + [ ## 接下来做什么？ ](#mwaa-cli-next-up) ## 先决条件 下一节介绍了使用本页上的命令和脚本所需的初步步骤。 ### 访问 + 在 AWS Identity and Access Management (IAM) 中访问 [Apache Airflow 用户界面访问策略：亚马逊 MWAAWeb ServerAccess](access-policies.md#web-ui-access) 中的 Amazon MWAA 权限策略的 AWS 账户。 + 在 AWS Identity and Access Management (IAM) 中访问 Amazon MWAA 权限策略 [完整的 API 和控制台访问政策：Amazon MWAAFull ApiAccess](access-policies.md#full-access-policy) 的 AWS 账户。 ### AWS CLI AWS Command Line Interface (AWS CLI) 是一种开源工具，您可以用来在命令行 Shell 中使用命令与 AWS 服务进行交互。要完成本节中的步骤，您需要以下满足以下条件： + [AWS CLI – 安装版本 2](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html)。 + [AWS CLI – 使用 `aws configure` 进行快速配置](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html)。 ## 使用 AWS CLI 以下示例使用 AWS CLI 中的 [create-cli-token](https://docs.aws.amazon.com/cli/latest/reference/mwaa/create-cli-token.html) 命令创建 Apache Airflow CLI 令牌。 ``` aws mwaa create-cli-token --name YOUR_ENVIRONMENT_NAME ``` ## 使用 curl 脚本 以下示例使用 curl 脚本调用 AWS CLI 中的 [create-web-login-token](https://docs.aws.amazon.com/cli/latest/reference/mwaa/create-cli-token.html) 命令，通过 Apache Airflow Web 服务器上的端点调用 Apache Airflow CLI。 ------ #### [ Apache Airflow v3 ] 1. 从文本文件中复制 curl 语句并将其粘贴到命令 shell 中。 **注意** 将其复制到剪贴板后，可能需要使用 shell 菜单中的**编辑 > 粘贴**。 ``` CLI_JSON=$(aws mwaa --region us-east-1 create-cli-token --name YOUR_ENVIRONMENT_NAME) \ && CLI_TOKEN=$(echo $CLI_JSON | jq -r '.CliToken') \ && WEB_SERVER_HOSTNAME=$(echo $CLI_JSON | jq -r '.WebServerHostname') \ && CLI_RESULTS=$(curl -L --request POST "https://$WEB_SERVER_HOSTNAME/aws_mwaa/cli" \ --header "Authorization: Bearer $CLI_TOKEN" \ --header "Content-Type: text/plain" \ --data-raw "dags trigger YOUR_DAG_NAME --logical-date $(date -u +"%Y-%m-%dT%H:%M:%SZ")") \ && echo "Output:" \ && echo $CLI_RESULTS | jq -r '.stdout' | base64 --decode \ && echo "Errors:" \ && echo $CLI_RESULTS | jq -r '.stderr' | base64 --decode ``` 1. 用环境的 AWS 区域、`YOUR_DAG_NAME` 和 `YOUR_ENVIRONMENT_NAME` 替换 *red* 中的占位符。例如，公有网络的主机名可能如下所示（没有 *https://）*： ``` 123456a0-0101-2020-9e11-1b159eec9000.c2.us-east-1.airflow.amazonaws.com ``` 您的命令提示符将显示： ``` { "stderr":"", "stdout":"" } ``` ------ #### [ Apache Airflow v2 ] 1. 从文本文件中复制 curl 语句并将其粘贴到命令 shell 中。 **注意** 将其复制到剪贴板后，可能需要使用 shell 菜单中的**编辑 > 粘贴**。 ``` CLI_JSON=$(aws mwaa --region us-east-1 create-cli-token --name YOUR_ENVIRONMENT_NAME) \ && CLI_TOKEN=$(echo $CLI_JSON | jq -r '.CliToken') \ && WEB_SERVER_HOSTNAME=$(echo $CLI_JSON | jq -r '.WebServerHostname') \ && CLI_RESULTS=$(curl --request POST "https://$WEB_SERVER_HOSTNAME/aws_mwaa/cli" \ --header "Authorization: Bearer $CLI_TOKEN" \ --header "Content-Type: text/plain" \ --data-raw "dags trigger YOUR_DAG_NAME") \ && echo "Output:" \ && echo $CLI_RESULTS | jq -r '.stdout' | base64 --decode \ && echo "Errors:" \ && echo $CLI_RESULTS | jq -r '.stderr' | base64 --decode ``` 1. 用环境的 AWS 区域、`YOUR_DAG_NAME` 和 `YOUR_ENVIRONMENT_NAME` 替换 *red* 中的占位符。例如，公有网络的主机名可能如下所示（没有 *https://）*： ``` 123456a0-0101-2020-9e11-1b159eec9000.c2.us-east-1.airflow.amazonaws.com ``` 您的命令提示符将显示： ``` { "stderr":"", "stdout":"" } ``` ------ ## 使用 bash 脚本 以下示例使用 bash 脚本来调用 AWS CLI 中的 [create-cli-token](https://docs.aws.amazon.com/cli/latest/reference/mwaa/create-cli-token.html) 命令创建 Apache Airflow CLI 令牌。 ------ #### [ Apache Airflow v3 ] 1. 复制以下代码示例的内容，并在本地另存为 `get-cli-token.sh`。 ``` # brew install jq aws mwaa create-cli-token --name YOUR_ENVIRONMENT_NAME | export CLI_TOKEN=$(jq -r .CliToken) && curl -L --request POST "https://YOUR_HOST_NAME/aws_mwaa/cli" \ --header "Authorization: Bearer $CLI_TOKEN" \ --header "Content-Type: text/plain" \ --data-raw "dags trigger YOUR_DAG_NAME --logical-date $(date -u +"%Y-%m-%dT%H:%M:%SZ")" ``` 1. 用*红色*占位符替换 `YOUR_ENVIRONMENT_NAME`、`YOUR_HOST_NAME` 和 `YOUR_DAG_NAME`。例如，公有网络的主机名可能如下所示（没有 *https://）*： ``` 123456a0-0101-2020-9e11-1b159eec9000.c2.us-east-1.airflow.amazonaws.com ``` 1. （可选）macOS 和 Linux 用户可能需要运行以下命令以确保脚本可执行。 ``` chmod +x get-cli-token.sh ``` 1. 运行以下脚本可创建 Apache Airflow CLI 令牌。 ``` ./get-cli-token.sh ``` ------ #### [ Apache Airflow v2 ] 1. 复制以下代码示例的内容，并在本地另存为 `get-cli-token.sh`。 ``` # brew install jq aws mwaa create-cli-token --name YOUR_ENVIRONMENT_NAME | export CLI_TOKEN=$(jq -r .CliToken) && curl --request POST "https://YOUR_HOST_NAME/aws_mwaa/cli" \ --header "Authorization: Bearer $CLI_TOKEN" \ --header "Content-Type: text/plain" \ --data-raw "dags trigger YOUR_DAG_NAME" ``` 1. 用*红色*占位符替换 `YOUR_ENVIRONMENT_NAME`、`YOUR_HOST_NAME` 和 `YOUR_DAG_NAME`。例如，公有网络的主机名可能如下所示（没有 *https://）*： ``` 123456a0-0101-2020-9e11-1b159eec9000.c2.us-east-1.airflow.amazonaws.com ``` 1. （可选）macOS 和 Linux 用户可以运行以下命令以确保脚本可执行。 ``` chmod +x get-cli-token.sh ``` 1. 运行以下脚本可创建 Apache Airflow CLI 令牌。 ``` ./get-cli-token.sh ``` ------ ## 使用 Python 脚本 以下示例使用 Python 脚本中的 [boto3 create\$1cli\$1token](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/mwaa.html#MWAA.Client.create_cli_token) 方法来创建 Apache Airflow CLI 令牌，并触发 DAG。您可以在 Amazon MWAA 之外运行此脚本。您只需安装 boto3 库。您可能需要创建一个虚拟环境来安装该库。该环境假设您已经为账户[配置了 AWS 身份验证凭证](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html#configuration)。 ------ #### [ Apache Airflow v3 ] 1. 复制以下代码示例的内容，并在本地另存为 `create-cli-token.py`。 ``` """ Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import boto3 import json import requests import base64 mwaa_env_name = 'YOUR_ENVIRONMENT_NAME' dag_name = 'YOUR_DAG_NAME' mwaa_cli_command = 'dags trigger' client = boto3.client('mwaa') mwaa_cli_token = client.create_cli_token( Name=mwaa_env_name ) mwaa_auth_token = 'Bearer ' + mwaa_cli_token['CliToken'] mwaa_webserver_hostname = 'https://{0}/aws_mwaa/cli'.format(mwaa_cli_token['WebServerHostname']) raw_data = '{0} {1}'.format(mwaa_cli_command, dag_name) mwaa_response = requests.post( mwaa_webserver_hostname, headers={ 'Authorization': mwaa_auth_token, 'Content-Type': 'text/plain' }, data=raw_data ) mwaa_std_err_message = base64.b64decode(mwaa_response.json()['stderr']).decode('utf8') mwaa_std_out_message = base64.b64decode(mwaa_response.json()['stdout']).decode('utf8') print(mwaa_response.status_code) print(mwaa_std_err_message) print(mwaa_std_out_message) ``` 1. 用占位符替换 `YOUR_ENVIRONMENT_NAME` 和 `YOUR_DAG_NAME`。 1. 运行以下脚本可创建 Apache Airflow CLI 令牌。 ``` python3 create-cli-token.py ``` ------ #### [ Apache Airflow v2 ] 1. 复制以下代码示例的内容，并在本地另存为 `create-cli-token.py`。 ``` """ Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import boto3 import json import requests import base64 mwaa_env_name = 'YOUR_ENVIRONMENT_NAME' dag_name = 'YOUR_DAG_NAME' mwaa_cli_command = 'dags trigger' client = boto3.client('mwaa') mwaa_cli_token = client.create_cli_token( Name=mwaa_env_name ) mwaa_auth_token = 'Bearer ' + mwaa_cli_token['CliToken'] mwaa_webserver_hostname = 'https://{0}/aws_mwaa/cli'.format(mwaa_cli_token['WebServerHostname']) raw_data = '{0} {1}'.format(mwaa_cli_command, dag_name) mwaa_response = requests.post( mwaa_webserver_hostname, headers={ 'Authorization': mwaa_auth_token, 'Content-Type': 'text/plain' }, data=raw_data ) mwaa_std_err_message = base64.b64decode(mwaa_response.json()['stderr']).decode('utf8') mwaa_std_out_message = base64.b64decode(mwaa_response.json()['stdout']).decode('utf8') print(mwaa_response.status_code) print(mwaa_std_err_message) print(mwaa_std_out_message) ``` 1. 用占位符替换 `YOUR_ENVIRONMENT_NAME` 和 `YOUR_DAG_NAME`。 1. 运行以下脚本可创建 Apache Airflow CLI 令牌。 ``` python3 create-cli-token.py ``` ------ ## 接下来做什么？ + 在 [createClitoken](https://docs.aws.amazon.com/mwaa/latest/API/API_CreateCliToken.html) 上浏览用于创建 CLI 令牌的 Amazon MWAA API 操作。 # 使用 Apache Airflow REST API 对于运行 Apache Airflow v2.4.3 及更高版本的环境，Amazon Managed Workflows for Apache Airflow (Amazon MWAA) 支持使用 Apache Airflow REST API 直接与您的 Apache Airflow 环境进行交互。这使您可以通过编程方式访问和管理您的 Amazon MWAA 环境，从而为调用数据编排工作流程、管理和监控各种 Apache Airflow 组件（例如元数据数据库 DAGs、触发器和调度程序）的状态提供了一种标准化的方式。 为了在使用 Apache Airflow REST API 时支持可扩展性，Amazon MWAA 提供了水平扩缩 Web 服务器容量的选项，以满足增加的需求，无论是来自 REST API 请求、命令行界面 (CLI) 的使用还是并发 Apache Airflow 用户界面 (UI) 用户数量的增加。有关 Amazon MWAA 如何扩展 Web 服务器的更多信息，请参阅 [配置 Amazon MWAA Web 服务器自动扩缩](mwaa-web-server-autoscaling.md)。 您可以使用 Apache Airflow REST API 实现环境的以下使用案例： + **编程访问** – 您现在可以在不依赖 Apache Airflow 用户界面或 CLI 的情况下，启动 Apache Airflow DAG 运行、管理数据集以及检索元数据数据库、触发器和调度器等各种组件的状态。 + **与外部应用程序和微服务集成** – 由于支持 REST API，您可以构建自定义解决方案以将您的 Amazon MWAA 环境与其他系统集成。例如，您可以启动工作流以响应来自外部系统的事件，例如已完成的数据库作业或新用户注册等。 + **集中监控** — 您可以构建监控控制面板，汇总多 DAGs 个 Amazon MWAA 环境中的状态，从而实现集中监控和管理。 有关 Apache Airflow REST API 的更多信息，请参阅 [Apache Airflow REST API 参考](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html)。 通过使用`InvokeRestApi`，您可以使用凭据访问 Apache Airflow REST API。 AWS 您也可以通过获取 Web 服务器访问令牌，然后使用该令牌进行调用的方式来访问。 如果您在使用 `InvokeRestApi` 操作时遇到消息 `Update your environment to use InvokeRestApi` 错误，则表示您需要更新 Amazon MWAA 环境。当您的 Amazon MWAA 环境与 `InvokeRestApi` 功能相关的最新更改不兼容时，就会发生此错误。要解决此问题，请更新您的 Amazon MWAA 环境以纳入 `InvokeRestApi` 功能的必要更改。 `InvokeRestApi` 操作的默认超时时间为 10 秒。如果操作未在这 10 秒的时间范围内完成，则会自动终止并引发错误。确保您的 REST API 调用设计为在此超时时间内完成，以避免出现错误。 为了在使用 Apache Airflow REST API 时支持可扩展性，Amazon MWAA 提供了水平扩缩 Web 服务器容量的选项，以满足增加的需求，无论是来自 REST API 请求、命令行界面 (CLI) 的使用还是并发 Apache Airflow 用户界面 (UI) 用户数量的增加。有关 Amazon MWAA 如何扩展 Web 服务器的更多信息，请参阅 [配置 Amazon MWAA Web 服务器自动扩缩](mwaa-web-server-autoscaling.md)。 您可以使用 Apache Airflow REST API 实现环境的以下使用案例： + **编程访问** – 您现在可以在不依赖 Apache Airflow 用户界面或 CLI 的情况下，启动 Apache Airflow DAG 运行、管理数据集以及检索元数据数据库、触发器和调度器等各种组件的状态。 + **与外部应用程序和微服务集成** – 由于支持 REST API，您可以构建自定义解决方案以将您的 Amazon MWAA 环境与其他系统集成。例如，您可以启动工作流以响应来自外部系统的事件，例如已完成的数据库作业或新用户注册等。 + **集中监控** — 您可以构建监控控制面板，汇总多 DAGs 个 Amazon MWAA 环境中的状态，从而实现集中监控和管理。 有关 Apache Airflow REST API 的更多信息，请参阅 [Apache Airflow REST API 参考](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html)。 通过使用`InvokeRestApi`，您可以使用凭据访问 Apache Airflow REST API。 AWS 您也可以通过获取 Web 服务器访问令牌，然后使用该令牌进行调用的方式来访问。 + 如果您在使用 `InvokeRestApi` 操作时遇到消息 `Update your environment to use InvokeRestApi` 错误，则表示您需要更新 Amazon MWAA 环境。当您的 Amazon MWAA 环境与 `InvokeRestApi` 功能相关的最新更改不兼容时，就会发生此错误。要解决此问题，请更新您的 Amazon MWAA 环境以纳入 `InvokeRestApi` 功能的必要更改。 + `InvokeRestApi` 操作的默认超时时间为 10 秒。如果操作未在这 10 秒的时间范围内完成，则会自动终止并引发错误。确保您的 REST API 调用设计为在此超时时间内完成，以避免出现错误。 **重要** 响应有效载荷大小不能超过 6 MB。如果超出限制，`RestApi` 将失败。 根据以下示例对 Apache Airflow REST API 进行 API 调用并启动新的 DAG 运行： **Topics** + [ ## 授予对 Apache Airflow REST API 的访问权限：`airflow:InvokeRestApi` ](#granting-access-MWAA-Enhanced-REST-API) + [ ## 调用 Apache Airflow REST API ](#listing-DAGs-creating-variables-using-restapi-script) + [ ## 创建 Web 服务器会话令牌并调用 Apache Airflow REST API ](#create-web-server-session-token) ## 授予对 Apache Airflow REST API 的访问权限：`airflow:InvokeRestApi` 要 AWS 使用证书访问 Apache Airflow REST API，您必须在 IAM 策略`airflow:InvokeRestApi`中授予权限。在以下策略示例中，在 `{airflow-role}` 中指定 `Admin`、`Op`、`User`、`Viewer` 或 `Public` 角色以自定义用户访问权限级别。有关更多信息，请参阅*《Apache Airflow 参考指南》*中的[默认角色](https://airflow.apache.org/docs/apache-airflow/1.10.6/security.html?highlight=ldap#default-roles)。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "AllowMwaaRestApiAccess", "Effect": "Allow", "Action": "airflow:InvokeRestApi", "Resource": [ "arn:aws:airflow:us-east-1:111122223333:role/{your-environment-name}/{airflow-role}" ] } ] } ``` ------ **注意** 配置私有 Web 服务器时，无法从虚拟私有云 (VPC) 之外调用 `InvokeRestApi` 操作。您可以使用 `aws:SourceVpc` 键对此操作执行更精细的访问控制。有关更多信息，请参阅 a [ws: SourceVpc](https://docs.aws.amazon.com//IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-sourcevpc)。 ## 调用 Apache Airflow REST API 以下示例脚本介绍了如何使用 Apache Airflow REST API 列出环境中的 DAGs 可用变量以及如何创建 Apache Airflow 变量： ``` import boto3 env_name = "MyAirflowEnvironment" def list_dags(client): request_params = { "Name": env_name, "Path": "/dags", "Method": "GET", "QueryParameters": { "paused": False } } response = client.invoke_rest_api( **request_params ) print("Airflow REST API response: ", response['RestApiResponse']) def create_variable(client): request_params = { "Name": env_name, "Path": "/variables", "Method": "POST", "Body": { "key": "test-restapi-key", "value": "test-restapi-value", "description": "Test variable created by MWAA InvokeRestApi API", } } response = client.invoke_rest_api( **request_params ) print("Airflow REST API response: ", response['RestApiResponse']) if __name__ == "__main__": client = boto3.client("mwaa") list_dags(client) create_variable(client) ``` ## 创建 Web 服务器会话令牌并调用 Apache Airflow REST API 要创建 Web 服务器访问令牌，请使用以下 Python 函数。该函数会首先调用 Amazon MWAA API 来获取 Web 登录令牌。Web 登录令牌将在 60 秒后过期，然后会交换为一个 Web *会话*令牌，后者可让您访问 Web 服务器和使用 Apache Airflow REST API。如果您需要每秒 10 个事务（TPS）以上的节流容量，则可以使用此方法访问 Apache Airflow REST API。 会话令牌将在 12 小时后过期。 **提示** 以下代码示例中从 Apache Airflow v2 到 v3 的主要变化是： REST API 路径从 `/api/v1` 更改为 `/api/v2` 登录路径从 `/aws_maa/login` 更改为 `/pluginsv2/aws_mwaa/login` 登录 `response.cookies["_token"]` 的响应包含令牌信息，您必须在后续的 API 调用中使用这些信息 对于 REST API 调用，您必须在标头中传递以下 `jwt_token` 信息： ``` headers = { "Authorization": f"Bearer {jwt_token}", "Content-Type": "application/json" } ``` ------ #### [ Apache Airflow v3 ] ``` def get_token_info(region, env_name): logging.basicConfig(level=logging.INFO) try: # Initialize MWAA client and request a web login token mwaa = boto3.client('mwaa', region_name=region) response = mwaa.create_web_login_token(Name=env_name) # Extract the web server hostname and login token web_server_host_name = response["WebServerHostname"] web_token = response["WebToken"] # Construct the URL needed for authentication login_url = f"https://{web_server_host_name}/pluginsv2/aws_mwaa/login" login_payload = {"token": web_token} # Make a POST request to the MWAA login url using the login payload response = requests.post( login_url, data=login_payload, timeout=10 ) # Check if login was successful if response.status_code == 200: # Return the hostname and the session cookie return ( web_server_host_name, response.cookies['_token'] ) else: # Log an error logging.error("Failed to log in: HTTP %d", response.status_code) return None except requests.RequestException as e: # Log any exceptions raised during the request to the MWAA login endpoint logging.error("Request failed: %s", str(e)) return None except Exception as e: # Log any other unexpected exceptions logging.error("An unexpected error occurred: %s", str(e)) return None ``` ------ #### [ Apache Airflow v2 ] ``` def get_session_info(region, env_name): logging.basicConfig(level=logging.INFO) try: # Initialize MWAA client and request a web login token mwaa = boto3.client('mwaa', region_name=region) response = mwaa.create_web_login_token(Name=env_name) # Extract the web server hostname and login token web_server_host_name = response["WebServerHostname"] web_token = response["WebToken"] # Construct the URL needed for authentication login_url = f"https://{web_server_host_name}/aws_mwaa/login" login_payload = {"token": web_token} # Make a POST request to the MWAA login url using the login payload response = requests.post( login_url, data=login_payload, timeout=10 ) # Check if login was succesfull if response.status_code == 200: # Return the hostname and the session cookie return ( web_server_host_name, response.cookies["session"] ) else: # Log an error logging.error("Failed to log in: HTTP %d", response.status_code) return None except requests.RequestException as e: # Log any exceptions raised during the request to the MWAA login endpoint logging.error("Request failed: %s", str(e)) return None except Exception as e: # Log any other unexpected exceptions logging.error("An unexpected error occurred: %s", str(e)) return None ``` ------ 完成身份验证后，您将会获得开始向 API 端点发送请求的凭证。在下一部分的示例中，使用端点 `dags/{dag_name}/dagRuns`。 ------ #### [ Apache Airflow v3 ] ``` def trigger_dag(region, env_name, dag_id): """ Triggers a DAG in a specified MWAA environment using the Airflow REST API. Args: region (str): AWS region where the MWAA environment is hosted. env_name (str): Name of the MWAA environment. dag_id (str): ID of the DAG to trigger. """ logging.info(f"Attempting to trigger DAG {dag_id} in environment {env_name} at region {region}") # Retrieve the web server hostname and token for authentication try: web_server_host_name, jwt_token = get_token_info(region, env_name) if not jwt_token: logging.error("Authentication failed, no jwt token retrieved.") return except Exception as e: logging.error(f"Error retrieving token info: {str(e)}") return # Prepare headers and payload for the request request_headers = { "Authorization": f"Bearer {jwt_token}", "Content-Type": "application/json" # Good practice to include, even for GET } # sample request body input json_body = {"logical_date": "2025-09-17T14:15:00Z"} # Construct the URL for triggering the DAG url = f"https://{web_server_host_name}/api/v2/dags/{dag_id}/dagRuns" # Send the POST request to trigger the DAG try: response = requests.post(url, headers=request_headers, json=json_body) # Check the response status code to determine if the DAG was triggered successfully if response.status_code == 200: logging.info("DAG triggered successfully.") else: logging.error(f"Failed to trigger DAG: HTTP {response.status_code} - {response.text}") except requests.RequestException as e: logging.error(f"Request to trigger DAG failed: {str(e)}") if __name__ == "__main__": logging.basicConfig(level=logging.INFO) # Check if the correct number of arguments is provided if len(sys.argv) != 4: logging.error("Incorrect usage. Proper format: python script_name.py {region} {env_name} {dag_id}") sys.exit(1) region = sys.argv[1] env_name = sys.argv[2] dag_id = sys.argv[3] # Trigger the DAG with the provided arguments trigger_dag(region, env_name, dag_id) ``` ------ #### [ Apache Airflow v2 ] ``` def trigger_dag(region, env_name, dag_name): """ Triggers a DAG in a specified MWAA environment using the Airflow REST API. Args: region (str): AWS region where the MWAA environment is hosted. env_name (str): Name of the MWAA environment. dag_name (str): Name of the DAG to trigger. """ logging.info(f"Attempting to trigger DAG {dag_name} in environment {env_name} at region {region}") # Retrieve the web server hostname and session cookie for authentication try: web_server_host_name, session_cookie = get_session_info(region, env_name) if not session_cookie: logging.error("Authentication failed, no session cookie retrieved.") return except Exception as e: logging.error(f"Error retrieving session info: {str(e)}") return # Prepare headers and payload for the request cookies = {"session": session_cookie} json_body = {"conf": {}} # Construct the URL for triggering the DAG url = f"https://{web_server_host_name}/api/v1/dags/{dag_id}/dagRuns" # Send the POST request to trigger the DAG try: response = requests.post(url, cookies=cookies, json=json_body) # Check the response status code to determine if the DAG was triggered successfully if response.status_code == 200: logging.info("DAG triggered successfully.") else: logging.error(f"Failed to trigger DAG: HTTP {response.status_code} - {response.text}") except requests.RequestException as e: logging.error(f"Request to trigger DAG failed: {str(e)}") if __name__ == "__main__": logging.basicConfig(level=logging.INFO) # Check if the correct number of arguments is provided if len(sys.argv) != 4: logging.error("Incorrect usage. Proper format: python script_name.py {region} {env_name} {dag_name}") sys.exit(1) region = sys.argv[1] env_name = sys.argv[2] dag_name = sys.argv[3] # Trigger the DAG with the provided arguments trigger_dag(region, env_name, dag_name) ``` ------ # Apache Airflow CLI 命令参考 本主题介绍了 Amazon MWAA 上支持和不支持的 Apache Airflow CLI 命令。 **提示** REST API 比 CLI 更现代，专为与外部系统的编程集成而设计。REST 是与 Apache Airflow 交互的首选方式。 **Contents** + [ ## 先决条件 ](#airflow-cli-command-prereqs) + [ ### 访问 ](#access-airflow-ui-prereqs-access) + [ ### AWS CLI ](#access-airflow-ui-prereqs-cli) + [ ## 更改了哪些内容？ ](#airflow-cli-command-changed) + [ ## 支持的 CLI 命令 ](#airflow-cli-commands) + [ ### 支持的命令 ](#airflow-cli-commands-supported) + [ ### 使用解析 DAG 的命令 ](#parsing-support) + [ ## 代码示例 ](#airflow-cli-command-examples) + [ ### 设置、获取或删除 Apache Airflow v2 变量 ](#example-airflow-cli-commands-bash) + [ ### 触发 DAG 时添加配置 ](#example-airflow-cli-commands-trigger) + [ ### 在通往堡垒主机的 SSH 隧道上运行 CLI 命令。 ](#example-airflow-cli-commands-private) ## 先决条件 下一节介绍了使用本页上的命令和脚本所需的初步步骤。 ### 访问 + 在 AWS Identity and Access Management (IAM) 中访问 [Apache Airflow 用户界面访问策略：亚马逊 MWAAWeb ServerAccess](access-policies.md#web-ui-access) 中的 Amazon MWAA 权限策略的 AWS 账户。 + 在 AWS Identity and Access Management (IAM) 中访问 Amazon MWAA 权限策略 [完整的 API 和控制台访问政策：Amazon MWAAFull ApiAccess](access-policies.md#full-access-policy) 的 AWS 账户。 ### AWS CLI AWS Command Line Interface (AWS CLI) 是一种开源工具，您可以用来在命令行 Shell 中使用命令与 AWS 服务进行交互。要完成本节中的步骤，您需要以下满足以下条件： + [AWS CLI – 安装版本 2](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html)。 + [AWS CLI – 使用 `aws configure` 进行快速配置](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html)。 ## 更改了哪些内容？ + **v3：Airflow 架构**。Apache Airflow v3 引入了突破性的架构更改，以提高安全性和可扩展性，并简化维护。要了解更多信息，请参阅[升级到 Airflow 3](https://airflow.apache.org/docs/apache-airflow/stable/installation/upgrading_to_airflow3.html)。 + **v2：Airflow CLI 命令结构**。Apache Airflow v2 CLI 的组织方式是将相关命令分组为子命令，这意味着如果您想升级到 Apache Airflow v2，则需要更新 Apache Airflow v1 脚本。例如，Apache Airflow v1 中的 `unpause` 已更新为 Apache Airflow v2 中的 `dags unpause`。要了解更多信息，请参阅 [2.0 中的 Airflow CLI 更改](http://airflow.apache.org/docs/apache-airflow/2.0.2/upgrading-to-2.html#airflow-cli-changes-in-2-0)。 ## 支持的 CLI 命令 下一节列出了 Amazon MWAA 上可用的 Apache Airflow CLI 命令。 ### 支持的命令 ------ #### [ Apache Airflow v3 ] | 次要版本 | 命令 | | --- | --- | | v3.0.6 | [资产详情](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#details) | | v3.0.6 | [资产列表](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#list) | | v3.0.6 | [资产实体化](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#materialize) | | v3.0.6 | [回填创建](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#create) | | v3.0.6 | [备忘单](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#cheat-sheet) | | v3.0.6 | [添加连接](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#add) | | v3.0.6 | [删除连接](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#delete) | | v3.0.6 | [删除 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#delete_repeat1) | | v3.0.6 | [dags 列表](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat2) | | v3.0.6 | [dags 列表-任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list-jobs) | | v3.0.6 | [dags 列表导入错误](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#list-import-errors) | | v3.0.6 | [dags 列表-运行次数](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list-runs) | | v3.0.6 | [dags 下次-执行](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#next-execution) | | v3.0.6 | [暂停 dag](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#pause) | | v3.0.6 | [报告 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#report) | | v3.0.6 | [重新序列化 dags](https://airflow.apache.org/docs/apache-airflow/2.4.3/cli-and-env-variables-ref.html#reserialize) | | v3.0.6 | [显示 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#show) | | v3.0.6 | [陈述 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#state) | | v3.0.6 | [测试 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#test) | | v3.0.6 | [触发 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#trigger) | | v3.0.6 | [取消暂停 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#unpause) | | v3.0.6 | [清理 db](https://airflow.apache.org/docs/apache-airflow/2.4.3/cli-and-env-variables-ref.html#clean) | | v3.0.6 | [提供商的行为](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#behaviours) | | v3.0.6 | [获得提供商](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#get_repeat2) | | v3.0.6 | [提供商挂钩](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#hooks) | | v3.0.6 | [提供商链接](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#links) | | v3.0.6 | [提供商列表](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat4) | | v3.0.6 | [提供程序通知](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#notifications) | | v3.0.6 | [提供商密钥](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#secrets) | | v3.0.6 | [提供商触发器](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#triggerer) | | v3.0.6 | [提供商小部件](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#widgets) | | v3.0.6 | [角色添加权限](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#add-perms) | | v3.0.6 | [角色删除权限](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#del-perms) | | v3.0.6 | [角色创建](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#create) | | v3.0.6 | [列出角色](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat5) | | v3.0.6 | [清除任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#clear) | | v3.0.6 | [任务失败-部署](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#failed-deps) | | v3.0.6 | [列出任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat6) | | v3.0.6 | [渲染任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#render) | | v3.0.6 | [陈述任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#state_repeat1) | | v3.0.6 | [dag 运行的任务状态](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#states-for-dag-run) | | v3.0.6 | [测试任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#test_repeat1) | | v3.0.6 | [删除变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#delete_repeat4) | | v3.0.6 | [获取变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#get_repeat3) | | v3.0.6 | [设置变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#set_repeat1) | | v3.0.6 | [列出变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat8) | | v3.0.6 | [version](http://airflow.apache.org/docs/apache-airflow/1.10.12/cli-ref.html#version) | ------ #### [ Apache Airflow v2 ] | 次要版本 | 命令 | | --- | --- | | v2.0\$1 | [备忘单](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#cheat-sheet) | | v2.0\$1 | [添加连接](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#add) | | v2.0\$1 | [删除连接](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#delete) | | v2.2\$1（[注意](#parsing-support)） | [回填 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#backfill) | | v2.0\$1 | [删除 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#delete_repeat1) | | v2.2\$1（[注意](#parsing-support)） | [dags 列表](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat2) | | v2.0\$1 | [dags 列表-任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list-jobs) | | v2.6\$1 | [dags 列表导入错误](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#list-import-errors) | | v2.2\$1（[注意](#parsing-support)） | [dags 列表-运行次数](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list-runs) | | v2.2\$1（[注意](#parsing-support)） | [dags 下次-执行](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#next-execution) | | v2.0\$1 | [暂停 dag](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#pause) | | v2.0\$1 | [报告 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#report) | | v2.4\$1 | [重新序列化 dags](https://airflow.apache.org/docs/apache-airflow/2.4.3/cli-and-env-variables-ref.html#reserialize) | | v2.0\$1 | [显示 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#show) | | v2.0\$1 | [陈述 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#state) | | v2.0\$1 | [测试 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#test) | | v2.0\$1 | [触发 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#trigger) | | v2.0\$1 | [取消暂停 dags](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#unpause) | | v2.4\$1 | [清理 db](https://airflow.apache.org/docs/apache-airflow/2.4.3/cli-and-env-variables-ref.html#clean) | | v2.0\$1 | [提供商的行为](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#behaviours) | | v2.0\$1 | [获得提供商](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#get_repeat2) | | v2.0\$1 | [提供商挂钩](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#hooks) | | v2.0\$1 | [提供商链接](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#links) | | v2.0\$1 | [提供商列表](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat4) | | v2.8\$1 | [提供程序通知](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#notifications) | | v2.6\$1 | [提供商密钥](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#secrets) | | v2.7\$1 | [提供商触发器](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#triggerer) | | v2.0\$1 | [提供商小部件](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#widgets) | | v2.6\$1 | [角色添加权限](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#add-perms) | | v2.6\$1 | [角色删除权限](https://airflow.apache.org/docs/apache-airflow/stable/cli-and-env-variables-ref.html#del-perms) | | v2.6\$1 | [角色创建](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#create) | | v2.0\$1 | [列出角色](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat5) | | v2.0\$1 | [清除任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#clear) | | v2.0\$1 | [任务失败-部署](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#failed-deps) | | v2.0\$1 | [列出任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat6) | | v2.0\$1 | [渲染任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#render) | | v2.0\$1 | [运行任务](https://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#run) | | v2.0\$1 | [陈述任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#state_repeat1) | | v2.0\$1 | [dag 运行的任务状态](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#states-for-dag-run) | | v2.0\$1 | [测试任务](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#test_repeat1) | | v2.0\$1 | [删除变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#delete_repeat4) | | v2.0\$1 | [获取变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#get_repeat3) | | v2.0\$1 | [设置变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#set_repeat1) | | v2.0\$1 | [列出变量](http://airflow.apache.org/docs/apache-airflow/2.2.2/cli-and-env-variables-ref.html#list_repeat8) | | v2.0\$1 | [版本](http://airflow.apache.org/docs/apache-airflow/1.10.12/cli-ref.html#version) | ------ ### 使用解析 DAG 的命令 如果环境运行的是 Apache Airflow v2.0.2，则如果 DAG 使用的插件依赖于通过 `requirements.txt` 安装的程序包，则解析 DAG 的 CLI 命令将会失败： **Apache Airflow v2.0.2** + `dags backfill` + `dags list` + `dags list-runs` + `dags next-execution` 如果 DAG 不使用依赖于通过 `requirements.txt` 安装的程序包的插件，则可以使用这些 CLI 命令。 ## 代码示例 下一节包含使用 Apache Airflow CLI 的不同方法的示例。 ### 设置、获取或删除 Apache Airflow v2 变量 您可以使用以下示例代码设置、获取或删除 `