はじめに

BigQuery Data Transfer Service（以下、DTS）は、Google広告やGA4などのデータを定期的にBigQueryへ取り込んでくれる、とても便利な仕組みです。一度設定してしまえば、毎日決まった時間に自動でデータが流れ込んでくるので、手作業でのエクスポートから解放されます。

ところが、この「自動で動いてくれる」という性質が、ときに落とし穴になります。転送が失敗していても、誰も気づかないまま日々が過ぎてしまうのです。

たとえば、認証トークンの期限切れ、連携元APIの一時的な障害、スキーマの変更といった理由で、特定の日の転送だけがこっそり失敗していることがあります。ダッシュボードを見ても、昨日までのデータは揃っているので一見正常に見えます。けれど、よく確認すると「ある日からデータがぱったり止まっている」のです。

このデータ欠損に気づくのが月末の集計時だったりすると、目も当てられません。欠損した期間のデータは後から取り直せないこともあり、レポートの数字そのものが信頼できなくなります。

特に困るのは、欠損に気づかないまま、その不完全なデータを使って意思決定をしてしまうケースです。たとえば広告のパフォーマンスが急に落ちたように見えても、それは実際の不調ではなく、単に転送が止まっていただけだった、ということが起こり得ます。データを扱う仕事では、「数字が間違っている」よりも「間違っていることに気づけない」ほうが、はるかに怖いのです。

こうした事故を防ぐには、「転送が失敗したら、すぐ人間に知らせる」仕組みをあらかじめ用意しておくのが確実です。本記事では、DTSの転送エラーを自動で検知して、Slackへ通知する方法を整理します。難しい構成も紹介しますが、まずは小さく始められる方法から押さえていきます。

なお、DTSのセットアップそのものについては別記事の「Google広告のデータをBigQuery Data Transfer Serviceで取り込む」で扱っているので、まだ転送設定が済んでいない方はそちらも参考にしてください。

転送状態の確認方法

通知の仕組みを作る前に、まずは「転送が成功したか失敗したか」をどこで確認できるのかを押さえておきます。検知の仕組みは、結局のところこれらの情報源を機械的に監視するものだからです。

Google Cloudコンソールで確認する

もっとも手軽なのは、Google CloudコンソールのBigQueryページにある「データ転送」画面です。ここには設定済みの転送構成が一覧で並び、それぞれの直近の実行結果（成功・失敗・実行中）が表示されます。

個別の転送構成をクリックすると、過去の実行履歴（実行ごとのRun）が時系列で並びます。失敗したRunを開けば、エラーメッセージや詳細なログを確認できます。原因の切り分けには、まずここを見るのが基本です。

ただし、コンソールは「人が見に行って初めて分かる」場所です。毎朝かならず確認する運用ができるなら問題ありませんが、現実にはなかなか続きません。自動検知が必要になるのは、まさにこの「見に行かないと分からない」を解消するためです。

ログで確認する

DTSの実行結果は、Cloud Loggingにもログとして記録されます。コンソールのログエクスプローラから、転送サービスに関するログを絞り込んで確認できます。

ログには転送の開始・完了・失敗といったイベントが残るため、「いつ、どの転送が、どういう結果だったか」を機械的に読み取る土台になります。後述する自動検知の仕組みは、このログを起点にするのがひとつの王道です。

コンソールが「人が見に行く場所」だとすれば、ログは「プログラムが見に行ける場所」だと考えると分かりやすいかもしれません。同じ転送結果でも、機械が処理しやすい形で残っているからこそ、自動通知の入り口として使えるわけです。

ログの正確なリソース種別やフィルタ条件は変わることがあるため、実装時には最新の公式ドキュメントでログの構造を確認してください。

検知の仕組み

転送状態の確認方法が分かったところで、本題の「自動検知」をどう組むかを考えます。大きく分けて、二つのアプローチがあります。

アプローチ1: ログをトリガーにする

ひとつめは、Cloud Loggingに記録される転送結果のログをトリガーにする方法です。流れとしては次のようになります。

  1. Cloud Loggingで「転送が失敗したことを示すログ」だけを抽出するシンクを作る
  2. そのシンクの送り先をPub/Subトピックに設定する
  3. Pub/SubトピックをトリガーにCloud Functionsを起動する
  4. Cloud Functionsが失敗内容を整形してSlackへ通知する

この方式の利点は、失敗が起きた瞬間に近いタイミングで通知が飛ぶことです。ログが出た直後に処理が走るため、検知の遅れがほとんどありません。

一方で、構成要素が少し多くなります。ログシンク、Pub/Sub、Cloud Functionsと、複数のサービスをつなぐ必要があるため、最初のセットアップにはやや手間がかかります。それでも、一度組んでしまえば手放しで動いてくれるので、本格的に運用するならこちらをおすすめします。

なお、DTS自体にもメール通知の設定項目があります。まずは公式の通知機能で最低限のアラートを確保しておき、Slack連携はその上に重ねる、という考え方も実用的です。

アプローチ2: 定期チェッククエリで確認する

ふたつめは、もっとシンプルに「定期的にデータの状態を確認するクエリ」を回す方法です。

DTSが書き込む先のテーブルに対して、「直近のデータがちゃんと入っているか」を問い合わせるクエリを用意します。たとえば「今日の日付のレコードが1件もなければ異常」とみなす、といった具合です。このチェッククエリをスケジュールしておき、異常を検知したら通知する形にします。

この方式は、転送の仕組みそのものを監視するというより「結果としてデータが入ったか」を監視する発想です。ログ連携ほど即時性はありませんが、構成がシンプルで理解しやすく、「最終的に欲しいデータが揃っているか」という一番大事な点を直接確認できる強みがあります。

データの状態を監視するという意味では、コストや件数を継続的に見張る「INFORMATION_SCHEMAでBigQueryのコストを監視する」の考え方とも相性が良いです。監視クエリの土台として併せて参考にしてください。

下のクエリは、注文データの転送先テーブルに当日分のレコードが入っているかを確認する例です。ここでは、テーブルが持つ実際の日付列（例として order_date）の最大値を見て、最新データの日付が今日かどうかを判定しています。

SELECT
  MAX(order_date) AS latest_data_date
FROM
  `your_project.your_dataset.transferred_orders`;

この latest_data_date が今日の日付（CURRENT_DATE('Asia/Tokyo')）より前であれば、「今日の転送が来ていない可能性が高い」と判断できます。あとは、この判定結果を通知につなげるだけです。

ここで使う日付列の名前は、取り込み方法やテーブルのスキーマによって異なる点に注意してください。たとえばGoogle広告のDTSなら _DATA_DATE、GA4由来のデータなら event_date、自前で日付列を持たせている場合はその列名、という具合です。実装前に、対象テーブルのスキーマを開いて「データの鮮度を表す列」がどれなのかを必ず確認してください。

なお、適切な日付列が見当たらない場合や、テーブルが日付でパーティション分割されている場合は、パーティション擬似列 _PARTITIONTIME を使って最新パーティションの時刻を確認する方法もあります。

SELECT
  MAX(_PARTITIONTIME) AS latest_partition_time
FROM
  `your_project.your_dataset.transferred_orders`;

データの中身ではなく転送そのものの成否を直接見たい場合は、前述のCloud Logging（ログ）やコンソールの実行履歴で転送実行の状態を確認するアプローチが確実です。チェックの目的に合わせて使い分けてください。

なお、転送のスケジュールによっては「まだ今日の分が来ていないだけ」という正常な状態もあり得ます。チェックを走らせる時刻は、転送が完了するはずの時刻より十分あとに設定しておくと、無用な誤報を減らせます。

Slack Webhook通知

検知できたら、最後はそれをSlackへ届けます。Slackへの通知は、Incoming Webhookを使うのがもっとも手軽です。Slack側でWebhook URLを発行し、そのURLにメッセージ本文をPOSTするだけで、指定したチャンネルに投稿できます。

Webhookの発行手順や利用できるメッセージ形式は変わることがあるため、設定時には最新の公式ドキュメントを確認してください。

Cloud FunctionsからWebhookへ通知する処理は、Pythonならおおむね次のような形になります。実際のWebhook URLはコードに直書きせず、環境変数やSecret Managerから読み込むようにしてください。

import json
import os
import urllib.request


def notify_slack(message: str) -> None:
    webhook_url = os.environ["SLACK_WEBHOOK_URL"]
    payload = {"text": message}
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        webhook_url,
        data=data,
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req) as res:
        if res.status != 200:
            raise RuntimeError(f"Slack通知に失敗しました: {res.status}")


def handle_transfer_failure(transfer_name: str, error_detail: str) -> None:
    text = (
        f":rotating_light: BigQuery転送が失敗しました\n"
        f"・転送構成: {transfer_name}\n"
        f"・内容: {error_detail}"
    )
    notify_slack(text)

ポイントは、通知メッセージに「どの転送が」「いつ」「どんな理由で」失敗したのかを盛り込んでおくことです。通知を受け取った人が、メッセージを読んだだけで状況を把握し、すぐコンソールの該当箇所へ飛べるようにしておくと、対応がぐっと速くなります。

WebhookのURLは、それ自体が投稿権限を持つ秘密情報です。ソースコードやリポジトリに直接書き込まず、必ず安全な場所で管理してください。万が一URLが漏れると、第三者が勝手にチャンネルへ投稿できてしまうため、扱いは認証情報と同じ慎重さが求められます。

⚠️ 通知過多を避ける

監視の仕組みを作るときに、意外と見落としがちなのが「通知が多すぎる問題」です。

良かれと思って細かく通知を仕込むと、ちょっとした再試行や一時的なエラーまで毎回飛んでくるようになります。最初のうちは真剣に見ていても、通知が鳴りやまない状態が続くと、人はだんだん見なくなります。そうなると、本当に大事な失敗通知まで見過ごしてしまい、監視の意味がなくなってしまいます。

通知過多を避けるために、次のような工夫が有効です。

  • 一時的なエラーと致命的なエラーを区別し、再試行で回復するものは通知しない
  • 同じ失敗が連続したときは、毎回ではなくまとめて1回だけ通知する
  • 通知の宛先を、本当に対応できる人がいるチャンネルに絞る
  • 「成功通知」は基本的に送らず、異常時だけ知らせる

監視の目的は「異常に気づくこと」であって「通知を浴びること」ではありません。通知は少なく、けれど見逃さない。このバランスを意識して設計すると、長く使える仕組みになります。

通知が形骸化していないかを定期的に振り返るのも大切です。「最近この通知、ちゃんと反応できているか」を時々見直し、不要になった通知は思い切って止める。こうした手入れを続けることで、監視は生きた仕組みであり続けます。

最初から完璧を目指す必要はありません。まずは「失敗したら1回Slackに飛ぶ」という最小構成から始めて、運用しながら通知のしきい値や宛先を調整していくのが現実的です。

まとめ

DTSは設定すれば自動で動いてくれる頼もしい仕組みですが、その「自動さ」ゆえに、失敗してもしばらく気づけないという弱点を抱えています。データ欠損に月末の集計で気づくような事故を防ぐには、転送エラーを自動で検知して人に知らせる仕組みが欠かせません。

本記事では、転送状態の確認方法として、コンソールとログという二つの情報源を確認しました。そのうえで、検知の仕組みを「ログをトリガーにPub/SubとCloud Functionsで即時通知する方法」と「定期チェッククエリでデータの有無を確認する方法」の二つに整理し、最後にSlackへWebhookで通知する流れを見てきました。

大切なのは、即時性と運用のしやすさのどちらを重視するかで方式を選ぶことと、通知過多に陥らないように設計することです。

まずは小さく始めて、自分たちの運用に合った形へ育てていけば、データの信頼性を守る心強い仕組みになります。データは「正しく入っていること」が前提です。その前提を静かに支えてくれるのが、こうした監視の仕組みなのだと思います。