命令行接口 · Airflow 中文文檔

# 命令行接口 > 貢獻者：[@ImPerat0R\_](https://github.com/tssujt) Airflow 具有非常豐富的命令行接口，允許在 DAG 上執行多種類型的操作，啟動服務以及支持開發和測試。 ```py usage: airflow [-h] {resetdb,render,variables,connections,create_user,pause,task_failed_deps,version,trigger_dag,initdb,test,unpause,dag_state,run,list_tasks,backfill,list_dags,kerberos,worker,webserver,flower,scheduler,task_state,pool,serve_logs,clear,upgradedb,delete_dag} ... ``` ## 必填參數 | 子命令 | 可能的選擇：resetdb，render，variables，connections，create_user，pause，task_failed_deps，version，trigger_dag，initdb，test，unpause，dag_state，run，list_tasks，backfill，list_dags，kerberos，worker，webserver，flower，scheduler，task_state，pool ，serve_logs，clear，upgrab，delete_dag 子命令幫助 | ## 子命令： ### resetdb 刪除并重建元數據數據庫 ```py airflow resetdb [-h] [-y] ``` #### 可選參數 | -y, --yes | 不要提示確認重置。請小心使用！默認值：False | ### render 渲染任務實例的模板 ```py airflow render [-h] [-sd SUBDIR] dag_id task_id execution_date ``` #### 必填參數 | dag_id | dag 的 id | | task_id | 任務的 id | | execution_date | DAG 的執行日期 | #### 可選參數 | -sd, --subdir | 從中查找 dag 的文件位置或目錄默認值：“[AIRFLOW_HOME]/dags” | ### 變量對變量的 CRUD 操作 ```py airflow variables [-h] [-s KEY VAL] [-g KEY] [-j] [-d VAL] [-i FILEPATH] [-e FILEPATH] [-x KEY] ``` #### 可選參數 | -s, --set | 設置變量 | | -g, --get | 獲取變量的值 | | -j, --json | 反序列化 JSON 變量默認值：False | | -d, --default | 如果變量不存在，則返回默認值 | | -i, --import | 從 JSON 文件導入變量 | | -e, --export | 將變量導出到 JSON 文件 | | -x, --delete | 刪除變量 | ### connections 列表/添加/刪除連接 ```py airflow connections [-h] [-l] [-a] [-d] [--conn_id CONN_ID] [--conn_uri CONN_URI] [--conn_extra CONN_EXTRA] [--conn_type CONN_TYPE] [--conn_host CONN_HOST] [--conn_login CONN_LOGIN] [--conn_password CONN_PASSWORD] [--conn_schema CONN_SCHEMA] [--conn_port CONN_PORT] ``` #### 可選參數 | -l，--list | 列出所有連接，默認值：False | | -a，--add | 添加連接，默認值：False | | -d，--delete | 刪除連接，默認值：False | | --conn_id | 連接 ID，添加/刪除連接時必填 | | --conn_uri | 連接 URI，添加沒有 conn_type 的連接時必填 | | --conn_extra | 連接的 Extra 字段，添加連接時可選 | | --conn_type | 連接類型，添加沒有 conn_uri 的連接時時必填 | | --conn_host | 連接主機，添加連接時可選 | | --conn_login | 連接登錄，添加連接時可選 | | --conn_password | 連接密碼，添加連接時可選 | | --conn_schema | 連接架構，添加連接時可選 | | --conn_port | 連接端口，添加連接時可選 | ### create_user 創建管理員帳戶 ```py airflow create_user [-h] [-r ROLE] [-u USERNAME] [-e EMAIL] [-f FIRSTNAME] [-l LASTNAME] [-p PASSWORD] [--use_random_password] ``` #### 可選參數 | -r，--role | 用戶的角色。現有角色包括 Admin，User，Op，Viewer 和 Public | | -u，--username | 用戶的用戶名 | | -e，--電子郵件 | 用戶的電子郵件 | | -f，--firstname | 用戶的名字 | | -l，--lastname | 用戶的姓氏 | | -p，--password | 用戶密碼 | | --use_random_password | 不提示輸入密碼。改為使用隨機字符串默認值：False | ### pause 暫停 DAG ```py airflow pause [-h] [-sd SUBDIR] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | ### task_failed_deps 從調度程序的角度返回任務實例的未滿足的依賴項。換句話說，為什么任務實例不會被調度程序調度然后排隊，然后由執行程序運行。 ```py airflow task_failed_deps [-h] [-sd SUBDIR] dag_id task_id execution_date ``` #### 必填參數 | dag_id | dag 的 id | | task_id | 任務的 id | | execution_date | DAG 的執行日期 | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | ### version 顯示版本 ```py airflow version [-h] ``` ### trigger_dag 觸發 DAG 運行 ```py airflow trigger_dag [-h] [-sd SUBDIR] [-r RUN_ID] [-c CONF] [-e EXEC_DATE] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | -r，--run_id | 幫助識別此次運行 | | -c，--conf | JSON 字符串被腌制到 DagRun 的 conf 屬性中 | | -e，--exec_date | DAG 的執行日期 | ### initdb 初始化元數據數據庫 ```py airflow initdb [-h] ``` ### 測試測試任務實例。這將在不檢查依賴關系或在數據庫中記錄其狀態的情況下運行任務。 ```py airflow test [-h] [-sd SUBDIR] [-dr] [-tp TASK_PARAMS] dag_id task_id execution_date ``` #### 必填參數 | dag_id | dag 的 id | | task_id | 任務的 id | | execution_date | DAG 的執行日期 | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | -dr，--dr_run | 進行干運行默認值：False | | -tp，--task_params | 向任務發送 JSON params dict | ### unpause 恢復暫停的 DAG ```py airflow unpause [-h] [-sd SUBDIR] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | ### dag_state 獲取 dag run 的狀態 ```py airflow dag_state [-h] [-sd SUBDIR] dag_id execution_date ``` #### 必填參數 | dag_id | dag 的 id | | execution_date | DAG 的執行日期 | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | ### run 運行單個任務實例 ```py airflow run [-h] [-sd SUBDIR] [-m] [-f] [--pool POOL] [--cfg_path CFG_PATH] [-l] [-A] [-i] [-I] [--ship_dag] [-p PICKLE] [-int] dag_id task_id execution_date ``` #### 必填參數 | dag_id | dag 的 id | | task_id | 任務的 id | | execution_date | DAG 的執行日期 | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | -m，--mark_success | 將作業標記為成功而不運行它們默認值：False | | -f，--force | 忽略先前的任務實例狀態，無論任務是否已成功/失敗，都重新運行，默認值：False | | --pool | 要使用的資源池 | | --cfg_path | 要使用的配置文件的路徑而不是 airflow.cfg | | -l，--local | 使用 LocalExecutor 運行任務，默認值：False | | -A，--ignore_all_dependencies | 忽略所有非關鍵依賴項，包括 ignore_ti_state 和 ignore_task_deps，默認值：False | | -i，--ignore_dependencies | 忽略特定于任務的依賴項，例如 upstream，depends_on_past 和重試延遲依賴項，默認值：False | | -I，--signore_depends_on_past | 忽略 depends_on_past 依賴項（但尊重上游依賴項），默認值：False | | --ship_dag | 泡菜（序列化）DAG 并將其運送給工人，默認值：False | | -p，--pickle | 整個 dag 的序列化 pickle 對象（內部使用） | | -int，--interactive | 不捕獲標準輸出和錯誤流（對交互式調試很有用），默認值：False | ### list_tasks 列出 DAG 中的任務 ```py airflow list_tasks [-h] [-t] [-sd SUBDIR] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -t，--tree | 樹視圖，默認值：False | | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | ### backfill 在指定的日期范圍內運行 DAG 的子部分如果使用 reset_dag_run 選項，則回填將首先提示用戶 Airflow 是否應清除回填日期范圍內的所有先前 dag_run 和 task_instances。如果使用 rerun_failed_tasks，則回填將自動重新運行回填日期范圍內的先前失敗的任務實例。 ```py airflow backfill [-h] [-t TASK_REGEX] [-s START_DATE] [-e END_DATE] [-m] [-l] [-x] [-i] [-I] [-sd SUBDIR] [--pool POOL] [--delay_on_limit DELAY_ON_LIMIT] [-dr] [-v] [-c CONF] [--reset_dagruns] [--rerun_failed_tasks] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -t，--task_regex | | | 用于過濾特定 task_ids 以回填的正則表達式（可選） | | -s，--start_date | | | 覆蓋 start_date YYYY-MM-DD | | -e，--end_date | 覆蓋 end_date YYYY-MM-DD | | -m，--mark_success | | | 將作業標記為成功而不運行它們，默認值：False | | -l，--local | 使用 LocalExecutor 運行任務，默認值：False | | -x，--donot_pickle | | | 不要試圖挑選 DAG 對象發送給工人，只要告訴工人運行他們的代碼版本。默認值：False | | -i，--ignore_dependencies | | | 跳過上游任務，僅運行與正則表達式匹配的任務。僅適用于 task_regex，默認值：False | | -I，--signore_first_depends_on_past | | | 僅忽略第一組任務的 depends_on_past 依賴關系（回填 DO 中的后續執行依賴 depends_on_past）。默認值：False | | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | --pool | 要使用的資源池 | | --delay_on_limit | | | 在嘗試再次執行 dag 運行之前達到最大活動 dag 運行限制（max_active_runs）時等待的時間（以秒為單位）。默認值：1.0 | | -dr，--dr_run | 進行干運行，默認值：False | | -v，--verbose | 使日志輸出更詳細，默認值：False | | -c，--conf | JSON 字符串被腌制到 DagRun 的 conf 屬性中 | | --reset_dagruns | | | 如果設置，則回填將刪除現有的與回填相關的 DAG 運行，并重新開始運行新的 DAG 運行，默認值：False | | --rerun_failed_tasks | | | 如果設置，則回填將自動重新運行回填日期范圍的所有失敗任務，而不是拋出異常，默認值：False | ### list_dags 列出所有 DAG ```py airflow list_dags [-h] [-sd SUBDIR] [-r] ``` #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | -r，--report | 顯示 DagBag 加載報告，默認值：False | ### kerberos 啟動 kerberos 票證續訂 ```py airflow kerberos [-h] [-kt [KEYTAB]] [--pid [PID]] [-D] [--stdout STDOUT] [--stderr STDERR] [-l LOG_FILE] [principal] ``` #### 必填參數 | principal | kerberos principal 默認值：airflow | #### 可選參數 | -kt，--keytab | 密鑰表默認值：airflow.keytab | | --pid | PID 文件位置 | | -D，--daemon | 守護進程而不是在前臺運行默認值：False | | --stdout | 將 stdout 重定向到此文件 | | --stderr | 將 stderr 重定向到此文件 | | -l，--log-file | 日志文件的位置 | ### worker 啟動 Celery 工作節點 ```py airflow worker [-h] [-p] [-q QUEUES] [-c CONCURRENCY] [-cn CELERY_HOSTNAME] [--pid [PID]] [-D] [--stdout STDOUT] [--stderr STDERR] [-l LOG_FILE] ``` #### 可選參數 | -p，--do_pickle | | | 嘗試將 DAG 對象發送給工作人員，而不是讓工作人員運行他們的代碼版本。默認值：False | | -q，--queue | 以逗號分隔的隊列列表，默認值：default | | -c, --concurrency | | | 工作進程的數量，默認值：16 | | -cn，--slowry_hostname | | | 如果一臺計算機上有多個 worker，請設置 celery worker 的主機名。 | | --pid | PID 文件位置 | | -D，--daemon | 守護進程而不是在前臺運行，默認值：False | | --stdout | 將 stdout 重定向到此文件 | | --stderr | 將 stderr 重定向到此文件 | | -l，--log-file | 日志文件的位置 | ### webserver 啟動 Airflow 網絡服務器實例 ```py airflow webserver [-h] [-p PORT] [-w WORKERS] [-k {sync,eventlet,gevent,tornado}] [-t WORKER_TIMEOUT] [-hn HOSTNAME] [--pid [PID]] [-D] [--stdout STDOUT] [--stderr STDERR] [-A ACCESS_LOGFILE] [-E ERROR_LOGFILE] [-l LOG_FILE] [--ssl_cert SSL_CERT] [--ssl_key SSL_KEY] [-d] ``` #### 可選參數 | -p，--port | 運行服務器的端口，默認值：8080 | | -w，--workers | 運行 Web 服務器的工作者數量，默認值：4 | | -k，--workerclass | | | 可能的選擇：sync，eventlet，gevent，tornado 用于 Gunicorn 的 worker class，默認值：sync | | -t，--worker_timeout | | | 等待 Web 服務器工作者的超時時間，默認值：120 | | -hn，--hostname | | | 設置運行 Web 服務器的主機名，默認值：0.0.0.0 | | --pid | PID 文件位置 | | -D，--daemon | 守護進程而不是在前臺運行，默認值：False | | --stdout | 將 stdout 重定向到此文件 | | --stderr | 將 stderr 重定向到此文件 | | -A，--access_logfile | | | 用于存儲 Web 服務器訪問日志的日志文件。使用'-'打印到 stderr。默認值：- | | -E，--error_logfile | | | 用于存儲 Web 服務器錯誤日志的日志文件。使用'-'打印到 stderr。默認值：- | | -l，--log-file | 日志文件的位置 | | --ssl_cert | Web 服務器的 SSL 證書的路徑 | | --ssl_key | 用于 SSL 證書的密鑰的路徑 | | -d，--debug | 在調試模式下使用 Flask 附帶的服務器，默認值：False | ### flower 運行 Celery Flower ```py airflow flower [-h] [-hn HOSTNAME] [-p PORT] [-fc FLOWER_CONF] [-u URL_PREFIX] [-a BROKER_API] [--pid [PID]] [-D] [--stdout STDOUT] [--stderr STDERR] [-l LOG_FILE] ``` #### 可選參數 | -hn，--hostname | | | 設置運行服務器的主機名，默認值：0.0.0.0 | | -p，--port | 運行服務器的端口，默認值：5555 | | -fc，--flowers_conf | | | celery 的配置文件 | | -u，--url_prefix | | | Flower 的 URL 前綴 | | -a，--broker_api | | | Broker api | | --pid | PID 文件位置 | | -D，--daemon | 守護進程而不是在前臺運行，默認值：False | | --stdout | 將 stdout 重定向到此文件 | | --stderr | 將 stderr 重定向到此文件 | | -l，--log-file | 日志文件的位置 | ### scheduler 啟動調度程序實例 ```py airflow scheduler [-h] [-d DAG_ID] [-sd SUBDIR] [-r RUN_DURATION] [-n NUM_RUNS] [-p] [--pid [PID]] [-D] [--stdout STDOUT] [--stderr STDERR] [-l LOG_FILE] ``` #### 可選參數 | -d，--dag_id | 要運行的 dag 的 id | | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | -r，--run-duration | | | 設置退出前執行的秒數 | | -n，--num_runs | 設置退出前要執行的運行次數，默認值：-1 | | -p，--do_pickle | | | 嘗試將 DAG 對象發送給工作人員，而不是讓工作人員運行他們的代碼版本。默認值：False | | --pid | PID 文件位置 | | -D，--daemon | 守護進程而不是在前臺運行默認值：False | | --stdout | 將 stdout 重定向到此文件 | | --stderr | 將 stderr 重定向到此文件 | | -l，--log-file | 日志文件的位置 | ### task_state 獲取任務實例的狀態 ```py airflow task_state [-h] [-sd SUBDIR] dag_id task_id execution_date ``` #### 必填參數 | dag_id | dag 的 id | | task_id | 任務的 id | | execution_date | DAG 的執行日期 | #### 可選參數 | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | ### pool pool 的 CRUD 操作 ```py airflow pool [-h] [-s NAME SLOT_COUNT POOL_DESCRIPTION] [-g NAME] [-x NAME] ``` #### 可選參數 | -s，--set | 分別設置池槽數和描述 | | -g，--get | 獲取池信息 | | -x，--delete | 刪除池 | ### serve_logs 由 worker 生成的服務日志 ```py airflow serve_logs [-h] ``` ### clear 清除一組任務實例，就好像它們從未運行過一樣 ```py airflow clear [-h] [-t TASK_REGEX] [-s START_DATE] [-e END_DATE] [-sd SUBDIR] [-u] [-d] [-c] [-f] [-r] [-x] [-xp] [-dx] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -t，--task_regex | | | 用于過濾特定 task_ids 以回填的正則表達式（可選） | | -s，--start_date | | | 覆蓋 start_date YYYY-MM-DD | | -e，--end_date | 覆蓋 end_date YYYY-MM-DD | | -sd，--subdir | 從中查找 dag 的文件位置或目錄，默認值：“[AIRFLOW_HOME]/dags” | | -u，--upstream | 包括上游任務，默認值：False | | -d，--downstream | | | 包括下游任務，默認值：False | | -c，--no_confirm | | | 請勿要求確認，默認值：False | | -f，--only_failed | | | 只有失敗的工作，默認值：False | | -r，--only_running | | | 只運行工作，默認值：False | | -x，--exclude_subdags | | | 排除子標記，默認值：False | | -dx，--dag_regex | | | 將 dag_id 搜索為正則表達式而不是精確字符串，默認值：False | ### upgradedb 將元數據數據庫升級到最新版本 ```py airflow upgradedb [-h] ``` ### delete_dag 刪除與指定 DAG 相關的所有 DB 記錄 ```py airflow delete_dag [-h] [-y] dag_id ``` #### 必填參數 | dag_id | dag 的 id | #### 可選參數 | -y，--是的 | 不要提示確認重置。小心使用！默認值：False |