BigQueryへWEBアプリケーションのログを登録してみる(その２）

こんにちは。次世代システム研究室のM.Mです。

前回のブログ(BigQueryへWEBアプリケーションのログを登録してみる)では、Pub/Subのサブスクリプションの配信タイプである「BigQueryへの書き込み」を利用して、WEBアプリケーションのログをBigQueryへ登録しました。
今回は、配信タイプにある「Cloud Storageへの書き込み」を利用して、Cloud Storageにログを貯めていき、まとめてBigQueryへ登録してみます。

1. Cloud Storageへ書き込む Pub/Sub サブスクリプション作成と簡易テスト

■準備

Cloud Storage バケットと、Pub/Sub トピック・サブスクリプションを作成します。
今回は以下の名前で作成しました。

• Cloud Storage バケット: mmtest-bucket
• Pub/Sub トピック: mmtest-topic
• Pub/Sub サブスクリプション: mmtest-topic-sub

■Pub/Sub サブスクリプション(mmtest-topic-sub)の配信タイプの設定を行う

上記の図のように、配信タイプとして「Cloud Storageへの書き込み」を選択すると、利用するバケットやファイル形式、ファイル名の接頭辞、接尾辞、ファイルのバッチ処理などの情報を設定できるようになります。

ファイルのバッチ処理の設定にて、Cloud Storageに新しい出力ファイルを作成するタイミングを指定できるようです。
詳細はこちら

今回は以下のように設定しました。

• 配信タイプ: Cloud Storage への書き込み
• ストレージファイル形式: text
• ストレージファイル接頭辞: mmtestapp-
• ストレージファイル接尾辞: .data

他はデフォルトのまま。ストレージバッチ期間は5分としています。

■簡易テストを行う（その１）

メッセージとして以下の内容を入力し公開します。

{"user_id": 101, "message": "test101", "logged_at": "2024-06-20 10:37:11"}

5分後には反映されるはず。

5分もたたずに出力されたようです。
ファイルのバッチ処理のストレージバッチの最長時間を5分に設定していましたが、最長時間の設定であり、5分より早くファイル出力されることもあるようです。
では、ファイルが出力されているかも確認します。

意図した接頭辞、接尾辞にもなっています。
ファイル名は[file-prefix][UTC-date-time]_[uuid][file-suffix]の形式になるようです。
詳細はこちら
また、ファイルの中身を確認したところ、メッセージの内容がそのまま出力されていました。

■簡易テストを行う（その２）

上記、簡易テストを1分以内に2回実施しました。
すると、バケットには以下のようなファイルが作成されていました。

1メッセージ1ファイルなのか？大量にファイルが作られることになるのか？
1ファイルに複数メッセージが出力される場合もあるとのことだったが。

■簡易テストを行う（その３）

以下の図のように、配信するメッセージ数を多くして公開してみます。
上記簡易テストと違い、メッセージの数を20に設定しています。

結果、以下の図のようなファイルが作成されました。

サイズを見ても分かりますが、サイズが増えているので、複数メッセージが1つのファイルに出力されているようです。
ファイルの中身を確認すると、以下のように1ファイルに5メッセージ出力されていました。

{"user_id": 201, "message": "test201", "logged_at": "2024-06-20 11:01:00"}
{"user_id": 201, "message": "test201", "logged_at": "2024-06-20 11:01:00"}
{"user_id": 201, "message": "test201", "logged_at": "2024-06-20 11:01:00"}
{"user_id": 201, "message": "test201", "logged_at": "2024-06-20 11:01:00"}
{"user_id": 201, "message": "test201", "logged_at": "2024-06-20 11:01:00"}

■わかったこと

• ファイルのバッチ処理の設定にて、Cloud Storageへのファイル作成タイミングの制御ができるが、厳密に5分間隔でファイル出力するといったことはできない。
• 1分以内に複数メッセージを送っても、1ファイルに複数メッセージが出力されるとは限らない。１ファイル１メッセージになっていた。
• 数秒間で複数メッセージを送った場合、1ファイルに複数メッセージが出力された。ただ、それでも複数ファイルに分割されていた。

勝手な想像ですが、ファイル出力するストレージバッチは複数並列で動いているのではと思いました。
なので、厳密な5分間隔にもならなければ、ストレージバッチ単体がキューからどれだけメッセージを取得できたかで、1ファイル1件だったり複数だったりするのかなと。
また、どのようなWEBアプリかにもよりますが、大量にアクセスがあるようなWEBアプリの場合、1つのフォルダ内に相当な数のファイルが作られることになりそう。
ちょっとそれは嫌だなという気持ちにもなり、日付単位のフォルダに分けることができるか試してみることにします。

2. 日付で分けたフォルダ内にファイル出力させる

■そもそもフォルダを設定できるのか？

上記の図のようにファイル名の接頭辞をlog/hogehoge/mmtestapp-に変更して実施します。
結果、以下の図のようにバケット直下だけでなく、フォルダも指定できることが分かりました。

■YYYYMMDDといった指定ができるのか？

ドキュメントには日付がカスタム可能とある。
ファイル名の接頭辞をlog/YYYYMMDD/hhmm/ss-にして変更して実施します。

ダメでした。日付に変換されることもなく、YYYYMMDDといったそのままのフォルダ名で作られてしまいました。
ドキュメントには、

たとえば、ファイル名の日時形式の値が YYYY-MM-DD/hh_mm_ssZ の場合、サンプル オブジェクト名は prod_2023-09-25/04_10_00Z_uNiQuE_archive です。

と書かれているのだが・・
よくあるコンソール画面からは設定できない項目かなと思い調べたところ、やはりそうでした。
gcloudコマンドには以下の日付フォーマットのパラメータがありました。
cloud-storage-file-datetime-format
コンソール画面からはファイル名の接頭辞、接尾辞は設定できるが、ファイル名の日付形式は設定できないようです。

・gcloudコマンドで設定反映します

gcloud pubsub subscriptions update mmtest-topic-sub
 --cloud-storage-bucket=mmtest-bucket
 --cloud-storage-file-prefix=
 --cloud-storage-file-suffix=-mmtestapp.data
 --cloud-storage-file-datetime-format=YYYYMMDD/hhmm/ss

※注意: YYYYMMDDだけだとエラーになりました。hhmmssも必要なようです。

・コンソール画面では設定されたか確認できないのでgcloudコマンドで確認します。

設定されていることが確認できました。
メッセージを送って確認します。

いいですね。日付フォルダになりました。
その後、再度実行して、時間フォルダが変わるかも確認します。

0445フォルダが作れました。時間によって新たなフォルダが作られることが分かりました。

前回のブログで行ったようにCloudRunサービスからメッセージを送るようにします。
前回とほとんど変わらないですが、以下が利用したプログラムになります。

import datetime
import random
import json
from flask import Flask
from google.cloud import pubsub_v1
app = Flask(__name__)

@app.route('/', methods=['GET'])
def publish():
    publisher = pubsub_v1.PublisherClient()
    topic_name = 'projects/{project_id}/topics/{topic}'.format(
        project_id='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
        topic='mmtest-topic',
    )

    log_data = _get_log_data()
    publisher.publish(topic_name, json.dumps(log_data).encode('utf-8'))
    return 'OK'

def _get_log_data():
    user_id = random.randint(1, 1000000)
    message = f"test_{user_id}"
    logged_at = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    return {
        "user_id": user_id,
        "message": message,
        "logged_at": logged_at
    }

if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=8080)

送信するメッセージ内容は、上記簡易テストで実施していた内容と同じです。

• user_id
• message
• logged_at

のJSONフォーマットです。

10qpsで30分間実行したところ、フォルダは以下の図のように5分間隔で作成されていました。

100pqsでも実施しましたが特に問題はありませんでした。
日付フォーマットを設定することで、日付フォルダに振り分けることができました。

3. Cloud Storageに作成された複数ファイルをBigQueryに取り込む

■スキーマファイルを用意する

[
  {
    "name": "user_id",
    "mode": "NULLABLE",
    "type": "INTEGER"
  },
  {
    "name": "message",
    "mode": "NULLABLE",
    "type": "STRING"
  },
  {
    "name": "logged_at",
    "mode": "NULLABLE",
    "type": "TIMESTAMP"
  }
]

mmtest_app_log.jsonというファイル名で作りました。

BigQueryに取り込み実行

bq load
 --replace
 --synchronous_mode=false
 --source_format=NEWLINE_DELIMITED_JSON
 --schema=./mmtest_app_log.json
 mmtest_dataset.mmtest_app_log
 gs://mmtest-bucket/20240620/*.data

• BigQuery データセット、テーブル: mmtest_dataset.mmtest_app_log
• スキーマ: mmtest_app_log.json（上記にて作成したスキーマファイルを指定）
• ソース: gs://mmtest-bucket/20240620/*.data (上記にて作成された日付フォルダ配下すべてを取り込み対象としています）

・取り込み結果

問題なく登録されていることが確認できました。
日次で定期的にBigQueryに取り込むようにすればよさそうです。
（BigQueryに取り込まなくても外部データとしてCloud Storageを参照すればいいような気もしてきましたが。）

4. まとめ

Cloud Storageへの書き込みサブスクリプションを利用してバケットにログファイルを貯めていき、そのログファイルを利用してBigQueryへ登録することができました。
ただ、コストについてのドキュメントを確認したところ、下記の記載がありました。

BigQuery サブスクリプションでは、サブスクリプションからの読み取り（サブスクライブのスループット）と BigQuery への書き込みに対して、すべての Google Cloud リージョンにおいて TiB あたり$50 の料金がかかります。BigQuery データ取り込みに追加料金はかかりません。

Cloud Storage サブスクリプションでは、すべての Google Cloud リージョンにおいて、サブスクリプションからの読み取り（サブスクライブのスループット）と Cloud Storage への書き込みに TiB あたり$50 の料金がかかります。

どちらもTiB あたり$50のようです。
Cloud Storage書き込みにするとコストが抑えられるわけでもなく、BigQueryへログを取り込むという目的であれば、素直にBigQueryサブスクリプションを利用するほうがよさそうです。

最後に、次世代システム研究室では、グループ全体のインテグレーションを支援してくれるアーキテクトを募集しています。アプリケーション開発の方、次世代システム研究室にご興味を持って頂ける方がいらっしゃいましたら、ぜひ募集職種一覧からご応募をお願いします。

皆さんのご応募をお待ちしています。