GA4データをDatabricksのBronze層へ─BigQuery・GCS・S3をつなぐデータ連携

2026年11月10日 18:59

GA4のイベントデータを、ほかのデータソースと同じ基盤で扱いたい。そのために、既存のAWS・Databricks基盤へ取り込む構成を用意しました。

私たちのデータ基盤は、各データソースをAWSのS3に集め、Databricksで処理する構成です。GA4のデータは、BigQueryからCloud Storage（以下、GCS）へ書き出し、AWS DataSyncでS3へ転送します。

本稿では、DatabricksのBronze層へ取り込むまでの構成と、後から変わる日次データを扱う際の注意点を紹介します。

1. BigQueryから、既存のS3・Databricksの経路へつなぐ

全体の流れは次のとおりです。

図1：Google CloudからAWSへデータを転送し、DatabricksのBronze層へ取り込む。

それぞれの役割を分けています。

この構成では、Databricksの入力をほかのデータソースと同じS3にそろえられます。GCSへの書き出しとクラウド間転送を分けることで、問題が起きた区間も調べやすくなります。ただし、BigQuery・GCS・S3それぞれの費用と、処理状況を追う運用が必要です。

2. 日次連携でも、昨日のデータだけを読まない

GA4のBigQuery日次エクスポートでは、events_YYYYMMDD というテーブルが作られます。遅れて到着するイベントを反映するため、日次テーブルはイベント発生日から最大3日後まで更新されます。

そこで、書き出しは初回と日次に分けています。

初回は、BigQueryに存在する対象期間のテーブルを日付ごとに書き出す。
日次は、実行日の前日から3日前までを、日付ごとに再出力する。

例えば11月5日の処理では、11月4日、3日、2日を対象にします。ここでいう日次の増分は、新しく追加された行だけを抽出する方式ではありません。対象日を直近に絞り、その日について取得時点の全行を書き出す方式です。

最終対象日の書き出し後にも更新は起こり得るため、3日分の再出力だけで遅延分をすべて回収できるとは限りません。再取得範囲は、実際の更新状況と照合して調整します。

初回も、対象はBigQueryに存在するデータです。連携開始前の期間を復元する処理は含みません。

3. 日付ごとにParquetへ書き出す

GCSには、対象日がパスから分かる形で配置します。以下は説明用の例です。

gs://<gcs-bucket>/events/
  dt=2026-11-04/events_20261104_*.parquet
  dt=2026-11-03/events_20261103_*.parquet
  dt=2026-11-02/events_20261102_*.parquet

日付で区切ると、取り込み漏れの調査や特定期間の再出力で対象を指定しやすくなります。形式はParquetとし、GA4の配列や構造体を保持して後段へ渡します。

BigQueryの EXPORT DATA を使った、1日分の書き出し例です。

EXPORT DATA OPTIONS (
  uri = 'gs://<gcs-bucket>/events/dt=2026-11-04/events_20261104_*.parquet',
  format = 'PARQUET',
  compression = 'SNAPPY',
  overwrite = true
) AS
SELECT *
FROM `<project>.<dataset>.events_20261104`;

書き出し前には、対象日のテーブルが存在するかを確認します。存在しない日をスキップした場合も分かるようにし、処理の成功と対象日の取得状況を合わせて確認します。

再出力時には、overwrite = true の動作にも注意が必要です。同じURIのファイルは上書きされますが、余った既存ファイルは自動削除されません。出力ファイル数が減ると旧ファイルが残る可能性があるため、読み込むファイル一式が今回の出力と一致するかを確認します。

4. GCSからS3への転送をDataSyncに任せる

GCSからS3への転送には、AWS DataSyncを使います。今回のTerraform実装ではEnhanced modeを指定し、転送用エージェントを置かない構成です。
DataSyncはGCSのXML APIを利用します。GCS側ではサービスアカウントに関連付けたHMACキーで認証し、対象バケットへの読み取り権限を付与します。AWS側では専用IAMロールに転送先S3へのアクセスを許可し、両クラウドの権限をそれぞれ設定します。

また、この経路ではオブジェクトタグの転送を無効にしています。GCSのXML APIがオブジェクトタグの読み書きに対応していないためです。

コードで管理する範囲は、DataSyncの転送元・転送先、タスク、スケジュールです。GA4とBigQueryのリンク設定、BigQuery側の定期クエリ設定、GCSのHMACキー作成は、現行の設計では手動設定の範囲にしています。特にHMACのSecretはリポジトリへ直接書かず、CIのSecret経由で渡す方針です。

5. Auto LoaderでBronze層へ取り込む

S3に届いたParquetファイルは、DatabricksのAuto LoaderでBronze層へ取り込みます。ここでは、元のイベント項目やネスト構造を保持し、取り込み時刻を付けて保存します。

以下は、Lakeflow Declarative Pipelinesのテーブル定義で使う読み込み部分の抜粋です。source_path と schema_path は環境ごとの設定から組み立て、この結果を返すテーブル定義内に配置します。

from pyspark.sql import functions as F

raw_events = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.schemaLocation", schema_path)
    .option("cloudFiles.includeExistingFiles", "true")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load(source_path)
)

bronze_events = raw_events.withColumn(
    "_ingest_timestamp", F.current_timestamp()
)

初回はS3上の既存ファイルも対象にし、その後は取り込み状態を引き継いで新しいファイルを読み込みます。パイプラインでは、この結果をBronzeテーブルとして定義し、イベント日でパーティションを分けています。

event_timestamp は、Google Analyticsがイベントを受信した時刻を表すUTCのマイクロ秒値です。Bronzeでは元の値を保持し、Databricksでの取り込み時刻を _ingest_timestamp として別に記録します。両者の差は、端末上の発生から取り込みまでの時間をそのまま表すものではありません。

5-1. スキーマ変更への対応

addNewColumns は、新しい列を検出すると保存済みスキーマを更新し、ストリームを UnknownFieldException で一度停止します。再起動後は更新されたスキーマで処理を続けます。既存列の型変更は対象外です。

Jobには再試行を1回設定しています。新しい列を含むファイルで、停止後の再起動と取り込み結果を確認する必要があります。

5-2. 再出力データの取り込み

現在の実装は、ファイルを読み込んでBronzeへ追加する構成です。日付ごとの最新スナップショットへ置き換える処理はなく、GCSへの再出力だけではBronzeの更新を保証できません。

Auto Loaderは、既定では上書きファイルの再処理を有効にしていません。cloudFiles.allowOverwrites を有効にした場合もファイル全体が再処理されるため、重複への対応が必要です。現行コードではこの設定を明示しておらず、再出力後の最新内容がBronzeにそろうことは未検証です。

実行ごとの出力先へ変更しないファイルを保存し、出力完了を確認して取り込む方式も考えられます。取得実行を識別する情報に加え、複数回取得したイベントから最新状態を選ぶ規則も必要です。

おわりに

BigQueryで書き出し、DataSyncで転送し、Auto Loaderで取り込む。この役割分担で、GA4を既存のS3・Databricks基盤へ接続しています。