WED株式会社でデータエンジニアをしているthimi0412です。
以前、こちらの記事でDatastream for BigQueryを使用し、アプリケーションで使用しているデータベースから BigQuery へデータを転送する構成を紹介しました。
しかし、2025年4月をもって Datastream の利用を停止し、別の方法に移行しました。本記事ではその背景と、移行後の構成についてまとめます。
辞めた理由
単刀直入に言って、理由は「コスト」です。
Datastream for BigQuery は、データベースのイベントを検知して、対応する BigQuery のテーブルに対して INSERT / UPDATE / DELETE を実行します。
その際にbigquery-adminbot@system.gserviceaccount.comというサービスアカウントを介して以下のようなSQLが実行されます。
CALL BQ.APPLY_UPSERT_STREAM('<project_id>.<dataset>.<table>')
当然、このクエリの実行にもコストが発生します。
WEDが運営するレシート買取アプリONEでは1日で約100万枚のレシート画像を買い取っており、データベース上では insert や update が絶え間なく発生しています。
その結果、BigQuery側の実行コスト増加してしまいました。
新たな転送方法
コスト削減を主な目的として、以下の観点から転送方法を再設計しました。
- リアルタイムに送る必要はない(コストの観点から)
- 可能な限りマネージドな構成にしたい
上記の点からAlloyDBのfederated queryを使用したデータの転送を使おうと考えました。 元々Datastream for BigQueryを使うことに至ったのも、アプリケーションで使用しているデータベースをCloudSQLからAlloyDBへ移行する一環で行われたものでした。
もともとDatastreamを採用した背景にはCloudSQLからAlloyDBへの移行がありました。ただし、当時は Federated Query が発表されておらず、見送っていました。
転送の流れ
以下のような流れでBigQueryへのデータの転送を行なっています。
- 対象のテーブルから1日分のデータを取得してlogテーブルを作成
- データを蓄積しているテーブルに対してlogテーブルをmerge
上記のワークフローをCloudComposer(Airflow)で全てのテーブルのデータを転送しています。
データの転送が終わり次第データマート作成のDAGをトリガーしています。
サンプル
CREATE OR REPLACE TABLE raw.users_log
AS
SELECT
id,
name
FROM
EXTERNAL_QUERY(
"<connection_name>",
"""
SELECT
id::text AS id,
name
FROM
users
WHERE
(updated_at AT TIME ZONE 'UTC' AT TIME ZONE 'Asia/Tokyo')::DATE = '2025-04-17'
"""
);
上記のクエリは各テーブルごとに動的に作成していて、PostgreSQLの型からBigQueryへの方の変換を行なっています。詳しくは以下の記事で解説しています。
Datastream for BigQueryの良さもある
今回のように、ONEのような大量のデータ更新があるサービスではコスト面で見直しが必要でしたが、更新頻度がそこまで高くないサービスであれば、Datastream for BigQuery は非常に便利な選択肢だと感じました。
- テーブルやカラムの追加時に、アプリケーションとデータ基盤側の連携がシームレスになる
- アプリケーション側の変更が即座にBigQueryに反映されるため、コミュニケーションコストが下がる
といったメリットがあります。
特に、データエンジニアがまだいないフェーズのチームや、分析基盤の立ち上げ初期段階では非常に有効だと思います。
まとめ
Datastream for BigQuery をやめ、AlloyDB Federated Query を使った新たな構成に移行した背景と構成について紹介しました。
今後もこれらの課題に向き合いながら、より安定したデータ基盤を目指して改善を続けていきます。
Discussion