翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # Amazon Managed Workflows for Apache Airflow のベストプラクティス このガイドでは、Amazon Managed Workflows for Apache Airflow を使用する場合に推奨されるベストプラクティスについて説明します。 **Topics** + [Amazon MWAA での Apache Airflow のパフォーマンス調整](best-practices-tuning.md) + [requirements.txt での Python 依存関係の管理](best-practices-dependencies.md) # Amazon MWAA での Apache Airflow のパフォーマンス調整 このトピックでは、[Amazon MWAA での Apache Airflow 設定オプションの使用](configuring-env-variables.md) を使用して Amazon Managed Workflows for Apache Airflow 環境のパフォーマンスを調整する方法について説明します。 **Contents** + [Apache Airflow 設定オプションの追加](#best-practices-tuning-console-add) + [Apache Airflow スケジューラー](#best-practices-tuning-scheduler) + [パラメータ](#best-practices-tuning-scheduler-params) + [制限](#best-practices-tuning-scheduler-limits) + [DAG フォルダー](#best-practices-tuning-dag-folders) + [パラメータ](#best-practices-tuning-dag-folders-params) + [DAG ファイル](#best-practices-tuning-dag-files) + [パラメータ](#best-practices-tuning-dag-files-params) + [タスク](#best-practices-tuning-tasks) + [パラメータ](#best-practices-tuning-tasks-params) ## Apache Airflow 設定オプションの追加 環境に Airflow 設定オプションを追加するには、次の手順に従います。 1. Amazon MWAA コンソールで、[環境ページ](https://console.aws.amazon.com/mwaa/home#/environments) を開きます。 1. 環境を選択します。 1. **編集** を選択します。 1. **次へ** を選択します。 1. **Airflow 設定オプション** ペインで **カスタム設定を追加** を選択します。 1. ドロップダウンリストから設定を選択して値を入力するか、カスタム設定を入力して値を入力します。 1. 追加する設定ごとに **カスタム設定を追加** を選択します。 1. **保存** を選択します。 詳細については、[Amazon MWAA での Apache Airflow 設定オプションの使用](configuring-env-variables.md) を参照してください。 ## Apache Airflow スケジューラー Apache Airflow スケジューラーは Apache Airflow のコアコンポーネントです。スケジューラーに問題があると、DAG の解析やタスクのスケジュール設定ができなくなる可能性があります。Apache Airflow スケジューラーの調整の詳細については、Apache Airflow ドキュメンテーションウェブサイトの [スケジューラーのパフォーマンスのファインチューニング](https://airflow.apache.org/docs/apache-airflow/2.2.2/concepts/scheduler.html#fine-tuning-your-scheduler-performance) を参照してください。 ### パラメータ このセクションでは、Apache Airflow スケジューラー (Apache Airflow v2 以降) で使用できる設定オプションとそのユースケースについて説明します。 ------ #### [ Apache Airflow v3 ] | 設定 | ユースケース | | --- | --- | | **[celery.sync\$1parallelism](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#parallelism)** Celery エグゼキュターがタスクの状態を同期するために使用するプロセスの数です。 **デフォルト:** 1 | このオプションを使用すると、Celery エグゼキュターが使用するプロセスを制限することでキューの競合を防ぐことができます。デフォルトでは、CloudWatch Logs にタスクログを配信する際にエラーが発生しないように値が `1` に設定されています。この値を `0` に設定すると最大数のプロセスを使用することになりますが、タスクログの配信時にエラーが発生する可能性があります。 | | **[scheduler.scheduler\$1idle\$1sleep\$1time](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#scheduler-idle-sleep-time)** スケジューラーの「ループ」で DAG ファイルが連続して処理されるまでに待機する秒数。 **デフォルト:** 1 | このオプションを使用すると、DAG 解析結果の取得、タスクの検索とキューへの追加、*エグゼキューター* でのキュー内のタスクの実行が完了した後に、スケジューラーがスリープする時間を **長くする** ことで、スケジューラーの CPU 使用率を解放できます。この値を増やすと、Airflow v2 および Apache Airflow v3 の `dag_processor.parsing_processes` 環境で実行されるスケジューラーのスレッド数が消費されます。これにより、スケジューラーが DAG を解析する容量が減り、DAG がウェブサーバーに取り込まれるのにかかる時間が長くなる可能性があります。 | | **[scheduler.max\$1dagruns\$1to\$1create\$1per\$1loop](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#max-dagruns-to-create-per-loop)** スケジューラー「ループ」ごとに作成する *DagRuns* の DAG の最大数。 **デフォルト**: 10 | このオプションを使用すると、スケジューラーの「ループ」での *DAGRun* の最大数を **減らす** ことで、タスクのスケジューリングに必要なリソースを解放できます。 | | **[dag\$1processor.parsing\$1processes](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#parsing-processes)** スケジューラーが DAG をスケジュールするために並列して実行できるスレッドの数。 **デフォルト:** `(2 * number of vCPUs) - 1` を使用する | このオプションを使用すると、スケジューラーが DAG を解析するために並列して実行するプロセスの数を **減らす** ことで、リソースを解放できます。DAG の解析がタスクスケジューリングに影響している場合は、この数を低く抑えることをお勧めします。環境の vCPU 数よりも小さい値を指定する **必要があります**。詳細については、[制限](#best-practices-tuning-scheduler-limits) を参照してください。 | ------ #### [ Apache Airflow v2 ] | 設定 | ユースケース | | --- | --- | | **[celery.sync\$1parallelism](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#parallelism)** Celery エグゼキュターがタスクの状態を同期するために使用するプロセスの数です。 **デフォルト:** 1 | このオプションを使用すると、Celery エグゼキュターが使用するプロセスを制限することでキューの競合を防ぐことができます。デフォルトでは、CloudWatch Logs にタスクログを配信する際にエラーが発生しないように値が `1` に設定されています。この値を `0` に設定すると最大数のプロセスを使用することになりますが、タスクログの配信時にエラーが発生する可能性があります。 | | **[scheduler.idle\$1sleep\$1time](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#scheduler-idle-sleep-time)** スケジューラーの「ループ」で DAG ファイルが連続して処理されるまでに待機する秒数。 **デフォルト:** 1 | このオプションを使用すると、DAG 解析結果の取得、タスクの検索とキューへの追加、*エグゼキューター* でのキュー内のタスクの実行が完了した後に、スケジューラーがスリープする時間を **長くする** ことで、スケジューラーの CPU 使用率を解放できます。この値を増やすと、Airflow v2 および Apache Airflow v3 の `scheduler.parsing_processes` 環境で実行されるスケジューラーのスレッド数が消費されます。これにより、スケジューラーが DAG を解析する容量が減り、DAG がウェブサーバーに取り込まれるのにかかる時間が長くなる可能性があります。 | | **[scheduler.max\$1dagruns\$1to\$1create\$1per\$1loop](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#max-dagruns-to-create-per-loop)** スケジューラー「ループ」ごとに作成する *DagRuns* の DAG の最大数。 **デフォルト**: 10 | このオプションを使用すると、スケジューラーの「ループ」での *DAGRun* の最大数を **減らす** ことで、タスクのスケジューリングに必要なリソースを解放できます。 | | **[scheduler.parsing\$1processes](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#parsing-processes)** スケジューラーが DAG をスケジュールするために並列して実行できるスレッドの数。 **デフォルト:** `(2 * number of vCPUs) - 1` を使用する | このオプションを使用すると、スケジューラーが DAG を解析するために並列して実行するプロセスの数を **減らす** ことで、リソースを解放できます。DAG の解析がタスクスケジューリングに影響している場合は、この数を低く抑えることをお勧めします。環境の vCPU 数よりも小さい値を指定する **必要があります**。詳細については、[制限](#best-practices-tuning-scheduler-limits) を参照してください。 | ------ ### 制限 このセクションでは、スケジューラーのデフォルトパラメータを調整する際に考慮する制限について説明します。 **scheduler.parsing\$1processes、scheduler.max\$1threads (v2のみ)** 1 つの環境クラスの vCPU ごとに 2 つのスレッドを使用できます。環境クラスのスケジューラー用に少なくとも 1 つのスレッドを予約する必要があります。タスクのスケジュールが遅れていることに気付いた場合は、[環境クラス](environment-class.md) を増やす必要があるかもしれません。例えば、大規模な環境では、スケジューラ用に 4 vCPU の Fargate コンテナインスタンスがあります。つまり、他のプロセスに使用できるスレッドの `7` 合計は最大数になります。つまり、2 つのスレッドに 4 つの vCPUs を乗算した値から、スケジューラー自体の 1 を引いた値です。`scheduler.max_threads` (v2のみ) および `scheduler.parsing_processes` で指定する値は、環境クラスで使用できるスレッド数 (以下を参照) を超えてはなりません。 + **mw1.small** — 他のプロセスの `1` スレッド数を超えてはいけません。残りのスレッドはスケジューラー用に予約されています。 + **mw1.medium** — 他のプロセスの `3` スレッド数を超えてはいけません。残りのスレッドはスケジューラー用に予約されています。 + **mw1.large** — 他のプロセスの `7` スレッド数を超えてはいけません。残りのスレッドはスケジューラー用に予約されています。 ## DAG フォルダー Apache Airflow スケジューラーは、環境内の DAG フォルダーを継続的にスキャンします。含まれている `plugins.zip` ファイル、または「airflow」インポートステートメントを含む Python (`.py`) ファイル。生成されたすべての Python DAG オブジェクトは *DagBag* に格納され、そのファイルがスケジューラーによって処理され、スケジュールする必要があるタスク (ある場合) が決定されます。DAG ファイルの解析は、そのファイルに実行可能な DAG オブジェクトが含まれているかどうかに関係なく行われます。 ### パラメータ このセクションでは、DAG フォルダー (Apache Airflow v2 以降) で使用できる設定オプションとそのユースケースについて説明します。 ------ #### [ Apache Airflow v3 ] | 設定 | ユースケース | | --- | --- | | **[dag\$1processor.refresh\$1interval](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#config-dag-processor-refresh-interval)** DAG フォルダーをスキャンして新しいファイルを探す必要のある秒数。 **デフォルト:** 300 秒 | このオプションを使用すると、DAGs フォルダーを解析する秒数を **増やす** ことでリソースを解放できます。DAG フォルダーに大量のファイルがあることが原因で、`total_parse_time metrics` で解析時間が長くなる場合は、この値を増やすことをお勧めします。 | | **[dag\$1processor.min\$1file\$1process\$1interval](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#min-file-process-interval)** スケジューラーが DAG を解析して DAG への更新が反映されるまでの秒数。 **デフォルト:** 30 秒 | このオプションを使用して、DAG を解析する前にスケジューラーが待機する秒数を **増やす** ことで、リソースを解放できます。例えば、`30` の値を指定すると、DAG ファイルは 30 秒ごとに解析されます。お使いの環境の CPU 使用率を下げるには、この数値を高くしておくことをお勧めします。 | ------ #### [ Apache Airflow v2 ] | 設定 | ユースケース | | --- | --- | | **[scheduler.dag\$1dir\$1list\$1interval](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#dag-dir-list-interval)** DAG フォルダーをスキャンして新しいファイルを探す必要のある秒数。 **デフォルト:** 300 秒 | このオプションを使用すると、DAGs フォルダーを解析する秒数を **増やす** ことでリソースを解放できます。DAG フォルダーに大量のファイルがあることが原因で、`total_parse_time metrics` で解析時間が長くなる場合は、この値を増やすことをお勧めします。 | | **[scheduler.min\$1file\$1process\$1interval](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#min-file-process-interval)** スケジューラーが DAG を解析して DAG への更新が反映されるまでの秒数。 **デフォルト:** 30 秒 | このオプションを使用して、DAG を解析する前にスケジューラーが待機する秒数を **増やす** ことで、リソースを解放できます。例えば、`30` の値を指定すると、DAG ファイルは 30 秒ごとに解析されます。お使いの環境の CPU 使用率を下げるには、この数値を高くしておくことをお勧めします。 | ------ ## DAG ファイル Apache Airflow スケジューラーループの一部として、個々の DAG ファイルが解析されて DAG Python オブジェクトが抽出されます。Apache Airflow v2 以降では、スケジューラーは同時に最大数の [解析プロセス](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#parsing-processes) を解析します。同じファイルが再び解析される前に、`scheduler.min_file_process_interval` (v2) または `dag_processor.min_file_process_interval` (v3) で指定された秒数が経過する必要があります。 ### パラメータ このセクションでは、Apache Airflow DAG ファイル (Apache Airflow v2 以降) で使用できる設定オプションとそのユースケースについて説明します。 ------ #### [ Apache Airflow v3 ] | 設定 | ユースケース | | --- | --- | | **[dag\$1processor.dag\$1file\$1processor\$1timeout](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#dag-file-processor-timeout)** *DagFileProcessor* が DAG ファイルの処理をタイムアウトするまでの秒数。 **デフォルト:** 50 秒 | このオプションを使用すると、*DAGFileProcessor* がタイムアウトするまでの時間を **増やす** ことでリソースを解放できます。DAG 処理ログにタイムアウトが発生し、実行可能な DAG が読み込まれなくなる場合は、この値を増やすことをお勧めします。 | | **[core.dagbag\$1import\$1timeout](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#dagbag-import-timeout)** Python ファイルをインポートするまでの秒数がタイムアウトします。 **デフォルト:** 30 秒 | このオプションを使用すると、Python ファイルをインポートして DAG オブジェクトを抽出する際にスケジューラーがタイムアウトになるまでにかかる時間を **増やす** ことで、リソースを解放できます。このオプションはスケジューラーの「ループ」の一部として処理され、`dag_processor.dag_file_processor_timeout` で指定されている値よりも小さい値が含まれている必要があります。 | | **[core.min\$1serialized\$1dag\$1update\$1interval](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#min-serialized-dag-update-interval)** データベース内のシリアル化された DAG が更新されるまでの最小秒数。 **デフォルト:** 30 | このオプションを使用すると、データベース内のシリアル化された DAG が更新されるまでの秒数を **増やす** ことで、リソースを解放できます。DAG の数が多い場合、または複雑な DAG がある場合は、この値を増やすことをおすすめします。この値を増やすと、DAG がシリアル化されるときのスケジューラーとデータベースへの負荷が軽減されます。 | | **[core.min\$1serialized\$1dag\$1fetch\$1interval](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#min-serialized-dag-fetch-interval)** シリアル化された DAG が DAGBag に既に読み込まれているときに、データベースから再フェッチされる秒数。 **デフォルト**: 10 | このオプションを使用すると、シリアル化された DAG を再フェッチする秒数を **増やす** ことでリソースを解放できます。データベースの「書き込み」速度を下げるには、この値を `core.min_serialized_dag_update_interval` で指定した値より大きくする必要があります。この値を増やすと、DAG がシリアル化される際のウェブサーバーとデータベースの負荷が軽減されます。 | ------ #### [ Apache Airflow v2 ] | 設定 | ユースケース | | --- | --- | | **[core.dag\$1file\$1processor\$1timeout](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#dag-file-processor-timeout)** *DagFileProcessor* が DAG ファイルの処理をタイムアウトするまでの秒数。 **デフォルト:** 50 秒 | このオプションを使用すると、*DAGFileProcessor* がタイムアウトするまでの時間を **増やす** ことでリソースを解放できます。DAG 処理ログにタイムアウトが発生し、実行可能な DAG が読み込まれなくなる場合は、この値を増やすことをお勧めします。 | | **[core.dagbag\$1import\$1timeout](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#dagbag-import-timeout)** Python ファイルをインポートするまでの秒数がタイムアウトします。 **デフォルト:** 30 秒 | このオプションを使用すると、Python ファイルをインポートして DAG オブジェクトを抽出する際にスケジューラーがタイムアウトになるまでにかかる時間を **増やす** ことで、リソースを解放できます。このオプションはスケジューラーの「ループ」の一部として処理され、`core.dag_file_processor_timeout` で指定されている値よりも小さい値が含まれている必要があります。 | | **[core.min\$1serialized\$1dag\$1update\$1interval](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#min-serialized-dag-update-interval)** データベース内のシリアル化された DAG が更新されるまでの最小秒数。 **デフォルト:** 30 | このオプションを使用すると、データベース内のシリアル化された DAG が更新されるまでの秒数を **増やす** ことで、リソースを解放できます。DAG の数が多い場合、または複雑な DAG がある場合は、この値を増やすことをおすすめします。この値を増やすと、DAG がシリアル化されるときのスケジューラーとデータベースへの負荷が軽減されます。 | | **[core.min\$1serialized\$1dag\$1fetch\$1interval](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#min-serialized-dag-fetch-interval)** シリアル化された DAG が DAGBag に既に読み込まれているときに、データベースから再フェッチされる秒数。 **デフォルト**: 10 | このオプションを使用すると、シリアル化された DAG を再フェッチする秒数を **増やす** ことでリソースを解放できます。データベースの「書き込み」速度を下げるには、この値を `core.min_serialized_dag_update_interval` で指定した値より大きくする必要があります。この値を増やすと、DAG がシリアル化される際のウェブサーバーとデータベースの負荷が軽減されます。 | ------ ## タスク Apache Airflow スケジューラーとワーカーはどちらもタスクのキューイングとデキューに関与します。スケジューラーは、解析済みのスケジュール設定が完了したタスクを **なし** ステータスから **スケジュール済み** ステータスに移行します。同じく Fargate のスケジューラーコンテナーで実行されているエグゼキューターは、それらのタスクをキューに入れ、そのステータスを **キューで待機中** に設定します。ワーカーにキャパシティがあると、キューからタスクを取り出してステータスを **実行中** に設定し、その後、タスクが成功したか失敗したかに基づいてステータスを **成功** または **失敗** に変更します。 ### パラメータ このセクションでは、Apache Airflow タスクで使用できる設定オプションとそのユースケースについて説明します。 Amazon MWAA がオーバーライドするデフォルトの設定オプションは *赤* でマークされています。 ------ #### [ Apache Airflow v3 ] | 設定 | ユースケース | | --- | --- | | **[core.parallelism](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#parallelism)** `Running` のステータスを持つことができるタスクインスタンスの最大数。 **デフォルト:** `(maxWorkers * maxCeleryWorkers) / schedulers * 1.5` に基づいて動的に設定されます。 | このオプションを使用すると、同時に実行できるタスクインスタンスの数を **増やす** ことでリソースを解放できます。指定する値は、使用可能なワーカーの数にワーカーのタスク密度を掛けた値である必要があります。この値を変更するのは、多数のタスクが「実行中」または「キューで待機中」の状態で停止している場合のみにすることをおすすめします。 | | **[core.execute\$1tasks\$1new\$1python\$1interpreter](https://airflow.apache.org/docs/apache-airflow/3.0.6/configurations-ref.html#execute-tasks-new-python-interpreter)** Apache Airflow が親プロセスをフォークするか、新しい Python プロセスを作成してタスクを実行するかを決定します。 **デフォルト:** `True` | `True` に設定すると、Apache Airflow はプラグインに加えた変更を、タスクを実行するために作成された新しい Python プロセスとして認識します。 | | **[celery.worker\$1concurrency](https://airflow.apache.org/docs/apache-airflow-providers-celery/stable/configurations-ref.html#worker-concurrency)** Amazon MWAA は、このオプションの Airflow ベースインストールをオーバーライドして、自動スケーリングコンポーネントの一部としてワーカーをスケーリングします。 **デフォルト:** 該当なし | *このオプションに指定された値はすべて無視されます。* | | **[celery.worker\$1autoscale](https://airflow.apache.org/docs/apache-airflow-providers-celery/stable/configurations-ref.html#worker-autoscale)** ワーカー向けのタスク同時実行性。 **デフォルト:** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/ja_jp/mwaa/latest/userguide/best-practices-tuning.html) | このオプションを使用すると、ワーカーの `minimum`、`maximum` のタスク同時実行を **減らす** ことでリソースを解放できます。ワーカーは、そのための十分なリソースがあるかどうかに関係なく、構成された `maximum` までの同時タスク受け入れます。十分なリソースがない状態でタスクをスケジュールすると、タスクはすぐに失敗します。リソースを大量に消費するタスクの場合は、タスクあたりの容量を増やすために値をデフォルトより小さくしてこの値を変更することをお勧めします。 | ------ #### [ Apache Airflow v2 ] | 設定 | ユースケース | | --- | --- | | **[core.parallelism](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#parallelism)** `Running` のステータスを持つことができるタスクインスタンスの最大数。 **デフォルト:** `(maxWorkers * maxCeleryWorkers) / schedulers * 1.5` に基づいて動的に設定されます。 | このオプションを使用すると、同時に実行できるタスクインスタンスの数を **増やす** ことでリソースを解放できます。指定する値は、使用可能なワーカーの数にワーカーのタスク密度を掛けた値である必要があります。この値を変更するのは、多数のタスクが「実行中」または「キューで待機中」の状態で停止している場合のみにすることをおすすめします。 | | **[core.dag\$1concurrency](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#dag-concurrency)** 各 DAG で同時に実行できるタスクインスタンスの数。 **デフォルト:** 10000 | このオプションを使用すると、同時に実行できるタスクインスタンスの数を **増やす** ことでリソースを解放できます。例えば、10 個の並列タスクを含む 100 個の DAG があり、すべての DAG を同時に実行したい場合、利用可能なワーカーの数に `celery.worker_concurrency` でのワーカータスクの密度を掛け、それを DAG の数で割って最大並列処理を計算できます。 | | **[core.execute\$1tasks\$1new\$1python\$1interpreter](https://airflow.apache.org/docs/apache-airflow/2.10.3/configurations-ref.html#execute-tasks-new-python-interpreter)** Apache Airflow が親プロセスをフォークするか、新しい Python プロセスを作成してタスクを実行するかを決定します。 **デフォルト:** `True` | `True` に設定すると、Apache Airflow はプラグインに加えた変更を、タスクを実行するために作成された新しい Python プロセスとして認識します。 | | **[celery.worker\$1concurrency](https://airflow.apache.org/docs/apache-airflow-providers-celery/stable/configurations-ref.html#worker-concurrency)** Amazon MWAA は、このオプションの Airflow ベースインストールをオーバーライドして、自動スケーリングコンポーネントの一部としてワーカーをスケーリングします。 **デフォルト:** 該当なし | *このオプションに指定された値はすべて無視されます。* | | **[celery.worker\$1autoscale](https://airflow.apache.org/docs/apache-airflow-providers-celery/stable/configurations-ref.html#worker-autoscale)** ワーカー向けのタスク同時実行性。 **デフォルト:** [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/ja_jp/mwaa/latest/userguide/best-practices-tuning.html) | このオプションを使用すると、ワーカーの `minimum`、`maximum` のタスク同時実行を **減らす** ことでリソースを解放できます。ワーカーは、そのための十分なリソースがあるかどうかに関係なく、構成された `maximum` までの同時タスク受け入れます。十分なリソースがない状態でタスクをスケジュールすると、タスクはすぐに失敗します。リソースを大量に消費するタスクの場合は、タスクあたりの容量を増やすために値をデフォルトより小さくしてこの値を変更することをお勧めします。 | ------ # requirements.txt での Python 依存関係の管理 このトピックでは、Amazon Managed Workflows for Apache Airflow 環境の `requirements.txt` ファイルに Python の依存関係をインストールして管理する方法について説明します。 **Contents** + [Amazon MWAA CLI ユーティリティを使用した DAG のテスト](#best-practices-dependencies-cli-utility) + [PyPI.org 要件ファイルフォーマットを使用した Python 依存関係のインストール](#best-practices-dependencies-different-ways) + [オプション 1: Python Package インデックスからの Python 依存関係](#best-practices-dependencies-pip-extras) + [オプション 2: Python wheel (.whl)](#best-practices-dependencies-python-wheels) + [Amazon S3 バケット上の `plugins.zip` ファイルを使用する](#best-practices-dependencies-python-wheels-s3) + [URL にホストされている WHL ファイルを使用する](#best-practices-dependencies-python-wheels-url) + [DAG から WHL ファイルを作成](#best-practices-dependencies-python-wheels-dag) + [オプション3: PyPI/PEP-503 準拠のプライベートリポジトリでホストされている Python の依存関係](#best-practices-dependencies-custom-auth-url) + [Amazon MWAA コンソールでログを有効にします。](#best-practices-dependencies-troubleshooting-enable) + [CloudWatch Logs コンソールでログにアクセスする](#best-practices-dependencies-troubleshooting-view) + [Apache Airflow UI でエラーにアクセスする](#best-practices-dependencies-troubleshooting-aa) + [Apache Airflow へのログイン](#airflow-access-and-login) + [`requirements.txt` シナリオ例](#best-practices-dependencies-ex-mix-match) ## Amazon MWAA CLI ユーティリティを使用した DAG のテスト + コマンドラインインターフェイス (CLI) ユーティリティは、Amazon Managed Workflows for Apache Airflow 環境をローカルに複製します。 + CLI は、Amazon MWAA のプロダクションイメージに似た Docker コンテナイメージをローカルでビルドします。これを使用すると、Amazon MWAA にデプロイする前に、ローカルの Apache Airflow 環境を実行して DAG、カスタムプラグイン、依存関係を開発およびテストできます。 + CLI を実行するには、GitHub の [aws-mwaa-docker-images](https://github.com/aws/amazon-mwaa-docker-images) を参照してください。 ## PyPI.org 要件ファイルフォーマットを使用した Python 依存関係のインストール 次のセクションでは、PyPI.org [要件ファイル形式](https://pip.pypa.io/en/stable/reference/pip_install/#requirements-file-format) に従って Python 依存関係をインストールするさまざまな方法について説明します。 ### オプション 1: Python Package インデックスからの Python 依存関係 次のセクションでは、`requirements.txt` ファイルの [Python Package インデックス](https://pypi.org/) から Python 依存関係を指定する方法について説明します。 ------ #### [ Apache Airflow v3 ] 1. **ローカルでテストします**。`requirements.txt` ファイルを作成する前に、ライブラリを繰り返し追加してパッケージとバージョンの適切な組み合わせを見つけてください。Amazon MWAA CLI ユーティリティを実行するには、GitHub の [aws-mwaa-docker-images](https://github.com/aws/amazon-mwaa-docker-images) を参照してください。 1. **Apache Airflow パッケージのエクストラを確認してください**。Amazon MWAA で Apache Airflow v3 にインストールされているパッケージのリストにアクセスするには、GitHub ウェブサイトの [aws-mwaa-docker-images`requirements.txt`](https://github.com/aws/amazon-mwaa-docker-images/blob/main/requirements.txt) を参照してください。 1. **制約ステートメントを追加します**。ファイルの上部に Apache Airflow v3 環境の制約`requirements.txt`ファイルを追加します。Apache Airflow の制約ファイルには、Apache Airflow のリリース時点で利用可能なプロバイダーのバージョンが指定されています。 次の例では、*\$1environment-version\$1* をお使いの環境のバージョン番号に、*\$1Python-version\$1* を環境と互換性のある Python のバージョンに置き換えます。 Apache Airflow 環境と互換性のある Python のバージョンについては、[Apache Airflow のバージョン](airflow-versions.md#airflow-versions-official) を参照してください。 ``` --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-{Airflow-version}/constraints-{Python-version}.txt" ``` 制約ファイルが `xyz==1.0` パッケージが環境内の他のパッケージと互換性がないと判断した場合、`pip3 install` は環境に互換性のないライブラリがインストールされるのを防ぐために失敗しま す。いずれかのパッケージのインストールが失敗した場合、CloudWatch Log の対応するログストリームで、各 Apache Airflow コンポーネント (スケジューラ、ワーカー、ウェブサーバー) のエラーログにアクセスできます。ログタイプの詳細については、[Amazon CloudWatch の Airflow ログへのアクセス](monitoring-airflow.md) を参照してください。 1. Apache Airflow パッケージ。[パッケージエクストラ](http://airflow.apache.org/docs/apache-airflow/2.5.1/extra-packages-ref.html) とバージョン (`==`) を追加します。これにより、同じ名前で異なるバージョンのパッケージが環境にインストールされるのを防ぐことができます。 ``` apache-airflow[package-extra]==2.5.1 ``` 1. Python ライブラリ パッケージ名とバージョン (`==`) を `requirements.txt` ファイルに追加します。これにより、[PyPI.org](https://pypi.org) からの将来の重大な更新が自動的に適用されるのを防ぐことができます。 ``` library == version ``` **Example Boto3 と psycopg2-binary** この例は、デモンストレーションのみを目的としています。boto と psycopg2 のバイナリライブラリは Apache Airflow v3 のベースインストールに含まれており、`requirements.txt` ファイルで指定する必要はありません。 ``` boto3==1.17.54 boto==2.49.0 botocore==1.20.54 psycopg2-binary==2.8.6 ``` パッケージがバージョンなしで指定されている場合、Amazon MWAA は [PyPI.org](https://pypi.org) からパッケージの最新バージョンをインストールします。このバージョンは、お客様の `requirements.txt` 内の他のパッケージと競合できます。 ------ #### [ Apache Airflow v2 ] 1. **ローカルでテストします**。`requirements.txt` ファイルを作成する前に、ライブラリを繰り返し追加してパッケージとバージョンの適切な組み合わせを見つけてください。Amazon MWAA CLI ユーティリティを実行するには、GitHub の [aws-mwaa-docker-images](https://github.com/aws/amazon-mwaa-docker-images) を参照してください。 1. **Apache Airflow パッケージのエクストラを確認してください**。Amazon MWAA で Apache Airflow v2 用にインストールされているパッケージのリストにアクセスするには、GitHub ウェブサイトの [aws-mwaa-docker-images `requirements.txt`](https://github.com/aws/amazon-mwaa-docker-images/blob/main/requirements.txt) にアクセスします。 1. **制約ステートメントを追加します**。Apache Airflow v2 環境用の制約ファイルを `requirements.txt` ファイルの先頭に追加します。Apache Airflow の制約ファイルには、Apache Airflow のリリース時点で利用可能なプロバイダーのバージョンが指定されています。 Apache Airflow v2.7.2 から、要件ファイルには `--constraint` ステートメントを含める必要があります。制約を指定しない場合、要件に記載されているパッケージが使用している Apache Airflow のバージョンと互換性があることを確認するため、Amazon MWAA はお客様に代わって制約を指定します。 次の例では、*\$1environment-version\$1* をお使いの環境のバージョン番号に、*\$1Python-version\$1* を環境と互換性のある Python のバージョンに置き換えます。 Apache Airflow 環境と互換性のある Python のバージョンについては、[Apache Airflow のバージョン](airflow-versions.md#airflow-versions-official) を参照してください。 ``` --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-{Airflow-version}/constraints-{Python-version}.txt" ``` 制約ファイルが `xyz==1.0` パッケージが環境内の他のパッケージと互換性がないと判断した場合、`pip3 install` は環境に互換性のないライブラリがインストールされるのを防ぐために失敗しま す。いずれかのパッケージのインストールが失敗した場合、CloudWatch Log の対応するログストリームで、各 Apache Airflow コンポーネント (スケジューラ、ワーカー、ウェブサーバー) のエラーログにアクセスできます。ログタイプの詳細については、[Amazon CloudWatch の Airflow ログへのアクセス](monitoring-airflow.md) を参照してください。 1. Apache Airflow パッケージ。[パッケージエクストラ](http://airflow.apache.org/docs/apache-airflow/2.5.1/extra-packages-ref.html) とバージョン (`==`) を追加します。これにより、同じ名前で異なるバージョンのパッケージが環境にインストールされるのを防ぐことができます。 ``` apache-airflow[package-extra]==2.5.1 ``` 1. Python ライブラリ パッケージ名とバージョン (`==`) を `requirements.txt` ファイルに追加します。これにより、[PyPI.org](https://pypi.org) からの将来の重大な更新が自動的に適用されるのを防ぐことができます。 ``` library == version ``` **Example Boto3 と psycopg2-binary** この例は、デモンストレーションのみを目的としています。boto と psycopg2 のバイナリライブラリは Apache Airflow v2 のベースインストールに含まれており、`requirements.txt` ファイルで指定する必要はありません。 ``` boto3==1.17.54 boto==2.49.0 botocore==1.20.54 psycopg2-binary==2.8.6 ``` パッケージがバージョンなしで指定されている場合、Amazon MWAA は [PyPI.org](https://pypi.org) からパッケージの最新バージョンをインストールします。このバージョンは、お客様の `requirements.txt` 内の他のパッケージと競合できます。 ------ ### オプション 2: Python wheel (.whl) Python wheel は、コンパイルされたアーティファクトをライブラリに同梱するために設計されたパッケージ形式です。Amazon MWAA に依存関係をインストールする方法として、ホイールパッケージにはいくつかの利点があります。 + **より速いインストール** — WHL ファイルは 1 つの ZIP としてコンテナにコピーされ、ローカルにインストールされます。各ファイルをダウンロードする必要はありません。 + **より少ないコンフリクト** — パッケージのバージョン互換性を事前に判断できます。その結果、`pip` が互換性のあるバージョンを再帰的に調べる必要がなくなります。 + **耐障害性の向上** — 外部でホストされているライブラリでは、ダウンストリームの要件が変更され、Amazon MWAA 環境上のコンテナ間でバージョン互換性がなくなる可能性があります。依存関係を外部ソースに依存しないことで、各コンテナがいつインスタンス化されたかに関係なく、上のすべてのコンテナに同じライブラリが割り当てられます。 `requirements.txt` の Python wheel アーカイブ (`.whl`) から Python 依存関係をインストールするには、次の方法をお勧めします。 **Topics** + [Amazon S3 バケット上の `plugins.zip` ファイルを使用する](#best-practices-dependencies-python-wheels-s3) + [URL にホストされている WHL ファイルを使用する](#best-practices-dependencies-python-wheels-url) + [DAG から WHL ファイルを作成](#best-practices-dependencies-python-wheels-dag) #### Amazon S3 バケット上の `plugins.zip` ファイルを使用する Apache Airflow スケジューラ、ワーカー、およびウェブサーバー (Apache Airflow v2.2.2 以降) は、 の環境の AWSマネージド Fargate コンテナで起動中にカスタムプラグインを検索します`/usr/local/airflow/plugins/*`。このプロセスは、Python 依存関係および Apache Airflow サービスの起動のための Amazon MWAA の `pip3 install -r requirements.txt` 前に始まります。`plugins.zip` ファイルは、環境実行中に継続的に変更されたくないファイル、またはDAGを作成するユーザーにアクセス権を与えたくないファイルに使用できます。例えば、Python ライブラリのホイールファイル、証明書 PEM ファイル、設定 YAML ファイルなどです。 次のセクションでは、`plugins.zip` ファイルにあるホイールを Amazon S3 バケットにインストールする方法について説明します。 1. **必要な WHL ファイルをダウンロードします**。Amazon MWAA [aws-mwaa-docker-images](https://github.com/aws/amazon-mwaa-docker-images)または別の [Amazon Linux 2](https://aws.amazon.com/amazon-linux-2) コンテナにある既存の `requirements.txt` と [https://pip.pypa.io/en/stable/cli/pip_download/](https://pip.pypa.io/en/stable/cli/pip_download/) を使用して、必要な Python wheel ファイルを解決してダウンロードできます。 ``` pip3 download -r "$AIRFLOW_HOME/dags/requirements.txt" -d "$AIRFLOW_HOME/plugins" cd "$AIRFLOW_HOME/plugins" zip "$AIRFLOW_HOME/plugins.zip" * ``` 1. **`requirements.txt` でパスを指定します。**次のコードに示すように、requirements.txt の先頭に [https://pip.pypa.io/en/stable/cli/pip_install/#install-find-links](https://pip.pypa.io/en/stable/cli/pip_install/#install-find-links) を使用してプラグインディレクトリを指定し、`pip` に他のソースからインストールしないように [https://pip.pypa.io/en/stable/cli/pip_install/#install-no-index](https://pip.pypa.io/en/stable/cli/pip_install/#install-no-index) を用いて指示します。ースからインストールしないように指示します。 ``` --find-links /usr/local/airflow/plugins --no-index ``` **Example requirements.txt 内のホイール** 次の例では、Amazon S3 バケットのルートにある `plugins.zip` ファイルにホイールをアップロードしたことを前提としています。例えば、次のようになります。 ``` --find-links /usr/local/airflow/plugins --no-index numpy ``` Amazon MWAA は `plugins` フォルダから `numpy-1.20.1-cp37-cp37m-manylinux1_x86_64.whl` ホイールを取得し、環境にインストールします。 #### URL にホストされている WHL ファイルを使用する 次のセクションでは、URL でホストされるホイールをインストールする方法について説明します。URL は、パブリックにアクセス可能であるか、Amazon MWAA 環境用に指定したカスタム Amazon VPC 内からアクセスできる必要があります。 + **URL を指定してください**。`requirements.txt` 内のホイールの URL を指定します。 **Example 公開 URL でのホイールアーカイブ** 次の例では、公開サイトからホイールをダウンロードします。 ``` --find-links https://files.pythonhosted.org/packages/ --no-index ``` Amazon MWAA は、指定した URL からホイールを取得し、お使いの環境にインストールします。 **注記** Amazon MWAA v2.2.2 以降では、要件をインストールするプライベートウェブサーバーから URL にアクセスすることはできません。 #### DAG から WHL ファイルを作成 Apache Airflow v2.2.2 以降を使用するプライベートウェブサーバーがあり、環境が外部リポジトリにアクセスできないために要件をインストールできない場合は、次の DAG を使用して既存の Amazon MWAA 要件を取得し、Amazon S3 にパッケージ化できます。 ``` from airflow import DAG from airflow.operators.bash_operator import BashOperator from airflow.utils.dates import days_ago S3_BUCKET = 'my-s3-bucket' S3_KEY = 'backup/plugins_whl.zip' with DAG(dag_id="create_whl_file", schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag: cli_command = BashOperator( task_id="bash_command", bash_command=f"mkdir /tmp/whls;pip3 download -r /usr/local/airflow/requirements/requirements.txt -d /tmp/whls;zip -j /tmp/plugins.zip /tmp/whls/*;aws s3 cp /tmp/plugins.zip s3://amzn-s3-demo-bucket/{S3_KEY}" ) ``` DAG を実行したら、この新しいファイルを Amazon MWAA `plugins.zip` として使用します。オプションで、他のプラグインと一緒にパッケージ化することもできます。次に、`--constraint` を追加せずに `--find-links /usr/local/airflow/plugins` および `--no-index` の前に `requirements.txt` を更新してください。 この方法では、同じライブラリをオフラインで使用できます。 ### オプション3: PyPI/PEP-503 準拠のプライベートリポジトリでホストされている Python の依存関係 次のセクションでは、認証付きのプライベート URL でホストされている Apache Airflow エクストラをインストールする方法について説明します。 1. ユーザー名とパスワードを [Apache Airflow 設定オプション](configuring-env-variables.md) として追加します。例えば、次のようになります。 + `foo.user` : `YOUR_USER_NAME` + `foo.pass` : `YOUR_PASSWORD` 1. `requirements.txt` ファイルを作成します。次の例のプレースホルダーは、プライベート URL と [Apache Airflow 設定オプション](configuring-env-variables.md) として追加したユーザー名とパスワードに置き換えてください。例えば、次のようになります。 ``` --index-url https://${AIRFLOW__FOO__USER}:${AIRFLOW__FOO__PASS}@my.privatepypi.com ``` 1. その他のライブラリを `requirements.txt` ファイルに追加します。例えば、次のようになります。 ``` --index-url https://${AIRFLOW__FOO__USER}:${AIRFLOW__FOO__PASS}@my.privatepypi.com my-private-package==1.2.3 ``` ## Amazon MWAA コンソールでログを有効にします。 Amazon MWAA 環境の [実行ロール](mwaa-create-role.md) には、CloudWatch Logs にログを送信するためのアクセス許可が必要です。実行ロールのアクセス許可を更新するには、[Amazon MWAA 実行ロール](mwaa-create-role.md) を参照してください。 Apache Airflow ログは `INFO`、`WARNING`、`ERROR` または `CRITICAL` レベルで有効にできます。ログレベルを選択すると、Amazon MWAA はそのレベルとそれ以上の重要度レベルのすべてのログを送信します。例えば、`INFO` レベルでログを有効にすると、Amazon MWAA は `INFO` ログと `WARNING`、`ERROR`、`CRITICAL` のログレベルを CloudWatch Logs に送信します。`requirements.txt` で受信したログをスケジューラーにアクセスできるように、`INFO` レベルで Apache Airflow ログを有効にすることをお勧めします。 ![\[この画像は、INFO レベルでログを有効にする方法を示しています。\]](http://docs.aws.amazon.com/ja_jp/mwaa/latest/userguide/images/mwaa-console-logs-info.png) ## CloudWatch Logs コンソールでログにアクセスする ワークフローのスケジュール設定と `dags` フォルダーの解析を行うスケジューラーの Apache Airflow ログにアクセスできます。次のステップでは、Amazon MWAA コンソールでスケジューラーのロググループを開き、CloudWatch Logs コンソールで Apache Airflow ログにアクセスする方法について説明します。 **`requirements.txt` のログにアクセスするには** 1. Amazon MWAA コンソールで、[環境ページ](https://console.aws.amazon.com/mwaa/home#/environments) を開きます。 1. 環境を選択します。 1. **モニタリング** ペインで **Airflow スケジューラーロググループ** を選択します。 1. **ログストリーム** の `requirements_install_ip` ログを選択します。 1. 環境にインストールされたパッケージのリストについては、`/usr/local/airflow/.local/bin` を参照してください。例えば、次のようになります。 ``` Collecting appdirs==1.4.4 (from -r /usr/local/airflow/.local/bin (line 1)) Downloading https://files.pythonhosted.org/packages/3b/00/2344469e2084fb28kjdsfiuyweb47389789vxbmnbjhsdgf5463acd6cf5e3db69324/appdirs-1.4.4-py2.py3-none-any.whl Collecting astroid==2.4.2 (from -r /usr/local/airflow/.local/bin (line 2)) ``` 1. パッケージのリストを確認し、インストール中にエラーが発生したパッケージがないか確認してください。何か問題が発生した場合、以下のようなエラーが表示されることがあります。 ``` 2021-03-05T14:34:42.731-07:00 No matching distribution found for LibraryName==1.0.0 (from -r /usr/local/airflow/.local/bin (line 4)) No matching distribution found for LibraryName==1.0.0 (from -r /usr/local/airflow/.local/bin (line 4)) ``` ## Apache Airflow UI でエラーにアクセスする Apache Airflow UI をチェックして、エラーが別の問題に関連しているかどうかを確認することもできます。Amazon MWAA の Apache Airflow で発生する可能性のある最も一般的なエラーは次のとおりです。 ``` Broken DAG: No module named x ``` Apache Airflow UI にこのエラーが表示される場合は、`requirements.txt` のファイルに必要な依存関係が欠けている可能性があります。 ### Apache Airflow へのログイン Apache Airflow UI にアクセスするには、 AWS アカウント in AWS Identity and Access Management (IAM) のアクセス[Apache Airflow UI アクセスポリシー: AmazonMWAAWebServerAccess](access-policies.md#web-ui-access)許可が必要です。 **Apache Airflow UI にアクセスするには** 1. Amazon MWAA コンソールで、[環境ページ](https://console.aws.amazon.com/mwaa/home#/environments) を開きます。 1. 環境を選択します。 1. **Airflow UI を開く** を選択します。 ## `requirements.txt` シナリオ例 `requirements.txt` では異なるフォーマットを組み合わせることができます。次の例では、さまざまな方法を組み合わせてエクストラをインストールしています。 **Example PyPI.org のエクストラとパブリック URL** パブリック URL 上のパッケージ (PEP 503 準拠のカスタム repo URL など) に加えて、PyPI.org からパッケージを指定する場合は、`--index-url` のオプションを使用する必要があります。 ``` aws-batch == 0.6 phoenix-letter >= 0.3 --index-url http://dist.repoze.org/zope2/2.10/simple zopelib ```