Apache Airflow CLI 命令参考

本页介绍了 Amazon MWAA 上支持和不支持的 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 列表导入错误
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+	dag 运行的任务状态
v2.0+	测试任务
v2.0+	删除变量
v2.0+	获取变量
v2.0+	设置变量
v2.0+	列出变量
v2.0+	版本

使用解析 DAG 的命令

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

如果 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 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 教程