PythonでBigQueryの実行情報をSlackへ共有する方法

この記事は GMOアドマーケティングAdvent Calendar 2020 3日目の記事です。

皆さん

こんにちは、GMOアドマーケティングのS.Rです。

前回のブログでは、GCPのSDKでBigQueryを実行する状況をBigQueryのテーブルへ格納するプログラムの作り方を皆さんへ紹介させていただきました。
実際の運用にはBigQueryの実行状況を集計することだけではなく、大きな課金が発生する場合にリアルタイムで責任者へ共有することも重要です。
今回はBigQueryの実行する情報を、よく使われているSlackという業務用チャットツールへ共有する方法を皆さんへ紹介します。

Slack側の設定

1 連携用Slack Appを作ります。

　Slackの管理画面を開いてAppの名前やwork placeの名前を指定してください。今回のブログではAppの名前をbigquery_cost_managementにしてみました。

2 新しいWebhookを作ります。

管理画面で左側の「Incoming Webhooks」を選択してWebhooksの設定画面へ遷移します。

Webhooksの設定画面の一番下で「 Add New Webhook to Workspace」を開いてみましょう。

次のページで連携したいChannel又は責任者のSlack ID(Direct Messages)を選択します。

3 権限の認証に必要な情報をメモします。

Webhookの管理画面で先に作成されたWebhookのURLをコピーしてメモします。

Pythonのプログラムを作成します。

Pythonプログラムの環境構築部分は前回のブログで紹介しましたので詳しくはそちらを参考にしてみてください。

1. Libraryをimportします。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pandas as pd
from google.cloud import bigquery
from google.oauth2.service_account import Credentials
import requests, json
from datetime import datetime,timedelta
import os
import io
import json
from pytz import timezone, utc

#!/usr/bin/env python

# -*- coding: utf-8 -*-

import pandas as pd

from google.cloud import bigquery

from google.oauth2.service_account import Credentials

import requests, json

from datetime import datetime,timedelta

import os

import io

import json

from pytz import timezone, utc

2. Service accountのkey fileをロードします。

with io.open('./bigquery_police.json', 'r', encoding='utf-8') as json_fi:
    credentials_info = json.load(json_fi)
    list_credentials = Credentials.from_service_account_info(credentials_info)
client = bigquery.Client(credentials=list_credentials)

with io.open('./bigquery_police.json', 'r', encoding='utf-8') as json_fi:

credentials_info = json.load(json_fi)

list_credentials = Credentials.from_service_account_info(credentials_info)

client = bigquery.Client(credentials=list_credentials)

3. 30分前から今の時点までのBigQueryのjobをリストします。

def list_jobs(client):
    min_creation_time = datetime.now() - timedelta(minutes=30)
    return client.list_jobs(max_results=10000, min_creation_time=min_creation_time, all_users=True,state_filter="done")

def list_jobs(client):

min_creation_time = datetime.now() - timedelta(minutes=30)

return client.list_jobs(max_results=10000, min_creation_time=min_creation_time, all_users=True,state_filter="done")

4.実際の課金金額を換算する関数を書きます。

実行履歴には実際の課金金額は記載されていないので手動で換算する必要があります。計算式は下記です。

job.total_bytes_billed:利用されたバイト数です。
BYTE2GB: バイトからGBへ変換する率です。
RATE: １GBの課金金額です。

def bytes_billed2price(job):
    BYTE2GB = 1073741824 * 1000
    RATE = 5 * 108
    return (job.total_bytes_billed /BYTE2GB) *RATE

def bytes_billed2price(job):

BYTE2GB = 1073741824 * 1000

RATE = 5 * 108

return (job.total_bytes_billed /BYTE2GB) *RATE

5. SlackのWebhookへ課金情報を送ります。

def notify(webhook_url, icon_emoji, jobInfo): 
    message = get_alert_message(job.total_bytes_billed)
    query_url = u"https://bigquery.cloud.google.com/results/gmo-am-pmp:" + jobInfo.job_id
    message += u'*jobid: ' + jobInfo.job_id + '*\n'
    message += u'*user_email: ' + jobInfo.user_email + '*\n'
    message += u'*実行時間: ' + job.created.strftime('%m/%d/%Y, %H:%M:%S') + '*\n'
    message += '*料金(円): ¥' + str(bytes_billed2price(jobInfo)) + '*\n';
    message += '*queryのurl*: ' + query_url + '\n'
    requests.postd(webhook_url, data = json.dumps({
        'text': message,  #通知内容
        'icon_emoji': icon_emoji,  #アイコン
        'link_names': 1,  #名前をリンク化
        'channel':channel
    }))

#前に作成されたWEBHOOKのUrl
WEBHOOK = "https://hooks.slack.com/services/T02EZ9DSX/*****" 
ICON_EMOJI =':money_with_wings:',

jobs = list_jobs(client)
for job in jobs:
  notify(WEBHOOK, ICON_EMOJI, jobInfo)

def notify(webhook_url, icon_emoji, jobInfo):

message = get_alert_message(job.total_bytes_billed)

query_url = u"https://bigquery.cloud.google.com/results/gmo-am-pmp:" + jobInfo.job_id

message += u'*jobid: ' + jobInfo.job_id + '*\n'

message += u'*user_email: ' + jobInfo.user_email + '*\n'

message += u'*実行時間: ' + job.created.strftime('%m/%d/%Y, %H:%M:%S') + '*\n'

message += '*料金(円): ¥' + str(bytes_billed2price(jobInfo)) + '*\n';

message += '*queryのurl*: ' + query_url + '\n'

requests.postd(webhook_url, data = json.dumps({

'text': message, #通知内容

'icon_emoji': icon_emoji, #アイコン

'link_names': 1, #名前をリンク化

'channel':channel

}))

#前に作成されたWEBHOOKのUrl

WEBHOOK = "https://hooks.slack.com/services/T02EZ9DSX/*****"

ICON_EMOJI =':money_with_wings:',

jobs = list_jobs(client)

for job in jobs:

notify(WEBHOOK, ICON_EMOJI, jobInfo)

実行した結果

弊社でこのプログラムでBigQueryを実行した時に”逮捕”されたSlackの通知は下記です。

纏め

今回はPythonで、GCPのSDKでBigQueryを実行する状況をSlackと連携するプログラムを作る方法を紹介しました。いかがだったでしょうか。
前回のブログと合わせて簡単にBigQueryの料金管理システムを作れると思います。
もしこの二つのブログが皆さんのBigQueryの料金管理にお役に立てば幸いです。

明日は、syamas10さんによる「ARCHモデルで時系列データの変動の大きさを見積もる」です。

引き続き、GMOアドマーケティングAdvent Calendar 2020をお楽しみください！

■エンジニア採用ページ　～福利厚生や各種制度のご案内はこちら～
https://www.gmo-ap.jp/engineer/

■noteページ　～ブログや採用、イベント情報を公開中！～
https://note.gmo-ap.jp/

shicho ryu