設定パターン

網羅的ではないですが、複数あります。以下の5つを試したり紹介だけだったりしていきたいと思います。

①Airflow UIで設定(GUI)
②DAG内で設定
③airflow.cfgで設定(マネコン & CLI)
④airflow cliで設定
⑤Secrets Managerに全振り

①Airflow UIで設定

Airflow UIでの環境変数設定を見ていきます。Airflow UIでのGUI操作になります。

手順

Airflow UIを開き、上部の[Admin]にカーソルを当て、[Variables]をクリック
左下の[+]ボタンをクリックします

Key/Valueの形式で入力します。ここではfoo/barを入れました。左下の[Save]をクリックします。

foo/barの環境変数が出来ました。

エクスポート/インポートも出来ます。作成した環境変数にチェックを入れエクスポートします

エクスポートしたファイルの中身はこんな感じ

variables.json

{
    "foo": "bar"
}

環境変数を画面で削除し、エクスポートしたファイルをインポートします。

元のfoo/barが戻ります

まとめ

GUIでよければ一番手軽です。エクスポート・インポートも出来るし。GUI操作なので自動化は出来ない

②DAG内で設定

環境変数設定のDAGを作っちゃうパターン。

手順

環境変数を書いたJSONファイルを用意します。

variables.json

{
    "red": "liver",
    "white": "real"
}

DAGファイルを作成します。

setvar.py

import airflow
from airflow.models import DAG
from airflow.operators.python import PythonOperator
import os
from airflow.models import Variable
import json

args = {
    "owner": "airflow",
    "start_date": airflow.utils.dates.days_ago(1),
    "provide_context": False,
    "email_on_failure": True,
    "email_on_retry": True,
    "retries":1,
    "retry_delay":120
    }

def setvar_modules():
    print("Starting here")
    with open('/usr/local/airflow/dags/variables.json') as f:
        data = json.load(f)
    for key in data:
        print(key, '->', data[key])
        variable = Variable.set(key, data[key])
    print("Debugging here")
    for key in data:
        print(key, '->', data[key])
        variable = Variable.get(key)
with DAG(
    dag_id="set_variables_job",
    description="set variables DAG",
    default_args=args,
    schedule_interval="*/60 * * * *",
    catchup=False,
    tags=['setenv']
) as dag:
    t1 = PythonOperator(
        task_id='setvar_task',
        python_callable=setvar_modules,
        dag=dag
        )

    t1

JSONファイルとDAGファイルの2つをS3のDAGフォルダにアップします。

数分するとAirflow UIにset_variables_jobのDAGが表示されます。

DAGは最初Pause状態なので、DAG名の左のつまみをクリックしUnpauseします。そうすると自動で実行されます。
次にDAG名をクリックして詳細を確認します。

[Graph View]をクリックし、真ん中の[setvar_task]をクリック

この画面が出るので[Log]をクリック

ログの表示から環境変数が設定されてることが分かります。variable.getで確認もしてるのでsetした変数がgetできてます

Admin->Variablesで確認
設定した環境変数が表示されています

まとめ

環境変数セットアップ用のDAGを作るパターンは良いんじゃないかと思います。
環境変数セットのロジックを切り出せてるので、環境変数の設定や修正時は環境変数のJSONとDAGだけ修正すれば良くて、ビジネスロジック側に手を入れなくて済みます。
本来のDAGではないDAGが作られるのがどうかな？とは思いますが、DAGであるので逆に管理運用はしやすいと思う。

③airflow.cfgで設定(マネコン & CLI)

airflow.cfgにカスタムな環境変数を書けます。MWAAでそれを行うやり方。

GUI手順

MWAAの環境の編集で、Airflow設定オプションを追加しましす。ここで入れた値はairflow.cfgに反映されます。
入力した値は、custom.envval/helloworld です

確認のためカスタム環境変数を表示するDAGを作ってS3にアップします。
確認用のDAGなので本来は不要なDAGです。

getvar.py

import boto3
import airflow
from airflow.models import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.operators.athena import AWSAthenaOperator
from airflow.utils.task_group import TaskGroup
import os

args = {
    "owner": "airflow",
    "start_date": airflow.utils.dates.days_ago(1),
    "provide_context": False,
    "email_on_failure": True,
    "email_on_retry": True,
    "retries":1,
    "retry_delay":120
    }

def getvar_module():
    words = os.getenv("AIRFLOW__CUSTOM__ENVVAL")
    print("Debugging here")
    print (words)

with DAG(
    dag_id="get_variables_job2",
    description="get variables from airflow.cfg DAG",
    default_args=args,
    schedule_interval="*/60 * * * *",
    catchup=False,
    tags=['getenv']
) as dag:
    t1 = PythonOperator(task_id="get_variables_task", python_callable=getvar_module)

    t1

Airflow UIでDAGをunpauseして実行します。
DAG名をクリックします

[Graph View]で表示します

ログを確認します。airflow.cfgで設定したカスタム環境変数のhelloworldが表示されてます

CLI手順

今回は試しませんが、MWAAの環境作成時に"airflow-configuration-options"で環境変数を設定できそうです。

aws mwaa create-environment --airflow-configuration-options {"custom.envval":"helloworld"}

まとめ

DAGに環境変数の記述をしないという意味では一番キレイでシンプルだと思います。
ただ、Airflow UIのVariablesに表示されないため可視性は低めかもしれません。見通しがいいという意味だとVariablesに表示されるといいなと思います。でも、どのみちVariablesを直接操作しないならそこまでデメリットではないかもです
あと、airflow.cfgに書く場合は「AIRFLOW__CUSTOM__ENVVAL」みたいな書式に従う必要があります。
それと、最大のデメリットがairflow.cfgの修正にはMWAA側の反映にかなり時間が掛ります。手元では10分くらい掛りますた、公式Docには「数分かかる」とあります。よって、時々レベルの頻度で更新が入ったり、更新してすぐに試したい環境変数はなるべくairflow.cfgと切り離した方がいいと思います。

④Airflow CLI

2.0.2からできるっぽいですね。サンプルコードもありました

Airflowの公式Doc的にはこちらのコマンドです

airflow variables get [-h] [-d VAL] [-j] [-v] key

まとめ

試してないですmm
試してないのに感想ですが、CLIなのでCLI実行環境が必要なので、手頃なコンピューティングがあれば良いいかなと。
Airflowとは別な環境で管理する形になるので、実行環境といい感じに疎結合になったバッチサーバーみたいな基盤があれば良さそう
AirflowはAirflowでまとめたいなら前述の手段がいいかも。Airflowでまとめなくてもいいのだけど。

⑤Secrets Manager全振り

環境変数の保存先にSecrets Managerも使えます。本来は秘匿性が高いDB接続情報(ID/Passwordなど)を保存することに使うAWSのナイスIntegrationですが、環境変数の保存先にも使えます

※注意点は、前述のvariables.getではMWAA(Airflow)のメタストア(MWAA内部のRDS)に保存された値を取得しますが、SecretsManagerを保存先にした場合は、variables.getでSecretsManagerに保存された値を取得します。キーが被っている場合はSecretsManagerが優先されそうです。
英語ですがこちらのブログにこの辺含めて、MWAA+SecretsManagerをすごく丁寧に検証してるので是非ご確認ください。

MWAAでSecrets Managerを使う設定はこちら

サンプルコードはこちらですが、variable.getを使う点では②とほとんど同じ感じです。
variable.setはSecrets Managerへの設定操作ということになりますね。

まとめ

試してないですmm
環境変数を入れ先としてはAWSらしい疎結合で、可視性と操作性もいいと思います。Secretでない値を入れるのに若干違和感ありますが、公式ドキュメントにも環境変数の設定手順とあるので利用用途してはアリだと思います。
デメリットとしては、Secrets Manager側にも若干ですがコストがかかる点があるかもしれません。
あと、Airflowで完結したい場合は、別サービスであるSecrets Managerが登場してきてしまいます。ただ、いい感じの結合度なのでmake senseだと思います。

全体まとめ

個人的には「⑤Secrets Manager全振り」でいいと思います(試してないのに一番お勧めmm)。「①環境変数設定DAGにするパターン」もユースケースによってはマッチしそうです。DAGに慣れてれば扱いやすいしMWAAで完結できますね。ほとんど変更がない環境変数であれば「②airflow.cfgで設定」も良いと思います。

参考リンク