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

<h2>使用したツール</h2>
<ul>
 	<li>BigQuery</li>
 	<li>Cloud Logging</li>
 	<li>Cloud PubSub</li>
 	<li>Cloud Functions</li>
</ul>
<h2>プロセス</h2>
<h3>BigQueryのログをCloud Loggingでフィルタリングする</h3>
<ul>
 	<li>私の場合は、メアドとジョブ名を指定することで、「自分がBigQueryでインサートジョブを時実行した」ログを抽出しました</li>
</ul>

```sql: ログに記載する内容
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"
```

<h3>BigQueryのログ抽出する際に参考にした記事</h3>
<a href="https://ayudante.jp/column/2022-07-22/11-05/">GA4 BigQueryエクスポート遅延に対応するイベントドリブンアーキテクチャの提案 | アユダンテ株式会社</a>
<h2><a href="https://console.cloud.google.com/logs">Cloud Logging</a>からログシンクを作成</h2>
<ul>
 	<li>下記のCreate sinkからシンクを作成します</li>
</ul>
<img class="alignnone size-full wp-image-52" src="http://tomojoz.com/wp-content/uploads/2022/11/スクリーンショット-2022-11-14-16.44.09-2.png" alt="" width="1270" height="593" />
<ul>
 	<li>Sink Name: 任意</li>
 	<li>Sink destinationのSelect sink service: Cloud PubSub Topic</li>
 	<li>Select a Cloud PubSub topic: 任意で作成</li>
 	<li>Choose logs to include in sink: すでにCloud Loggingで入力したクエリがある</li>
 	<li>Create sinkをクリック</li>
</ul>
<h2><a href="https://console.cloud.google.com/functions/">Cloud Functions</a> からPubSubトピックを選択し起動トリガーにする</h2>
<img class="alignnone size-full wp-image-53" src="http://tomojoz.com/wp-content/uploads/2022/11/スクリーンショット-2022-11-14-16.49.19.png" alt="" width="1274" height="471" />
<ul>
 	<li>Create functionから関数を作成する
<ul>
 	<li>Environment: 2nd gen</li>
 	<li>Function name: 任意</li>
 	<li>Region: 任意</li>
 	<li>Trigger type: Cloud PubSubを押し、作成したPubSubトピックを選択</li>
 	<li>Saveをクリック</li>
</ul>
</li>
 	<li>CodeにてCloud Functionsで起動する関数を作成する
<ul>
 	<li>Runtime: 今回はPython3.9を選択</li>
 	<li>Entry point: 任意(Codeに記述する関数と同じ関数名にしないといけない)</li>
</ul>
</li>
 	<li>Code内容（今回は、Cloud Loggingをトリガーにしてそのログの内容のsource名とmassage_idを使って文字列を出力するものにしました）</li>
 	<li>左下のデプロイボタンからデプロイ</li>
 	<li>print内容はログ上に出力されます</li>
</ul>

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

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

### この関数を作成する際に参考にした記事は【&lt;https://blog.g-gen.co.jp/entry/cloud-storage-to-bigquery#mainpy-%E3%81%AE%E5%86%85%E5%AE%B9】&gt; 
## 関数を宣言し, 引数に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で行を挿入したみたい。")
```

<h3>cloud_event関数を使ったコードを書く際の参考記事</h3>
<a href="https://blog.g-gen.co.jp/entry/cloud-storage-to-bigquery">Eventarcトリガーを利用してCloud StorageのファイルメタデータをBigQueryへ格納してみた - G-gen Tech Blog</a>
<div class="concept-box6"><p>
<ul>
 	<li>Cloud Functions第2世代からcloud_event関数を使用します
<ul>
 	<li><a href="https://www.notion.so/cloud_event-b7abe0a2f6504d68b820b91da0765965">cloud_event関数とは</a></li>
</ul>
</li>
 	<li>attributesのkeyを使って値を取得する場合
<pre><code class="language-json">cloud_event["キー名"]
</code></pre>
</li>
 	<li>dataのkeyを使って値を取得する場合
<pre><code class="language-json">cloud_event.data["キー名"]
</code></pre>
</li>
 	<li>dataのkeyの中のkeyを使う場合
<pre><code class="language-json">cloud_event.data["キー名"]["キー名"]</code></pre>
</li>
</ul>
</p></div>
<h2>Cloud Functionsが正常に起動したか確認</h2>
<ul>
 	<li>BigQueryでトリガーにしたジョブを起動する</li>
 	<li>Cloud FunctionsのLOGSで正しく出力されているか確認する</li>
</ul>
<h2>ここまでの感想</h2>
<ul>
 	<li>思ったよりも大変でした。。。</li>
 	<li>原因はCloud Event関数の仕様があまりわからなかったからです（泣）</li>
 	<li>ぜひ、BigQueryのイベントをトリガーにしてCloud Functionsから様々な操作をしてみてください〜</li>
</ul>