BigQueryログイベントをトリガーにしてCloud Functionsを起動させる方法

使用したツール

• BigQuery
• Cloud Logging
• Cloud PubSub
• Cloud Functions

プロセス

BigQueryのログをCloud Loggingでフィルタリングする

• 私の場合は、メアドとジョブ名を指定することで、「自分がBigQueryでインサートジョブを時実行した」ログを抽出しました

ログに記載する内容

resource.type="bigquery_dataset"
protoPayload.authenticationInfo.principalEmail="メアド"
protoPayload.serviceName: "bigquery.googleapis.com"
protoPayload.methodName="google.cloud.bigquery.v2.JobService.InsertJob"
protoPayload.metadata.tableDataChange.reason="QUERY"

BigQueryのログ抽出する際に参考にした記事

GA4 BigQueryエクスポート遅延に対応するイベントドリブンアーキテクチャの提案 | アユダンテ株式会社

Cloud Loggingからログシンクを作成

• 下記のCreate sinkからシンクを作成します

• Sink Name: 任意
• Sink destinationのSelect sink service: Cloud PubSub Topic
• Select a Cloud PubSub topic: 任意で作成
• Choose logs to include in sink: すでにCloud Loggingで入力したクエリがある
• Create sinkをクリック

Cloud Functions からPubSubトピックを選択し起動トリガーにする

• Create functionから関数を作成する
  • Environment: 2nd gen
  • Function name: 任意
  • Region: 任意
  • Trigger type: Cloud PubSubを押し、作成したPubSubトピックを選択
  • Saveをクリック
• CodeにてCloud Functionsで起動する関数を作成する
  • Runtime: 今回はPython3.9を選択
  • Entry point: 任意(Codeに記述する関数と同じ関数名にしないといけない)
• Code内容（今回は、Cloud Loggingをトリガーにしてそのログの内容のsource名とmassage_idを使って文字列を出力するものにしました）
• 左下のデプロイボタンからデプロイ
• print内容はログ上に出力されます

sample.py

# まずはPub/Subから取ってきたログをfunctionsの中の変数に読み込ませる 
## functionsの機能使いますよのおまじない 
import functions_framework 

## cloud_event関数を使いますよの宣言 
@functions_framework.cloud_event 

### この関数を作成する際に参考にした記事は【<https://blog.g-gen.co.jp/entry/cloud-storage-to-bigquery#mainpy-%E3%81%AE%E5%86%85%E5%AE%B9】> 
## 関数を宣言し, 引数にcloud_eventを指定 
def cloudevent_function(cloud_event): 
 topic_name = cloud_event["source"] 
 message_id = cloud_event.data["message"]["message_id"] 
### print内容はログに出力されます 
 print("topicの名前は" + topic_name + "で、messageidは" + message_id + "で、tomがbqで行を挿入したみたい。")

cloud_event関数を使ったコードを書く際の参考記事

Eventarcトリガーを利用してCloud StorageのファイルメタデータをBigQueryへ格納してみた - G-gen Tech Blog

• Cloud Functions第2世代からcloud_event関数を使用します
  • cloud_event関数とは
• attributesのkeyを使って値を取得する場合

  cloud_event["キー名"]
  

• dataのkeyを使って値を取得する場合

  cloud_event.data["キー名"]
  

• dataのkeyの中のkeyを使う場合

  cloud_event.data["キー名"]["キー名"]

Cloud Functionsが正常に起動したか確認

• BigQueryでトリガーにしたジョブを起動する
• Cloud FunctionsのLOGSで正しく出力されているか確認する

ここまでの感想

• 思ったよりも大変でした。。。
• 原因はCloud Event関数の仕様があまりわからなかったからです（泣）
• ぜひ、BigQueryのイベントをトリガーにしてCloud Functionsから様々な操作をしてみてください〜