本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。
在 Amazon MWAA 上使用 Apache Airflow 組態選項
Apache Airflow 組態選項可以做為環境變數連接到 Amazon Managed Workflows for Apache Airflow 環境。您可以從建議的下拉式清單中選擇,或在 Amazon MWAA 主控台上為您的 Apache Airflow 版本指定自訂組態選項。本主題說明可用的 Apache Airflow 組態選項,以及如何使用這些選項來覆寫您環境中的 Apache Airflow 組態設定。
內容
先決條件
您將需要下列項目,才能完成此頁面上的步驟。
-
許可 — 您的管理員 AWS 帳戶 必須已授予您環境的 AmazonMWAAFullConsoleAccess 存取控制政策的存取權。此外,您的執行角色必須允許 Amazon MWAA 環境,才能存取環境所使用的 AWS 資源。
-
存取 — 如果您需要存取公有儲存庫,才能直接在 Web 伺服器上安裝相依性,您的環境必須設定公有網路 Web 伺服器存取。如需詳細資訊,請參閱 Apache Airflow 存取模式。
-
Amazon S3 組態 — 用於在 中存放 DAGs、自訂外掛程式
plugins.zip和 Python 相依性的 Amazon S3 儲存貯體requirements.txt必須設定為啟用公開存取封鎖和版本控制。
運作方式
當您建立環境時,Amazon MWAA 會將您在 Airflow 組態選項中的 Amazon MWAA 主控台上指定的組態設定,做為環境變數連接到您環境的 AWS Fargate 容器。如果您在 中使用相同名稱的設定airflow.cfg,則您在 Amazon MWAA 主控台上指定的選項會覆寫 中的值airflow.cfg。
雖然預設不會在 Amazon MWAA 環境的 Apache Airflow UI airflow.cfg中公開 ,但您可以直接在 Amazon MWAA 主控台上變更 Apache Airflow 組態選項,包括webserver.expose_config公開組態的設定。
使用組態選項載入外掛程式
在 Apache Airflow v2 和更新版本中,外掛程式預設為使用 core.lazy_load_plugins : True設定「延遲」載入。如果您使用的是自訂外掛程式,則必須將 新增core.lazy_load_plugins : False為 Apache Airflow 組態選項,才能在每個 Airflow 程序開始時載入外掛程式,以覆寫預設設定。
組態選項概觀
當您在 Amazon MWAA 主控台上新增組態時,Amazon MWAA 會將組態寫入環境變數。
-
列出的選項。您可以在下拉式清單中選擇適用於 Apache Airflow 版本的其中一個組態設定。例如,
dag_concurrency:16。組態設定會轉譯為您環境的 Fargate 容器,做為AIRFLOW__CORE__DAG_CONCURRENCY : 16 -
自訂選項。您也可以在下拉式清單中指定 Apache Airflow 版本未列出的 Airflow 組態選項。例如,
foo.user:YOUR_USER_NAME。組態設定會翻譯為您環境的 Fargate 容器,做為AIRFLOW__FOO__USER : YOUR_USER_NAME
Apache Airflow 組態選項
下圖說明您可以在 Amazon MWAA 主控台上自訂 Apache Airflow 組態選項的位置。
Apache Airflow 參考
如需 Apache Airflow 支援的組態選項清單,請參閱《Apache Airflow 參考指南》中的組態
使用 Amazon MWAA 主控台
下列程序會逐步解說將 Airflow 組態選項新增至環境的步驟。
-
在 Amazon MWAA 主控台上開啟環境
頁面。 -
選擇環境。
-
選擇編輯。
-
選擇下一步。
-
在 Airflow 組態選項窗格中選擇新增自訂組態。
-
從下拉式清單中選擇組態並輸入值,或輸入自訂組態並輸入值。
-
針對您要新增的每個組態,選擇新增自訂組態。
-
選擇儲存。
組態參考
下一節包含 Amazon MWAA 主控台下拉式清單中可用 Apache Airflow 組態的清單。
電子郵件組態
以下清單顯示適用於 Apache Airflow v2 和 v3 的 Amazon MWAA 上可用的 Airflow 電子郵件通知組態選項。
我們建議將連接埠 587 用於 SMTP 流量。根據預設, 會 AWS 封鎖所有 Amazon EC2 執行個體連接埠 25 上的傳出 SMTP 流量。如果您想要在連接埠 25 上傳送傳出流量,您可以請求移除此限制
| 氣流組態選項 | 描述 | 範例值 |
|---|---|---|
|
email.email_backend |
用於 email_backend |
airflow.utils.email.send_email_smtp |
|
smtp.smtp_host |
用於 smtp_host |
localhost |
|
smtp.smtp_starttls |
Transport Layer Security (TLS) 用於在 smtp_starttls |
False |
|
smtp.smtp_ssl |
Secure Sockets Layer (SSL) 用於連接 smtp_ssl |
True |
|
smtp.smtp_port |
587 |
|
|
smtp.smtp_mail_from |
smtp_mail_from |
myemail@domain.com |
任務組態
以下清單顯示 Amazon MWAA for Apache Airflow v2 和 v3 上 Airflow 任務下拉式清單中可用的組態。
| 氣流組態選項 | 描述 | 範例值 |
|---|---|---|
|
core.default_task_retries |
在 default_task_retries |
3 |
|
core.parallelism |
可在整個環境中平行同時執行的任務執行個體數量上限 (平行處理 |
40 |
排程器組態
下列清單顯示在適用於 Apache Airflow v2 和 v3 的 Amazon MWAA 下拉式清單中可用的 Apache Airflow 排程器組態。
| 氣流組態選項 | 描述 | 範例值 |
|---|---|---|
|
scheduler.catchup_by_default |
指示排程器建立 DAG 執行,以在 catchup_by_default |
False |
|
scheduler.scheduler_zombie_task_threshold 注意不適用於 Apache Airflow v3。 |
告知排程器是否要將任務執行個體標記為失敗,並在 scheduler_zombie_task_threshold |
300 |
工作者組態
下列清單顯示在適用於 Apache Airflow v2 和 v3 的 Amazon MWAA 下拉式清單中可用的 Airflow 工作者組態。
| 氣流組態選項 | 描述 | 範例值 |
|---|---|---|
|
celery.worker_autoscale |
在 worker_autoscale |
16,12 |
Web 伺服器組態
下列清單顯示在適用於 Apache Airflow v2 和 v3 的 Amazon MWAA 下拉式清單中可用的 Apache Airflow Webserver 組態。
| 氣流組態選項 | 描述 | 範例值 |
|---|---|---|
|
webserver.default_ui_timezone 注意不適用於 Apache Airflow v3。 |
default_ui_timezone 注意設定 |
America/New_York |
觸發器組態
以下清單顯示適用於 Apache Airflow v2 和 v3 的 Amazon MWAA 上可用的 Apache Airflow 觸發器
| 氣流組態選項 | 描述 | 範例值 |
|---|---|---|
|
mwaa.trigger_enabled |
用於在 Amazon MWAA 上啟用和停用觸發器。依預設,此值是設為 |
True |
|
triggerer.default_capacity (在 v2 中) triggerer.capacity (在 v3 中) |
定義每個觸發器可以平行執行的觸發次數。在 Amazon MWAA 上,隨著兩個元件一起執行,每個觸發器和每個排程器都會設定此容量。對於 |
125 |
範例和範例程式碼
範例 DAG
您可以使用下列 DAG 來列印 email_backend Apache Airflow 組態選項。若要執行 以回應 Amazon MWAA 事件,請將程式碼複製到 Amazon S3 儲存貯體上的環境 DAGs 資料夾。
from airflow.decorators import dag from datetime import datetime def print_var(**kwargs): email_backend = kwargs['conf'].get(section='email', key='email_backend') print("email_backend") return email_backend @dag( dag_id="print_env_variable_example", schedule_interval=None, start_date=datetime(yyyy,m,d), catchup=False, ) def print_variable_dag(): email_backend_test = PythonOperator( task_id="email_backend_test", python_callable=print_var, provide_context=True ) print_variable_test = print_variable_dag()
電子郵件通知設定範例
下列 Apache Airflow 組態選項可用於使用應用程式密碼的 Gmail.com 電子郵件帳戶。如需詳細資訊,請參閱 Gmail 說明參考指南中的使用應用程式密碼登入
後續步驟?
-
了解如何將 DAG 資料夾上傳至 中的 Amazon S3 儲存貯體新增或更新 DAGs。