View a markdown version of this page

Amazon MWAA 上 Apache Airflow 的效能調校 - Amazon Managed Workflows for Apache Airflow
新增 Apache Airflow 組態選項Apache Airflow 排程器DAG 資料夾DAG 檔案任務​

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

Amazon MWAA 上 Apache Airflow 的效能調校

本主題說明如何使用 調整 Amazon Managed Workflows for Apache Airflow 環境的效能在 Amazon MWAA 上使用 Apache Airflow 組態選項。

新增 Apache Airflow 組態選項

使用下列程序將 Airflow 組態選項新增至您的環境。

  1. 在 Amazon MWAA 主控台上開啟環境頁面。

  2. 選擇環境。

  3. 選擇編輯。

  4. 選擇下一步。

  5. 在 Airflow 組態選項窗格中選擇新增自訂組態。

  6. 從下拉式清單中選擇組態並輸入值，或輸入自訂組態並輸入值。

  7. 針對您要新增的每個組態，選擇新增自訂組態。

  8. 選擇儲存。

若要進一步了解，請參閱 在 Amazon MWAA 上使用 Apache Airflow 組態選項。

Apache Airflow 排程器

Apache Airflow 排程器是 Apache Airflow 的核心元件。排程器的問題可防止剖析 DAGs 和排程任務。如需 Apache Airflow 排程器調校的詳細資訊，請參閱 Apache Airflow 文件網站上的微調排程器效能。

Parameters

本節說明 Apache Airflow 排程器 (Apache Airflow v2 及更新版本） 可用的組態選項及其使用案例。

Apache Airflow v3
組態 使用案例

celery.sync_parallelism

Celery Executor 用來同步任務狀態的程序數目。

預設值：1

使用此選項可透過限制 Celery Executor 使用的程序來防止佇列衝突。根據預設， 值會設定為 1 ，以防止將任務日誌交付至 CloudWatch Logs 時發生錯誤。將值設定為 0表示使用最大數量的程序，但在交付任務日誌時可能會導致錯誤。

scheduler.scheduler_idle_sleep_time

如果迴圈中沒有任何動作，則在連續排程器「迴圈」之間等待的秒數。

預設值：1

使用此選項可在排程器完成「迴圈」後增加排程器休眠的時間，以釋放排程器上的 CPU 用量。增加此值可減少 dag_processor.parsing_processes Apache Airflow v2 和 Apache Airflow v3 中可用的排程器執行緒。這可以減少排程器剖析 DAGs 的容量，並增加 DAGs 在 Web 伺服器中填入所需的時間。

scheduler.max_dagruns_to_create_per_loop

為每個排程器「迴圈」建立 DagRuns 的 DAGs 數目上限。

預設值：10

使用此選項可透過減少排程器「迴圈」的 DagRuns 數目上限，釋放用於排程任務的資源。

dag_processor.parsing_processes

排程器可以平行執行以排程 DAGs執行緒數目。

預設：使用 (2 * number of vCPUs) - 1

使用此選項可透過減少排程器平行執行以剖析 DAGs 的程序數量來釋放資源。如果 DAG 剖析影響任務排程，建議您將此數字保持低。您必須指定小於您環境中 vCPU 計數的值。若要進一步了解，請參閱限制。
Apache Airflow v2
組態 使用案例

celery.sync_parallelism

Celery Executor 用來同步任務狀態的程序數目。

預設值：1

使用此選項可透過限制 Celery Executor 使用的程序來防止佇列衝突。根據預設， 值會設定為 1 ，以防止將任務日誌交付至 CloudWatch Logs 時發生錯誤。將值設定為 0表示使用最大數量的程序，但在交付任務日誌時可能會導致錯誤。

scheduler.scheduler_idle_sleep_time

如果迴圈中沒有任何動作，則在連續排程器「迴圈」之間等待的秒數。

預設值：1

使用此選項可在排程器完成「迴圈」後增加排程器休眠的時間，以釋放排程器上的 CPU 用量。增加此值可減少 dag_processor.parsing_processes Apache Airflow v2 和 Apache Airflow v3 中可用的排程器執行緒。這可以減少排程器剖析 DAGs 的容量，並增加 DAGs 在 Web 伺服器中填入所需的時間。

scheduler.max_dagruns_to_create_per_loop

為每個排程器「迴圈」建立 DagRuns 的 DAGs 數目上限。

預設值：10

使用此選項可透過減少排程器「迴圈」的 DagRuns 數目上限，釋放用於排程任務的資源。

scheduler.parsing_processes

排程器可以平行執行以排程 DAGs執行緒數目。

預設：使用 (2 * number of vCPUs) - 1

使用此選項可透過減少排程器平行執行以剖析 DAGs 的程序數量來釋放資源。如果 DAG 剖析影響任務排程，建議您將此數字保持低。您必須指定小於您環境中 vCPU 計數的值。若要進一步了解，請參閱限制。

限制

本節說明調整排程器的預設參數時應考慮的限制。

scheduler.parsing_processes、scheduler.max_threads （僅限 v2)

環境類別的每個 vCPU 允許兩個執行緒。環境類別的排程器必須至少保留一個執行緒。如果您注意到任務排程延遲，您可能需要增加環境類別。例如，大型環境的排程器具有 4 個 vCPU Fargate 容器執行個體。這表示執行緒7總數上限為 ，可用於其他程序。也就是說，兩個執行緒會乘以四個 vCPUs，減去排程器本身的一個 vCPU。您在 scheduler.max_threads（僅限 v2) 中指定的值，且scheduler.parsing_processes不得超過 環境類別可用的執行緒數目，如下所示：

  • mw1.small – 不得超過其他程序的1執行緒。剩餘的執行緒會保留給排程器。

  • mw1.medium – 不得超過其他程序的3執行緒。剩餘的執行緒會保留給排程器。

  • mw1.large – 不得超過其他程序的7執行緒。剩餘的執行緒會保留給排程器。

DAG 資料夾

Apache Airflow 排程器會持續掃描您環境中DAGs 資料夾。任何包含plugins.zip的檔案，或包含「airflow」匯入陳述式的 Python (.py) 檔案。然後，任何產生的 Python DAG 物件都會放入 DagBag 中，以供排程器處理該檔案，以決定需要排程哪些任務。無論檔案是否包含任何可行的 DAG 物件，都會進行大型檔案剖析。

Parameters

本節說明 DAGs (Apache Airflow v2 及更新版本） 可用的組態選項及其使用案例。

Apache Airflow v3
組態 使用案例

dag_processor.refresh_interval

必須掃描 DAGs是否有新檔案的秒數。

預設：300 秒

使用此選項可透過增加剖析 DAGs秒數來釋放資源。如果您在 中遇到長時間的剖析時間total_parse_time metrics，建議您增加此值，這可能是因為 DAGs資料夾中有大量檔案所致。

dag_processor.min_file_process_interval

排程器剖析 DAG 並反映 DAG 更新的秒數。

預設：30 秒

使用此選項可透過增加排程器在剖析 DAG 之前等待的秒數來釋放資源。例如，如果您指定 的值30，則會每 30 秒剖析一次 DAG 檔案。我們建議您將此數字保持高，以減少環境中的 CPU 用量。
Apache Airflow v2
組態 使用案例

scheduler.dag_dir_list_interval

必須掃描 DAGs是否有新檔案的秒數。

預設：300 秒

使用此選項透過增加剖析 DAGs秒數來釋放資源。如果您在 中遇到長時間的剖析時間total_parse_time metrics，建議您增加此值，這可能是因為 DAGs資料夾中有大量檔案所致。

scheduler.min_file_process_interval

排程器剖析 DAG 並反映 DAG 更新的秒數。

預設：30 秒

使用此選項可透過增加排程器在剖析 DAG 之前等待的秒數來釋放資源。例如，如果您指定 的值30，則會每 30 秒剖析一次 DAG 檔案。我們建議您將此數字保持高，以減少環境中的 CPU 用量。

DAG 檔案

做為 Apache Airflow 排程器迴圈的一部分，會剖析個別 DAG 檔案以擷取 DAG Python 物件。在 Apache Airflow v2 和更新版本中，排程器會同時剖析最多 個剖析程序。scheduler.min_file_process_interval (v2) 或 dag_processor.min_file_process_interval(v3) 中指定的秒數必須先通過，才能再次剖析相同的檔案。

Parameters

本節說明 Apache Airflow DAG 檔案 (Apache Airflow v2 及更新版本） 可用的組態選項及其使用案例。

Apache Airflow v3
組態 使用案例

dag_processor.dag_file_processor_timeout

DagFileProcessor 逾時處理 DAG 檔案之前的秒數。

預設：50 秒

使用此選項可增加 DagFileProcessor 逾時之前所需的時間。如果您在 DAG 處理日誌中遇到逾時，導致沒有載入可行DAGs，建議您增加此值。

core.dagbag_import_timeout

匯入 Python 檔案逾時之前的秒數。

預設：30 秒

使用此選項可增加在匯入 Python 檔案以擷取 DAG 物件時，排程器逾時之前所花費的時間。此選項會做為排程器「迴圈」的一部分處理，且必須包含小於 中指定值的值dag_processor.dag_file_processor_timeout。

core.min_serialized_dag_update_interval

更新資料庫中序列化 DAGs的最小秒數。

預設：30

使用此選項可透過增加資料庫中序列化 DAGs更新後的秒數來釋放資源。如果您有大量 DAGs 或複雜的 DAGs，建議您增加此值。當 DAGs 序列化時，增加此值可減少排程器和資料庫的負載。

core.min_serialized_dag_fetch_interval

當已載入 DagBag 時，從資料庫重新擷取序列化 DAG 的秒數。

預設值：10

使用此選項可透過增加重新擷取序列化 DAG 的秒數來釋放資源。值必須大於 中指定的值core.min_serialized_dag_update_interval，以降低資料庫「寫入」速率。隨著 DAGs 序列化，增加此值可減少 Web 伺服器和資料庫的負載。
Apache Airflow v2
組態 使用案例

core.dag_file_processor_timeout

DagFileProcessor 逾時處理 DAG 檔案之前的秒數。

預設：50 秒

使用此選項可增加 DagFileProcessor 逾時之前所需的時間。如果您在 DAG 處理日誌中遇到逾時，導致沒有載入可行DAGs，建議您增加此值。

core.dagbag_import_timeout

匯入 Python 檔案逾時之前的秒數。

預設：30 秒

使用此選項可增加在匯入 Python 檔案以擷取 DAG 物件時，排程器逾時之前所花費的時間。此選項會做為排程器「迴圈」的一部分處理，且必須包含小於 中指定值的值core.dag_file_processor_timeout。

core.min_serialized_dag_update_interval

更新資料庫中序列化 DAGs的最小秒數。

預設：30

使用此選項可透過增加資料庫中序列化 DAGs更新後的秒數來釋放資源。如果您有大量 DAGs 或複雜的 DAGs，建議您增加此值。當 DAGs 序列化時，增加此值可減少排程器和資料庫的負載。

core.min_serialized_dag_fetch_interval

當已載入 DagBag 時，從資料庫重新擷取序列化 DAG 的秒數。

預設值：10

使用此選項可透過增加重新擷取序列化 DAG 的秒數來釋放資源。值必須大於 中指定的值core.min_serialized_dag_update_interval，以降低資料庫「寫入」速率。隨著 DAGs 序列化，增加此值可減少 Web 伺服器和資料庫的負載。

任務​

Apache Airflow 排程器和工作者都參與佇列和取消佇列任務。排程器會將準備從無狀態排程的剖析任務轉換為已排程狀態。執行器也會在 Fargate 的排程器容器上執行，將這些任務排入佇列，並將其狀態設定為已排入佇列。當工作者有容量時，會從佇列中取得任務，並將狀態設定為執行中，然後根據任務是否成功，將其狀態變更為成功或失敗。

Parameters

本節說明 Apache Airflow 任務可用的組態選項及其使用案例。

Amazon MWAA 覆寫的預設組態選項會以紅色標示。

Apache Airflow v3
組態 使用案例

core.parallelism

每個排程器可同時監控和執行的任務執行個體數量上限。

預設：根據 動態設定(maxWorkers * maxCeleryWorkers) / schedulers * 1.5。

使用此選項來控制執行中任務總數的硬式全域上限。例如，您可以增加此值，以保護中繼資料庫免於過多並行連線或限制成本。預設值很高，因此其他控制項 (worker_autoscale、集區插槽、max_active_tasks_per_dag) 可做為有效的並行限制。

core.max_active_tasks_per_dag

允許在每個 DAG 執行中同時執行的任務執行個體數目上限。

預設：16

使用此選項可透過增加允許同時執行的任務執行個體數量來釋放資源。例如，如果您有 100 DAGs 10 個平行任務的 DAG，並希望所有 DAGs 同時執行，請計算平行處理上限。將可用工作者的數量乘以 中的任務密度celery.worker_concurrency，然後除以 DAGs 的數量。

core.execute_tasks_new_python_interpreter

決定 Apache Airflow 是否透過強制父程序或建立新的 Python 程序來執行任務。

預設：True

設定為 時True，Apache Airflow 會將您對外掛程式所做的變更辨識為新的 Python 程序，以便建立以執行任務。

celery.worker_concurrency

Amazon MWAA 會覆寫此選項的 Airflow 基本安裝，將工作者擴展為其自動擴展元件的一部分。

預設：不適用

此選項指定的任何值都會遭到忽略。

celery.worker_autoscale

工作者的任務並行。

預設值：

  • mw1.micro - 3，0

  • mw1.small - 5，0

  • mw1.medium - 10，0

  • mw1.large - 20，0

  • mw1.xlarge - 40，0

  • mw1.2xlarge - 80，0

使用此選項可透過減少工作者的任務minimum並行maximum來釋放資源。工作者最多接受設定的maximum並行任務，無論是否有足夠資源可以這樣做。如果任務排程時沒有足夠的資源，任務會立即失敗。我們建議將此值變更為資源密集型任務，方法是將值減少為小於預設值，以允許每個任務更多的容量。
Apache Airflow v2
組態 使用案例

core.parallelism

每個排程器可同時監控和執行的任務執行個體數量上限。

預設：根據 動態設定(maxWorkers * maxCeleryWorkers) / schedulers * 1.5。

使用此選項來控制執行中任務總數的硬式全域上限。例如，您可以增加此值，以保護中繼資料庫免於過多並行連線或限制成本。預設值很高，因此其他控制項 (worker_autoscale、集區插槽、max_active_tasks_per_dag) 可做為有效的並行限制。

core.dag_concurrency

允許為每個 DAG 同時執行的任務執行個體數目。

預設：16

此組態自 Airflow 2.2.0 以來已棄用，並由 取代。 core.max_active_tasks_per_dag

使用此選項可透過增加允許同時執行的任務執行個體數量來釋放資源。例如，如果您有 100 DAGs 10 個平行任務的 DAG，並希望所有 DAGs 同時執行，請計算平行處理上限。將可用工作者的數量乘以 中的任務密度celery.worker_concurrency，然後除以 DAGs 的數量。

core.execute_tasks_new_python_interpreter

決定 Apache Airflow 是否透過強制父程序或建立新的 Python 程序來執行任務。

預設：True

設定為 時True，Apache Airflow 會將您對外掛程式所做的變更辨識為新的 Python 程序，以便建立以執行任務。

celery.worker_concurrency

Amazon MWAA 會覆寫此選項的 Airflow 基本安裝，將工作者擴展為其自動擴展元件的一部分。

預設：不適用

此選項指定的任何值都會遭到忽略。

celery.worker_autoscale

工作者的任務並行。

預設值：

  • mw1.micro - 3，0

  • mw1.small - 5，0

  • mw1.medium - 10，0

  • mw1.large - 20，0

  • mw1.xlarge - 40，0

  • mw1.2xlarge - 80，0

使用此選項可透過減少工作者的任務minimum並行maximum來釋放資源。工作者最多接受設定的maximum並行任務，無論是否有足夠資源可以這樣做。如果任務排程時沒有足夠的資源，任務會立即失敗。我們建議將此值變更為資源密集型任務，方法是將值減少為小於預設值，以允許每個任務更多的容量。