Apache Airflow CLI 命令參考

本主題說明 Amazon Managed Workflows for Apache Airflow 上支援和不支援的 Apache Airflow CLI 命令。

提示

REST API 比 CLI 更現代化，旨在與外部系統進行程式設計整合。REST 是與 Apache Airflow 互動的偏好方式。

內容

• 先決條件

  • 存取

  • AWS CLI

• v2 中的變更

• 支援的 CLI 命令

  • 支援的命令

  • 使用剖析 DAGs命令

• 範本程式碼

  • 設定、取得或刪除 Apache Airflow v2 變數

  • 觸發 DAG 時新增組態

  • 在對堡壘主機的 SSH 通道上執行 CLI 命令

  • GitHub 中的範例和 AWS 教學課程

先決條件

下一節說明使用此頁面上的命令和指令碼所需的初步步驟。

存取

• AWS AWS Identity and Access Management (IAM) 中對 Amazon MWAA 許可政策的 帳戶存取Apache Airflow UI 存取政策：AmazonMWAAWebServerAccess。

• AWS AWS Identity and Access Management (IAM) 中對 Amazon MWAA 許可政策 的 帳戶存取完整 API 和主控台存取政策：AmazonMWAAFullApiAccess。

AWS CLI

AWS Command Line Interface (AWS CLI) 是一種開放原始碼工具，可讓您使用命令列 shell 中的命令與 AWS 服務互動。若要完成此頁面上的步驟，您需要下列項目：

• AWS CLI – 安裝第 2 版。

• AWS CLI – 使用 進行快速組態aws configure。

v2 中的變更

• 新增：Airflow CLI 命令結構。Apache Airflow v2 CLI 經過組織，因此相關命令會分組為子命令，這表示如果您想要升級至 Apache Airflow v2，則需要更新 Apache Airflow v1 指令碼。例如，unpauseApache Airflow v1 現在位於 dags unpause Apache Airflow v2。若要進一步了解，請參閱《Apache Airflow 參考指南》中的 2 中的 Airflow CLI 變更。

支援的 CLI 命令

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

支援的命令

Apache Airflow v2

次要版本	Command
v2.0+	cheat-sheet
v2.0+	連線新增
v2.0+	連線刪除
v2.2+ (備註)	dags 回填
v2.0+	dags 刪除
v2.2+ (備註)	dags 清單
v2.0+	dags list-jobs
v2.6+	dags list-import-errors
v2.2+ (備註)	dags list-runs
v2.2+ (備註)	dags next-execution
v2.0+	dags 暫停
v2.0+	dags 報告
v2.4+	dags 重新序列化
v2.0+	dags show
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+	角色 add-perms
v2.6+	角色 del-perms
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+	version

使用剖析 DAGs命令

如果您的環境執行的是 Apache Airflow v1.10.12 或 v2.0.2，如果 DAG 使用依賴透過 安裝的套件的外掛程式，則剖析 DAGs CLI 命令將會失敗requirements.txt：

Apache Airflow 2.0.2 版

• dags backfill

• dags list

• dags list-runs

• dags next-execution

如果您的 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 命令

下列範例示範如何使用 SSH 通道代理對 Linux 堡壘主機執行 Airflow CLI 命令。

使用 curl

1. ssh -D 8080 -f -C -q -N YOUR_USER@YOUR_BASTION_HOST

2. curl -x socks5h://0:8080 --request POST https://YOUR_HOST_NAME/aws_mwaa/cli --header YOUR_HEADERS --data-raw YOUR_CLI_COMMAND

GitHub 中的範例和 AWS 教學課程

• 在 Amazon Managed Workflows for Apache Airflow 中使用 Apache Airflow 2.0.2 版參數和變數

• 透過命令列在 Amazon MWAA 上與 Apache Airflow v1.10.12 互動

• Amazon MWAA 上的 Apache Airflow v1.10.12 和 GitHub 上的 Bash Operator 的互動式命令 GitHub