■ はじめに

Apache Airflow で処理の失敗時やワークフロー成功時に
以下のサービスへ通知する方法を調べたので、メモっておく。
~~~~~~~~~~
【１】Slack
【２】Email / AWS SES
【３】AWS SNS
【４】Microsoft Teams
~~~~~~~~~~

【０】全般的な注意点
　１）メッセージ内容が日本語を使用する場合
【１】Slack
　０）注意事項
　１）SlackAPIOperator
　２）SlackAPIPostOperator
　３）SlackAPIFileOperator
　４）SlackWebhookOperator
【２】Email / AWS SES
　０）注意事項
　１）EmailOperator
　２）SESHook
【３】AWS SNS
　０）注意事項
　１）SnsPublishOperator
　２）AwsSnsHook
【４】Microsoft Teams
【５】通知に関する技術事項
　１）実装せずにタスク失敗をEmail通知する
　２）タスク成功・失敗時に通知する
　３）接続情報の保持

なお、サンプルについては、以下の関連記事を参照のこと。

Apache Airflow ～通知サンプル編～
https://dk521123.hatenablog.com/entry/2021/10/09/000000

【０】全般的な注意点

１）メッセージ内容が日本語を使用する場合

* DAGのPythonコードを、
　ファイル形式「UTF-8」で保存すること

 => 日本語のメッセージでテストした際にWeb UI側で
　　エラー「SyntaxError: ... 'utf-8' codec can't decode ...」が
　　表示されたので、何かなーって原因を探ったら、ファイル形式だった
 => 詳細は、以下の関連記事を参照のこと。

Apache Airflow に関するトラブル
https://dk521123.hatenablog.com/entry/2021/10/03/000000

【１】Slack

* 関連するAPI仕様の詳細は、以下のサイトを参照。

https://airflow.apache.org/docs/apache-airflow-providers-slack/stable/_api/airflow/providers/slack/operators/slack/index.html
https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/contrib/operators/slack_webhook_operator/index.html

０）注意事項

* Slack の Access Token を用意する必要がある
 => トークンに関しては、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2021/10/13/092235

* WebHook の場合は、 Incoming Webhookを有効にする

https://dk521123.hatenablog.com/entry/2021/10/15/091842

１）SlackAPIOperator

* 以下の１）～２）の親クラス

２）SlackAPIPostOperator

* Slack チェンネルへのPost送信を実行

構文

from airflow.operators.slack_operator import SlackAPIPostOperator

slack = SlackAPIPostOperator(
    task_id="post_hello",
    dag=dag,
    token="XXX",
    text="hello there!",
    channel="#random",
)

３）SlackAPIFileOperator

* Slack チャンネルへのファイル送信を実行

構文

# Send file with filename and filetype
slack = SlackAPIPostOperator(
    task_id="post_hello",
    dag=dag,
    token="XXX",
    text="hello there!",
    channel="#random",
)

４）SlackWebhookOperator

* 関連するAPI仕様の詳細は、以下のサイトを参照。

https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/contrib/operators/slack_webhook_operator/index.html

* Webhook については、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2021/10/15/091842

* Connection を使用する（「http_conn_id='slack_webhook'」部分）場合は、
　以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2021/10/16/000454

構文

from airflow.contrib.operators.slack_webhook_operator import SlackWebhookOperator

slack_job = SlackWebhookOperator(
  task_id='slack_task',
  # Connectionで指定する（webhook_tokenでトークンで指定することも可能）
  http_conn_id='slack_webhook',
  message='*Hello, world!* \nThis is a message)',
  username='slack-user-name',
  icon_emoji=':dog:',
  channel='#your-channel-name',
  dag=dag
)

コード
http://airflow.apache.org/docs/apache-airflow/1.10.13/_modules/airflow/contrib/operators/slack_webhook_operator.html

参考文献
https://qiita.com/munaita_/items/ad0f6f9590185741fff6

【２】Email / AWS SES

* 関連するAPI仕様の詳細は、以下のサイトを参照。

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/email/index.html
https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/_api/airflow/providers/amazon/aws/hooks/ses/index.html

* なお、Amazon SES に関する詳細は、以下の関連記事を参照のこと

Amazon SES ～ Emailサービス～
https://dk521123.hatenablog.com/entry/2017/04/28/234103

０）注意事項

* AirflowにSMTPの接続情報を設定する必要がある

設定方法

[1] airflow.cfgで設定する方法
 => 詳細は、以下のサイトを参照。

https://airflow.apache.org/docs/apache-airflow/stable/howto/email-config.html

[2] 環境変数で指定する方法
 => 詳細は、以下のサイトを参照。

https://dev.classmethod.jp/articles/apache-airflow-send-mail-using-amazon-ses/

* 設定値については、以下の公式ドキュメントを参照。

設定方法 / AWS MWAA

* MWAA（Amazon Managed Workflow for Apache Airflow）での設定は
　以下のサイトを参照のこと（基本的にやることは、前述と変わらない）

https://qiita.com/pioho07/items/58c97bbfa8163f713d80

Email
https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html?highlight=email#email
SMTP (Simple Mail Transfer Protocol)
https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#smtp

１）EmailOperator

* Email送信する
* 添付ファイルも引数「files (list)」で送れそう

構文

from airflow.operators.email import EmailOperator

send_mail = EmailOperator(
  task_id="sendmail",
  to="to_address@xxxx.com",
  cc="cc_address@xxxx.com",
  bcc="bcc_address@xxxx.com",
  subject="This is a title",
  html_content="<b>Hello World</b>",
  mime_charset="utf-8",
)

２）SESHook

https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/_api/airflow/providers/amazon/aws/operators/sns/index.html

* SES(Amazon Simple Email Service) のフック関数
 => 使うことあるか分からんが、、、

参考文献
https://dev.classmethod.jp/articles/apache-airflow-send-mail-using-amazon-ses/
https://qiita.com/pioho07/items/58c97bbfa8163f713d80
https://bhavaniravi.com/blog/sending-emails-from-airflow/

【３】AWS SNS

* 関連するAPI仕様の詳細は、以下のサイトを参照。

https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/contrib/operators/sns_publish_operator/index.html
https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/contrib/hooks/aws_sns_hook/index.html

* 他のサービスと比較すると、あまり参考になるサイトがない印象。
 => 以下のサイトくらい？

https://www.mikulskibartosz.name/send-sms-from-airflow-using-aws-sns/

* なお、Amazon SNS に関する詳細は、以下の関連記事を参照のこと

Amazon SNS ～通知サービス～
https://dk521123.hatenablog.com/entry/2021/06/03/175213

０）注意事項

* 以下を行う必要がある
 [1] AWS SNSで、トピックとサブスクリプションを追加

１）SnsPublishOperator

* Amazon SNSへメッセージをPublishする

https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/contrib/operators/sns_publish_operator/index.html

構文

from airflow.contrib.operators.sns_publish_operator import SnsPublishOperator

publish_sns = SnsPublishOperator(
  task_id='publish_sns_task',
  aws_conn_id='your_sns_connection_id',
  target_arn='your_sns_topic_arn',
  subject='This is a title',
  message='Hello World'
)

コード

* 内部で「２）AwsSnsHook」を呼び出している

https://airflow.apache.org/docs/apache-airflow/1.10.12/_modules/airflow/contrib/operators/sns_publish_operator.html

補足：引数「aws_conn_id」「target_arn」

* aws_conn_id : aws connection to use
 => 使用するAWSコネクション？？

https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/connections/aws.html

 => 以下のサイトが参考になりそう。

https://dev.classmethod.jp/articles/apache-airflow-aws-conn-assume-role/

* target_arn : either a TopicArn or an EndpointArn
 => トピックARNかエンドポイントARNのどちらか
 => トピックの ARN「arn:aws:sns:<RegionName>-xxxxxxx」
　　を指定すればよさそう

２）AwsSnsHook

* SNS への フック関数

https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/contrib/hooks/aws_sns_hook/index.html

* コードを参照したところ、boto3 API - public(通知依頼) を使用している模様

コード
http://airflow.apache.org/docs/apache-airflow/1.10.6/_modules/airflow/contrib/hooks/aws_sns_hook.html
boto3 API - public(通知依頼)
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sns.html#SNS.Client.publish

【４】Microsoft Teams

* HTTP Post送信でできそう。
 => 詳細は、以下の参考文献を参照。

参考文献
https://qiita.com/whata/items/dba1c2db1a38d9d4a4c0
https://zenn.dev/antyuntyun/articles/airflow_custom_notification

【５】通知に関する技術事項

１）実装せずにタスク失敗をEmail通知する

* Email送信であれば「'email_on_failure': True」って設定がある

https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/operators/index.html

より抜粋
~~~~~~~~~~~~
email_on_failure (bool)
 – Indicates whether email alerts should be sent when a task failed
=> タスク失敗時に、Emailアラートを送るかどうかを示す
~~~~~~~~~~~~

Pythonコード・一部抜粋

default_args = {
  'email': ['xxxxx@gmail.com'],
  'email_on_failure': True,
  'email_on_retry': True,
}

使用上の注意

* 「【２】Email / AWS SES」の「０）注意事項」で記載しているのと同様に
　「AirflowにSMTPの接続情報を設定する必要がある」。
 => 何も設定せずに行うと、ログに以下のエラーが表示されてしまう。

~~~~~~~~
ConnectionRefusedError: [Errno 111] Connection refused
~~~~~~~~

https://github.com/puckel/docker-airflow/issues/425

２）タスク成功・失敗時に通知する

* コールバックが用意されているので
　失敗時：on_failure_callback
　成功時：on_success_callback

* default_argsに設定すればDAG内の全タスクに適応可能
 => 失敗時は、default_argsに設定し、
　　成功時は、最後のタスクに設定すればよさそう

実装イメージ

def send_for_failure(context):
  try:
    # 例外内容
    exception_message = str(context['exception'])
    # タスク インスタンス(※１参照)
    task_instance = context.get('task_instance')
    # Task ID
    task_id = task_instance.task_id
    # DAG ID
    task_id = task_instance.dag_id
    # Execution Date
    task_id = task_instance.execution_date

    # 以降は、送信する処理（省略）
  except Exception as ex:
    print("Error " + str(ex))
    raise ex

※１：タスクインスタンス

* タスク インスタンスの詳細は、以下の公式ドキュメントを参照。

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/taskinstance/index.html

参考文献
https://qiita.com/munaita_/items/1a5b131839e01ea7280d
https://blog.imind.jp/entry/2019/02/08/170332
https://medium.com/@momota/airflow-dag%E3%81%AB%E3%81%8A%E3%81%91%E3%82%8B%E5%90%84%E3%82%BF%E3%82%B9%E3%82%AF%E3%81%AE%E6%88%90%E5%8A%9For%E5%A4%B1%E6%95%97%E3%82%92slack%E3%81%AB%E9%80%9A%E7%9F%A5%E3%81%99%E3%82%8B-5eecda9ec378

３）接続情報の保持

* Tokenなどの接続情報の保持は、「Variable」を使用していると良さそう
 => Variableの設定方法など詳細は、以下の関連記事を参照のこと。

https://dk521123.hatenablog.com/entry/2021/09/30/163020

プログラムの超個人的なメモ

Memo for Programming.

【Airflow】Apache Airflow ～通知あれこれ編～

■ はじめに

目次

【０】全般的な注意点

１）メッセージ内容が日本語を使用する場合

【１】Slack

０）注意事項

１）SlackAPIOperator

２）SlackAPIPostOperator

３）SlackAPIFileOperator

４）SlackWebhookOperator

【２】Email / AWS SES

０）注意事項

１）EmailOperator

２）SESHook

【３】AWS SNS

０）注意事項

１）SnsPublishOperator

２）AwsSnsHook

【４】Microsoft Teams

【５】通知に関する技術事項

１）実装せずにタスク失敗をEmail通知する

２）タスク成功・失敗時に通知する

３）接続情報の保持

関連記事