Apache Airflow CLI 命令参考
本主题介绍了 Amazon MWAA 上支持和不支持的 Apache Airflow CLI 命令。
提示
REST API 比 CLI 更现代，专为与外部系统的编程集成而设计。REST 是与 Apache Airflow 交互的首选方式。
目录
先决条件
下一节介绍了使用本页上的命令和脚本所需的初步步骤。
访问
-
在 AWS Identity and Access Management (IAM) 中访问 Apache Airflow UI 访问策略：AmazonMWAAWebServerAccess 中的 Amazon MWAA 权限策略的 AWS 账户。
-
在 AWS Identity and Access Management (IAM) 中访问 Amazon MWAA 权限策略 完整的 API 和控制台访问策略：AmazonMWAAFullApiAccess 的 AWS 账户。
AWS CLI
AWS Command Line Interface (AWS CLI) 是一种开源工具，您可以用来在命令行 Shell 中使用命令与 AWS 服务进行交互。要完成本节中的步骤，您需要以下满足以下条件：
更改了哪些内容？
-
v3：Airflow 架构。Apache Airflow v3 引入了突破性的架构更改，以提高安全性和可扩展性，并简化维护。要了解更多信息，请参阅升级到 Airflow 3
。 -
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 更改。
支持的 CLI 命令
下一节列出了 Amazon MWAA 上可用的 Apache Airflow CLI 命令。
支持的命令
使用解析 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 变量
您可以使用以下示例代码设置、获取或删除 <script> <mwaa env name> get | set | delete <variable> <variable value> </variable> </variable> 格式的变量。
[ $# -eq 0 ] && echo "Usage: $0 MWAA environment name " && exit if [[ $2 == "" ]]; then dag="variables list" elif [ $2 == "get" ] || [ $2 == "delete" ] || [ $2 == "set" ]; then dag="variables $2 $3 $4 $5" else echo "Not a valid command" exit 1 fi CLI_JSON=$(aws mwaa --region $AWS_REGION create-cli-token --name $1) \ && 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 "$dag" ) \ && echo "Output:" \ && echo $CLI_RESULTS | jq -r '.stdout' | base64 --decode \ && echo "Errors:" \ && echo $CLI_RESULTS | jq -r '.stderr' | base64 --decode
触发 DAG 时添加配置
您可以在 Apache Airflow v2 中使用以下示例代码在触发 DAG 时添加配置，例如 airflow trigger_dag 'dag_name' —conf '{"key":"value"}'。
import boto3 import json import requests import base64 mwaa_env_name = 'YOUR_ENVIRONMENT_NAME' dag_name = 'YOUR_DAG_NAME' key = "YOUR_KEY" value = "YOUR_VALUE" conf = "{\"" + key + "\":\"" + value + "\"}" 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 = "trigger_dag {0} -c '{1}'".format(dag_name, conf) 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)
在通往堡垒主机的 SSH 隧道上运行 CLI 命令。
根据以下示例使用连接到 Linux 堡垒主机的 SSH 隧道代理运行 Airflow CLI 命令。
使用 curl
-
ssh -D 8080 -f -C -q -NYOUR_USER@YOUR_BASTION_HOST -
curl -x socks5h://0:8080 --request POST https://YOUR_HOST_NAME/aws_mwaa/cli --headerYOUR_HEADERS--data-rawYOUR_CLI_COMMAND
