本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

Apache Airflow CLI 命令参考

本主题介绍适用于 Apache Airflow 的亚马逊托管工作流程上支持和不支持的 Apache Airflow CLI 命令。

先决条件

下一节介绍了使用本页上的命令和脚本所需的初步步骤。

访问

AWS CLI

AWS Command Line Interface (AWS CLI) 是一个开源工具，可让您使用命令行 shell 中的命令与 AWS 服务进行交互。要完成本节中的步骤，您需要以下满足以下条件：

v2 中发生了什么变化

支持的 CLI 命令

下一节列出了 Amazon MWAA 上可用的 Apache Airflow CLI 命令。

支持的 命令

Apache Airflow v2

次要版本	命令
v2.0+	备忘单
v2.0+	添加连接
v2.0+	删除连接
v2.2+（注意）	回填 dags
v2.0+	删除 dags
v2.2+（注意）	dags 列表
v2.0+	dags 列表-任务
v2.6+	dags list-import-errors
v2.2+（注意）	dags 列表-运行次数
v2.2+（注意）	dags 下次-执行
v2.0+	暂停 dag
v2.0+	报告 dags
v2.4+	重新序列化 dags
v2.0+	显示 dags
v2.0+	陈述 dags
v2.0+	测试 dags
v2.0+	触发 dags
v2.0+	取消暂停 dags
v2.4+	清理 db
v2.0+	提供商的行为
v2.0+	获得提供商
v2.0+	提供商挂钩
v2.0+	提供商链接
v2.0+	提供商列表
v2.8+	提供程序通知
v2.6+	提供商密钥
v2.7+	提供商触发器
v2.0+	提供商小部件
v2.6+	角色添加权限
v2.6+	角色删除权限
v2.6+	角色创建
v2.0+	列出角色
v2.0+	清除任务
v2.0+	任务失败-部署
v2.0+	列出任务
v2.0+	渲染任务
v2.0+	陈述任务
v2.0+	任务 states-for-dag-run
v2.0+	测试任务
v2.0+	删除变量
v2.0+	获取变量
v2.0+	设置变量
v2.0+	列出变量
v2.0+	版本

使用解析命令 DAGs

如果您的环境运行的是 Apache Airflow v1.10.12 或 v2.0.2，则如果 DAG 使用的插件依赖于通过以下方式安装的软件包 DAGs ，则解析的 CLI 命令将失败：requirements.txt

如果您 DAGs 不使用依赖于通过安装的软件包的插件，则可以使用这些 CLI 命令requirements.txt。

代码示例

下一节包含使用 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 v1 和 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 命令。

中的示例 GitHub 和 AWS 教程