Apache AirflowからAmazon SESを使ってメール送信をしてみた

こんにちは。サービスグループの武田です。

最近はApache Airflowの理解を深めるため検証作業などをしています。AirflowはDAG（有向非巡回グラフ）でワークフローを表現し、ジョブのスケジューリングや監視などを行ってくれるツールです。

手元で試したい場合は公式のQuick Startが参考になります。

またDockerを使いたい場合は山下のエントリも参考になります。

ちなみに私はDocker Composeでローカル環境を構築しました。さて今回は、ローカルで動作が確認できた後、ワークフローの中でメール送信を試してみたことの記録になります。

Airflowでのメール送信について

AirflowはPythonで書かれたソフトウェアであり、モジュールの追加もサポートしています。そのため自分で書けば基本的にどんなことでも可能です。現在の安定版である1.10.12ではSMTPサーバーまたはSendGridが簡単な設定で利用できます。

今回はAmazon SESを使ってメール送信を試してみました。SESはAPIがもちろんあるのですが、Airflowは現状サポートしていません。そのためSMTPエンドポイントを指定して、普通にSMTPサーバーとしてSESを使用します。

SESの初期設定

メール送信をするために必要な設定は石川のエントリなどを参考にしてください。

なお今回はサンドボックスの解除はしていません。自分のアドレスのみ承諾設定をしました。またリージョンは東京を使いました。

SESの認証情報の取得

さて、前述したようにSESでメール送信をするわけですが、今回は単にSESをSMTPサーバーとして利用します。SESのSMTPエンドポイントを利用するためにはユーザー名／パスワードなどのクレデンシャルやエンドポイントを指定する必要があります。マネジメントコンソールでSESのページにアクセスし、[SMTP Settings]のメニューから確認できます。

Server Nameがエンドポイントになります。リージョンによって異なるので注意です。また接続するための認証情報は[Create My SMTP Credentials]ボタンから作成できます。新しくIAMユーザーが作成され、認証情報がダウンロードできます。

AirflowにSMTPの接続情報を設定する

次にSMTPの接続情報を設定します。Airflowの設定方法はふたとおり提供されており、1個は設定ファイルであるairflow.cfgで設定する方法。もう1個は環境変数で指定する方法です。どちらでも同じですが、私の手元の環境ではDocker Composeを使っていて、envファイルで指定するのが簡単だったため環境変数で指定しました。

次のようなenvファイルを用意します（USERとPASSWORDは取得したクレデンシャルを指定します）。

AIRFLOW__SMTP__SMTP_HOST=email-smtp.ap-northeast-1.amazonaws.com
AIRFLOW__SMTP__SMTP_MAIL_FROM=no-reply@example.com
AIRFLOW__SMTP__SMTP_PASSWORD=ABCDEF/qwertyuiopasdfghjklzxcvbnmQWERTYUIOPA
AIRFLOW__SMTP__SMTP_PORT=587
AIRFLOW__SMTP__SMTP_SSL=False
AIRFLOW__SMTP__SMTP_STARTTLS=True
AIRFLOW__SMTP__SMTP_USER=ABCDEFGHIJKLMNOPQRST

Airflowの環境変数はAIRFLOW__${section}__${key}という形式で指定します。たとえばSMTPのホストは設定ファイルでは次のように指定します。

[smtp]
smtp_host = email-smtp.ap-northeast-1.amazonaws.com

これを環境変数で指定するので、AIRFLOW__SMTP__SMTP_HOSTという変数名になるわけです。

あとはcomposeファイルでこのenvファイルを指定すればOKです。

services:
  airflow:
    env_file:
      - mail_ses.env
  （略）

メール送信テスト

これで準備ができましたので、あとは簡単なDAGを登録して実際にメールが送信されるか試してみましょう。今回は次のようなファイルを用意しました（宛先であるtoは自分のアドレスを指定してください）。

import airflow
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.email_operator import EmailOperator

args = {
    "owner": "airflow",
    "start_date": airflow.utils.dates.days_ago(2),
    "provide_context": True,
}

with DAG(dag_id="test_ses_sendmail", default_args=args, schedule_interval=None) as dag:

    start = DummyOperator(task_id="start")

    sendmail = EmailOperator(
        task_id="sendmail",
        to="takeda@example.com",
        subject="SESでメール送信",
        html_content="うまくいくかな？",
        mime_charset="utf-8",
    )

    end = DummyOperator(task_id="end")

    start >> sendmail >> end

管理画面から確認できたら手動でDAGを起動します。

少し待てばタスクが完了します。次のようにメールを受信していました！

もしタスクが失敗していた場合は設定不備などが考えられるため、ログを確認して解消しましょう。

まとめ

今回はAmazon SESを使用してAirflowからメール送信する手順を確認しました。特に難しい作業などもなく受信まで確認できました。本番運用のためにはサンドボックスの解除申請やバウンスメールへの対応なども必要になりますが、まずは検証からはじめてみてはいかがでしょうか。