BlogiMind開発者ブログ2019-02-08Airflowでタスク失敗時にSlackへメッセージを送るairflow概要Airflowのタスクが失敗した際にSlackにメッセージを送るようにする。トークン等はVariablesに保存して扱う。3.6.7apache-airflow==1.10.1slackclient==1.3.0導入slackclientが必要になるので入れておく。下記あたりを参考にSlackのトークンを取得しておく。Variablesを選択。CreateタブからVariableを生成する。公式ドキュメントによるとVariableのキー名がパスワードっぽいものについては値がマスクされるらしい。‘access_token’のいずれかが文字列に含まれていれば対象になる。Valがマスクされた状態で登録されている。SlackAPIPostOperatorでメッセージ送信SlackAPIPostOperatorを利用してPythonOperatorのタスク失敗時にexceptionのメッセージをSlackに通知してみる。dag=dag)これで準備OK。runで上で定義したhelloタスクを叩くと、Slackに例外で設定したメッセージ(ここではテスト)が送信される。判定して例外のタイプに応じて動作を変えることもできる。BashOperatorの場合BashOperatorの場合の挙動も見てみる。slack_sendのコードはそのままに、PythonOperatorを利用していたhelloタスクをBashOperatorに切り替える。DAGの名前もslack_example_bashと変えておく。failed」というエラメッセージがSlackに飛んだ。中身を見てみるとAirflowExceptionという例外が入っていた。cross-communication)を使ってBashOperatorからメッセージを受け取る。BashOperatorではXComは下記のような動きになると書かれている。completes.https://airflow.incubator.apache.org/code.html?highlight=bashoperator#airflow.operators.BashOperator訳すと「xcom_pushをTrueに設定すると、stdoutに最後に書かれた内容がbash終了時にXComにpushされるよ」ということらしい。下記のように記述すると最後のechoに出力した値を取得できる。(以下略)TriggerRuleでエラーを送るケースを設定するすべてのタスクにon_failure_callbackを設定するとタスクごとにエラーが投げられることになる。TriggerRuleを利用して最後に1回だけエラーが投げられるように設定してみる。TriggerRuleには下記のような指定が用意されている。ALL_SUCCESSALL_FAILEDALL_DONEONE_SUCCESSONE_FAILEDDUMMYNONE_FAILED下記はop1(失敗する)、op2(成功する)、slackOp(slackに投げる)という3つのOperatorを用意している。op2の双方をParentに持つようにしておく。これで双方がエラーになればALL_FAILEDが満たされてSlackにメッセージが飛ぶが、今回はop2は成功するのでメッセージは飛ばない。triggerをONE_FAILEDに変更するとエラーが飛ぶ。contextに入っているtimestamp。エラーが起きた場合は下記のようなメッセージがSlackに送信される。開いてslackのwebhook用のConnectionを生成しておく。入っていたらbase_urlとして扱うとしているのでこれで良いらしい。urlのservice以降の部分)についてはVariablesの方に登録しておく。dag=dag)slackOpを上記のように変更すると、下記のようなメッセージがSlackに送信される。テンプレートを展開はしてくれないようだ。}}テンプレートの展開はクラスの中でtemplate_fields変数を定義するだけなので、下記のような自作のクラスを用意すれば展開できる。(1)リンク株式会社iMindスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。