Google Dataform を活用した分析基盤の構築

2025/02/07 公開7 min read

はじめに

株式会社 QualiArts でバックエンドエンジニアをしている筋野です。ゲームサーバーのインフラ構築や分析基盤の構築を主業務としつつ、必要に応じてゲーム機能の開発や内製ツールの開発も行っています。

ゲームを長く運用するためには、ユーザーの行動データを分析して適切な施策を打つことが重要になります。そのため、データの取得から分析までを効率的に行うためには、分析基盤が必要になります。本記事では、 QualiArts で利用している分析基盤の設計について紹介します。

分析基盤の概要

QualiArts ではインフラ環境に Google Cloud を利用することが多く、ゲームサーバーから送られてくるデータは BigQuery に集約しています。 BigQuery に集約されたデータを分析する際には、Google Cloud の Dataform というツールを利用して分析用のテーブルを作成しています。

Dataform は複数テーブルに跨るデータパイプラインを管理することができ、複雑な集計を行う際に便利なため、分析基盤として活用しています。

中間テーブルの構成

分析に利用する KPI (Key Performance Indicator) を計算する場合、複数のテーブルを結合して集計することになります。その際に、各テーブルを無造作に結合するとデータの整合性を取ることが難しく運用が困難になる可能性があります。

そのため、各中間テーブルは一定ルールに基づいたレイヤーに分けて管理しています。

また、各レイヤーのファイルは以下のディレクトリ構成で管理しています。

.
├── reporting/
│   └── kpi/
├── sources/
│   └── view/
└── staging/
    ├── extract/
    ├── snapshot/
    ├── summary/
    └── transform/

こちらのディレクトリ構成は、以前の Dataform の公式ドキュメントに記載されていたベストプラクティスの構成を参考にしています。しかし、2025年02月時点では以下の構成が推奨されているため、新しく構築する際には以下の構成を参考にすることをおすすめします。

リポジトリ構造のベストプラクティス

ワークフローのステージを反映するように、ファイルを definitions ディレクトリで構造化することをおすすめします。ニーズに最適なカスタム構造を採用できる点にもご留意ください。

推奨される次の definitions サブディレクトリの構造は、ほとんどの SQL ワークフローの主なステージを反映しています。

sources: データソース宣言を格納

intermediate: データ変換ロジックを格納

output: 出力テーブルの定義を格納

省略可: extras - 追加のファイルを格納

sources/view

ゲームサーバーから送られてきた生データを取得するレイヤーです。（Viewテーブル）

分析に必要なデータのみを抽出してデータ量を減らすことと、集計期間を限定する役割があります。基本的には日毎にデータを集計するため、この view レイヤーで前日までのデータを取得します。

対象テーブルには view_ という接頭辞を付ける事で、View レイヤーのテーブルであることを明示しています。

staging/extract

view レイヤーから必要なデータを抽出して保持するレイヤーです。（増分テーブル）

view レイヤーで取得したデータに対して、クレンジングや細かい調整を行います。具体的には、 Null 値の補完やアクセス状況のフラグ付け、重複の排除などを行っています。また、 extract レイヤーでは他テーブルとの結合は行わず、テーブル単体でのデータ整形を行います。

対象テーブルには ext_ という接頭辞をつける事で、extract レイヤーのテーブルであることを明示しています。

staging/summary

extract レイヤーで取得したデータを集計するレイヤーです。（増分テーブル）

extract レイヤーのテーブルを結合し、日毎のユーザーデータやイベントデータをそれぞれ集約して保持します。このレイヤーではデータの集約だけを行い、データの加工は行いません。

対象テーブルには sum_ という接頭辞をつける事で、summary レイヤーのテーブルであることを明示しています。

staging/snapshot

summary レイヤーで取得したデータを加工して分析に必要な情報を保持するレイヤーです。（増分テーブル）

summary レイヤーの情報を元にステータスの判定を行い、分析に利用するための情報としてデータを追加して保持します。具体的には、ユーザー毎のアクセス情報、リセマラ情報、ゲーム内進捗情報、課金情報などを追加します。分析を行う際はこのレイヤーのテーブルを利用してKPIの集計を行います。

対象テーブルには snap_ という接頭辞をつける事で、snapshot レイヤーのテーブルであることを明示しています。

staging/transform

snapshot レイヤーで取得したデータを再度加工、クレンジングして保持するレイヤーです。（増分テーブル）

snapshot レイヤーではユーザーのアカウント毎に必要な情報を保持しています。しかし、分析で利用する際にはリセマラアカウントの除外やデータ連携によるアカウント統合などを行う必要があります。これらを行うことで、より正確なKPIを出力するためのデータを保持します。

対象テーブルには tfm_ という接頭辞をつける事で、transform レイヤーのテーブルであることを明示しています。

reporting/kpi

transform レイヤーで取得したデータを元にKPIを計算するレイヤーです。（増分テーブル）

基本となる DAU (Daily Active User) や ARPU (Average Revenue Per User) などのKPIを計算して保持します。これらの情報はエンジニア以外も閲覧できるように Tableau などのBIツールに接続して可視化しています。また、 Tableau とは別に内製のBIツールが存在し、そこへ送るデータはこのレイヤーのテーブルを元に決められたフォーマットへ加工して送信します。

対象テーブルには kpi_ という接頭辞をつける事で、kpi レイヤーのテーブルであることを明示しています。

集計日までのループ処理

中間テーブルは作成する際に pre_operations と post_operations を利用することで、現在の集計日までの処理を一括で行うようにしています。

pre_operations {
    declare max_date default (
        ${fn_cmn.renderMaxDate(ref("sum_daily_access"))}
    );

    declare date_checkpoint default (
        ${when(incremental(),
            `${fn_cmn.renderCheckpoint(self(), true)}`,
            `${fn_cmn.renderCheckpoint(ref("sum_daily_access"), false)}`,
        )}
    );

    ${when(incremental(),
        `${fn_cmn.renderPreLoop(true)}`,
        `${fn_cmn.renderPreLoop(false)}`,
    )}
}

post_operations {
    ${when(incremental(),
        `${fn_cmn.renderPostLoop(true)}`,
        `${fn_cmn.renderPostLoop(false)}`,
    )}
}

max_date は、依存先テーブルに存在するレコードから集計日までの最大日が設定されます。

date_checkpoint は、集計中テーブルに存在するレコードから集計が完了している日付を取得し、次に集計される日付を設定します。

pre_operations では date_checkpoint が max_date に達するまでループする処理を記述し、 post_operations では date_checkpoint をインクリメントする処理を記述します。

上記の処理は複数のテーブルで共通して利用するため、共通関数として JavaScript のファイルに切り出しています。

/**
 * 計測期間の最大日を生成する
 * @param {string} table - 参照テーブル
 * @returns {string} max_date
 */
exports.renderMaxDate = (table) => {
    return `SELECT IFNULL(MAX(jp_date), ${constants.INIT_DATE}) FROM ${table}`;
};

/**
 * date_checkpointを生成する
 * @param {string} table - 参照テーブル
 * @param {boolean} incr - incremental判定
 * @returns {string} date_checkpoint
 */
exports.renderCheckpoint = (table, incr) => {
    if (incr) {
        return `SELECT IFNULL(DATE_ADD(MAX(jp_date), INTERVAL 1 DAY), ${constants.INIT_DATE}) FROM ${table}`;
    } else {
        return `SELECT MIN(jp_date) FROM ${table}`;
    }
};

/**
 * pre_operationsに記述するループ処理を生成する
 * @param {boolean} incr - incremental判定
 * @returns {string} ループ処理
 */
exports.renderPreLoop = (incr) => {
    if (incr) {
        return `LOOP IF date_checkpoint > max_date THEN RETURN; END IF;`;
    } else {
        return `IF date_checkpoint > max_date THEN RETURN; END IF;`;
    }
};

/**
 * post_operationsに記述するループ処理を生成する
 * @param {boolean} incr - incremental判定
 * @returns {string} ループ処理
 */
exports.renderPostLoop = (incr) => {
    if (incr) {
        return `SET date_checkpoint = DATE_ADD(date_checkpoint, INTERVAL 1 DAY); END LOOP;`;
    } else {
        return ``;
    }
};

これらの処理を記述することで、集計日までのループ処理を一括で行うことができます。また、テーブルの再作成が必要になった際も冪等性を保つことができるため、運用が容易になります。

内製BIツールへのデータ転送

BigQuery のデータから内製BIツールへデータを転送する際には、Cloud Run Jobs でバッチを作成して Cloud Scheduler と組み合わせることで定期的にデータを転送しています。その際、バッチ側では集計処理を行わず View テーブルを参照してデータを取得し、必要な形にフォーマットして送信しています。また、定期実行とは別にデータの修正などスポット作業が必要になる事を考慮して、 View テーブルに対応するテーブル関数も作成しています。

これらの View テーブルやテーブル関数も依存関係があるため、Dataform のパイプライン上で管理しています。

まとめ

Dataform を活用する事で BigQuery 上のデータを効率的に管理することができ、分析基盤の構築に役立てています。しかし、 Dataform 自体の運用やデータの保持方法など、運用していく中でまだまだ改善の余地があると感じています。今後も Dataform を活用して、より効率的な分析基盤の構築を目指していきたいと考えています。

この記事が皆様の開発に少しでもお役に立てれば幸いです。