【Kaggleで役立つ】計算過程およびエラーをslackに通知する

はじめに

KaggleのNotebookやGoogle Colab, GCPなどのクラウド環境で計算を回していると、途中でエラーが発生していて計算が止まってしまったり, 処理がどう進んでいるかが気になってしまうことが多々あったので, 欲しい情報をslackに通知するようにしてみました。

この記事ではプログラムの処理過程をslackに通知するための方法について説明していきます。

参考にしたサイト

tech.bita.jp amalog.hateblo.jp

slackの設定

まずはslackの設定を行なっていきます。slackに登録していない方はこちらからサインインしてください。 それでは初めにワークスペースを作成します。私は1人で使う個人開発用のワークスペースを作成しました。 その後作成したワークスペースに入り、以下の図に示すように左側のappというタグを選択し、その後中央の画面で『Incoming Webhook』というアプリを検索してslackに追加します。

追加したら『Incoming Webhook』をクリックするとweb上の設定画面にリンクします。 設定画面で大事なのは主にこちらの２項目です。

1つ目に通知を送りたいチャンネルを選択します。もともとあるgenralなどでもいいですし新しいチャンネルを作成しそちらに通知を送るようにしてもいいです。 2つ目にWebhook URLのコピーです。Webhook URLというものが割り当てられますのでそれをコピーしておいてください。後ほど使います。そのほかの設定はお好みで大丈夫です。 これでslackでの設定は完了です。

関数の定義

次にslackに通知を送るためのコードを書いていきます。 動かしたいプログラムの中に以下のコードを貼り付けます。 コード中の<自分のWEBHOOK_URL>の部分は先ほど取得したWebhook URLに書き換えてください。

import json
import requests

# 任意のメッセージを通知する関数
def send_slack_message_notification(message):
    webhook_url = '<自分のWEBHOOK_URL>'  
    data = json.dumps({'text': message})
    headers = {'content-type': 'application/json'}
    requests.post(webhook_url, data=data, headers=headers)

# errorを通知する関数
def send_slack_error_notification(message):
    webhook_url = '<自分のWEBHOOK_URL>' 
    # no_entry_signは行き止まりの絵文字を出力
    data = json.dumps({"text":":no_entry_sign:" + message})  
    headers = {'content-type': 'application/json'}
    requests.post(webhook_url, data=data, headers=headers)

これで準備は整いました。次に実際の使い方を説明していきます。

使い方

まず任意のメッセージを通知する方法ですが以下のように先ほど定義した関数の引数に通知したい文章を与えることで通知することが可能です。 str型であれば基本、なんでも大丈夫です。

message1 = "[START] training phase"

val_score = 0.89
message2 = "valid score:{}".format(val_score)

send_slack_message_notification(message1)
send_slack_message_notification(message2)

通知結果はこのようになりました。kaggleの場合だと現在の処理フェーズであったり、算出したスコアであったりを通知すると便利そうです。

次にerrorが起きたときに通知する方法です。
今回はmainという関数内でエラーがあった場合を想定して説明していきます。ここは別に関数でなくても大丈夫です。 try構文を用いて例外処理を記述し、例外(エラー)が発生した場合にexcept以下を実行します。 通知されるのはsend_slack_error_notificationの引数に与えられた文章です。今回はtraceback.format_exc()によってerror内容を通知しています。

import traceback
def main():
    prin("試しにわざとエラーを出してみます。")

try:
    main()
except:
    send_slack_error_notification("[ERROR]\n" + traceback.format_exc())  

このような形でエラー内容を通知することができました。「クラウド上でプログラムを回してたらいつの間にかerrorで止まっていた、」なんてことはよくあるので、そういったときに役立ちそうです。例外処理を部分部分でいちいち書くのは面倒なので、全ての処理をmainの関数に集約して、その関数に対して上記のように実行すると楽だと思います。

おわりに

kaggleなどの機械学習タスクでは、処理時間がかなり長いため処理がうまくいってるか気になってしまう人は多いと思います。なのでこのように通知を送ることによって、計算の過程やエラーを確認し効率よく計算を回していければと思います。