DoiT Internationalでは、Google Cloud Platform向けのコスト最適化プラットフォームreOptimize（現DoiT Cloud Intelligence™）のデータ分析基盤として、Google BigQueryを幅広く活用しています。
Google BigQueryは、標準SQL構文で大規模なデータセットに対するクエリを実行できる、優れたサーバーレスツールです。料金はクエリ実行時にスキャンされたデータ量に応じて決まり、最大50件のクエリを同時実行できます（キャッシュ済みのクエリはカウントされません）。
組織内に複数のチームがある場合、いま何が走っているのかを把握する手段が欠かせません。標準機能では実行中のクエリを確認できないため、ここで活躍するのがbqtopです。
実行中のクエリと過去の履歴を確認できる、シンプルなコマンドラインツールとWebアプリケーションをつくりたいと考えました。必要な情報を取得するために、完全サーバーレスのパイプラインを構築し、アプリケーション側から情報を参照できるようにしています。
まずは「サーバー」側から
最初に、BigQueryのログをGoogle Pub/Subにエクスポートするため、Stackdriver Logs Sinkを2つ作成します。1つ目はクエリ開始を示すログをフィルタリングするシンクです:
resource.type=bigquery_resource protoPayload.methodName=jobservice.insert2つ目は完了したクエリ用のシンクです:
resource.type=bigquery_resource protoPayload.methodName=jobservice.jobcompleted各シンクは、それぞれ別のPub/Subトピック（bqtop-running-jobs_と_bqtop-finished-jobs）にデータを書き込みます。
ログからPub/Subへデータが流れるようになったら、次は後からWebアプリやCLIで読み取れるようにデータベースへ書き込む必要があります。今回のユースケースで求めたのは、シンプルかつパワフルで、クライアントから直接読み取れる（本格的なバックエンドを組まずに済む）うえ、新規・更新データの通知機能を備えている（更新確認のためにデータベースを定期的にポーリングしなくてよい）仕組みです。GoogleのFirebaseはまさに自然な選択肢でした。
Pub/SubからFirebaseデータベースへデータを取り込むには、Firebase Cloud Functionsを利用します。Pub/Subトピックの新着イベントを購読し、データベースに書き込みます。
https://gist.github.com/avivl/30d92a579abd48dd4b3a9131b7f6abfb
データは2種類のデータベース「テーブル」に分けて保存します。1つは実行中のジョブ用、もう1つは完了したタスク用です。ジョブが完了したら、実行中ジョブのテーブルから削除する必要があります。さらに、データの変更を監視する関数もいくつか追加で用意しました。
https://gist.github.com/avivl/58dbe67e6ade2b45f89e5a5d698c3dd3
完了ジョブのリファレンスに新しいデータが追加されたイベントを受け取ると、実行中ジョブのリファレンスから同じjobIdのイベントを探し出し、削除します。
https://gist.github.com/avivl/c16af302a09a338f41480a11689a10d9
パフォーマンス向上のため、インデックスも作成しています（database.rules.jsonファイルで設定）。
https://gist.github.com/avivl/7ab33806e5d8da4caf62b9a57778433a
次は「クライアント」側
「サーバー」側が一通りそろったので、続いてクライアント側に移ります。Engineersが使うコマンドラインツールと、オフィスのダッシュボードTVに映し出すWebアプリの両方を用意したいと考えました。
コマンドラインアプリは、UI制御にcursesを利用した小さなPythonアプリケーションです。Firebase APIのラッパーとしてpyrebaseを採用しています。先ほども触れたとおり、データベースを常時ポーリングするのではなく、変更があったタイミングで通知を受け取る方式を選びました。2つのデータベースストリームを利用することで、変更が発生するたびにコールバックが呼び出されます。
https://gist.github.com/avivl/8ea27a7f98feb1d8735151ae3ba97bff
ハンドラ関数が呼び出されたら、データを取得してユーザーに表示します。
Webアプリ側は、ReactフレームワークとFirebase JavaScript SDKを用いて、データベースから読み取った情報をWebページ上に可視化しています。
まとめると、サーバーを一切デプロイすることなく、最小限のコーディングで、BigQueryのログをGoogle Pub/Subへ送り、Cloud Functionsで処理し、最終的にFirebase Realtime Databaseに格納する、なかなか気の利いたパイプラインを構築できました。