💨

GCPを利用した基盤構築及びデータ取得作業の定期実行について①

2022/09/13に公開

Google Cloud

はじめに

これは、Web APIとGCPを利用し5日間の天候データ取得の作業を自動化した記事です。
これらの記事を参考にしています。

必要なもの・作業の流れ

・OpenweathermapのAPIキー
・GCPのアカウント（要はgoogleアカウント）

Openweathermapは、世界の天気に関する情報を取得することが出来るサービスです。
https://openweathermap.org/
APIキーはアカウントを作成すれば発行されると思います。（ログインして[API keys]をクリック）

今回は、無料で利用できる5日間の天気データを取得し、GCPのバゲット及びBigQueryのテーブルに保存する作業を自動化していきます。具体的には以下の通りです。

Cloud Schedulerで指定した時間にPub/Subへメッセージを送信
メッセージを受信したPub/SubがCloud Functionsを起動
起動したCloud Functionsが以下の作業を実行
　・Openweathermapからデータを取得し、データフレームに変換
　・データフレームをCloud Storageにcsvファイルとして保存
　・データフレームをBigQueryのテーブルに登録

bigqueryのテーブル、Cloud Storageのバケット、Pub/Subを、上の2つの記事に従い先に作成しています。
bigqueryのテーブルはこのように作りました。（中身は空のテーブル）

Cloud Functionsの設定

・文字列をまとめたsettings.pyファイル
・api経由で天候データ取得を実行するopenweather.pyファイル
・データをGCSへcsvファイルで保存するstorage.pyファイル
・BigQueryのテーブルにデータを送るbigquery.pyファイル
・Cloud Functionsで実行するmain.pyファイル
・実行に必要な外部ライブラリを記載するrequirement.txt
以上5つのファイルと1つのテキストの形でコードを記述しています。

環境：第１世代
リージョン：us-west1
トリガー：Pub/Sub
割り当てられたメモリ：512 MB
タイムアウト：60 秒
ランタイム：Python3.8
エントリポイント：main
ソースコード：インラインエディタ

settings.py

city = '取得したい地名を入力' #ex)'Tokyo,JP'、'London,GB'
API_KEY = '自分のOpenweathermap APIを入力'
url = "http://api.openweathermap.org/data/2.5/forecast?q={city}&appid={key}&lang=ja&units=metric"
project_id = '自分のproject_idを入力'
bucket_name = 'csvファイルを保存する先のbucket_nameを入力' #ex)'wm'

units=metric：摂氏の温度を取得、lang=ja：日本語で出力

openweather.py

from datetime import datetime, timedelta, timezone
import json
import pandas as pd
import requests
import settings

# api経由で天候データ取得を実行する関数
def openweathermap():
    # Web APIから天気データを取得(json形式で取得することになる)
    url = settings.url.format(city=settings.city, key=settings.API_KEY)
    json_data = requests.get(url).json()

    tz = timezone(timedelta(hours=+9), 'JST') # 日本のタイムゾーンを作成
    j, m, p, s, wea, wind = [], [], [], [], [], []
    # 取得したデータを加工し、リストに追加する
    for dat in json_data['list']:
        jst = str(datetime.fromtimestamp(dat["dt"], tz))[:-9] # UTCの時間をJSTに変換
        j.append(jst)
        m.append(pd.Series(dat['main']))
        p.append(dat['pop'])
        s.append(pd.Series(dat['sys']))
        wea.append(pd.Series(dat['weather'][0]))
        wind.append(pd.Series(dat['wind']))
    
    # リストをデータフレームに変換し、連結する。
    dt = pd.Series(j, name='time')
    dfm = pd.DataFrame(m).drop(columns=['temp_kf'])
    dfp = pd.Series(p, name='pop')
    dfs = pd.DataFrame(s)
    dfw1 = pd.DataFrame(wea).drop(columns=['id','icon'])
    dfw2 = pd.DataFrame(wind)    
    df = pd.concat([dt, dfm, dfp, dfs, dfw1, dfw2], axis=1)
    return df

storage.py

from google.cloud import storage as gcs
import openweather
import datetime
import settings

# データをGCSへcsvファイルで保存する関数
def send(df):
    project_id = settings.project_id 
    bucket_name = settings.bucket_name #ex)settings.pyより'wm'
    gcs_path_1 = "data/openweathermap_data" # ファイルまでのパス
    # ファイル名でデータを取得した日付がわかるようにする
    to_day = str(datetime.date.today()) # import datetimeを書く必要がある。
    to_day = to_day.replace("-","")
    
    client = gcs.Client()
    bucket = client.get_bucket(bucket_name)
    
    blob_gcs_1 = bucket.blob(gcs_path_1 + "_" + to_day + ".csv")
    blob_gcs_1.upload_from_string(data=df.to_csv(index=False))

次のような形でcsvファイルに保存される。
wm/data/openweathermap_data_20220909.csv

datetime.date.today()で日付を付けるには「import datetime」でないと上手くいかない。
「from datetime import datetime, timedelta, timezone」で行うとエラーが発生する。

bigquery.py

from google.cloud import bigquery as gbq
import storage
import openweather
import settings

# BigQueryのテーブルにデータをインサートする関数
def insert(df):
    client = gbq.Client(settings.project_id)
    table = client.get_table("プロジェクト名.データセット名.テーブル名")
    client.insert_rows(table, df.values.tolist())

main.py

import settings
import openweather
import storage
import bigquery

# Cloud Functionsで実行する関数
def main(event, context):
    df = openweather.openweathermap()
    storage.send(df)
    bigquery.insert(df)

実行する関数の引数には（event, context）を設定する必要がある。（無いと上手くいかない）

requirements.txt

# Function dependencies, for example:
# package>=version
requests>=2.28.1
pandas>=1.3.5
google-cloud-storage>=1.33.0
google-cloud-bigquery>=2.20.0

標準ライブラリ（jsonなど）は記載しなくて良い。

Cloud Schedulerの設定

こちらも上にある参考記事に従い作成しました。
ただジョブ作成の頻度をunix-cronという見慣れない形式で作成することに注意が必要です。
（私は初めて知りました。）
私は22:00で5日おきに作成したかったので「0 22 */5 * *」と書きました。
書き方については以下の記事を参考にしています。

リージョン：us-west1
頻度：0 22 */5 * *
タイムゾーン：日本標準時（JST）
ターゲットタイプ：Pub/Sub

まとめ

ここまで設定すれば後は、Cloud SchedulerとOpenweathermapのAPIを起動させておけば自動でデータを取得することが出来るようになります。
一応自分でやった時は上手くできたので大丈夫だと思います。
注意点として、OpenweathermapのAPIは起動直後だと上手くデータを取得できないので予め起動しておく必要があります（20分前ぐらいで大丈夫です）。
起動するだけならスマートフォンでもできるので、難しくはないです。

はじめに

必要なもの・作業の流れ

Cloud Functionsの設定

Cloud Schedulerの設定

まとめ

Discussion