C データエンジニアリング — 在庫リアルタイムマート Datastream CDC（ソースタイムスタンプ+binlog位置での順序解決 × 削除イベント(_metadata_deleted)反映 × incremental merge戦略 × パーティション/クラスタリングでスキャンコスト削減 × on_schema_changeでスキーマ追従 × レプリケーション遅延のCloud Monitoring監視）（MOps 倉庫在庫管理サービス 欠品アラートLINE通知 Bad→Good）

概要

問題

ECサイト MOps チームでは、倉庫在庫管理サービス（Cloud SQL for MySQL 8.0）の在庫テーブル inventory.stock（product_id, warehouse_id, quantity, reserved_quantity, updated_at）を Datastream で BigQuery にCDC（Change Data Capture）レプリケーションし、cdc_raw.inventory_changelog という変更履歴（changelog）テーブルに追記している。この変更履歴を dbt Core 1.9+ でリアルタイムに近い「現在庫数マート」（mart_inventory_realtime）に変換し、倉庫担当者向けの欠品アラートLINE通知（在庫が閾値を下回ったら自動通知）の判定に使っている。

以下の Terraform（Datastream構成）・dbtモデル・schema.yml には 7つの設計・データ品質上の問題 が潜んでいる。問題点を全て洗い出し、Datastream変更履歴の正しい順序解決（source timestamp + binlog position）× 削除イベント（_metadata_deleted）の反映 × dbt incremental（merge戦略）× パーティション/クラスタリングによるスキャンコスト削減 × ソーススキーマ変更への追従（on_schema_change）× Datastreamレプリケーション遅延の監視アラート を活用した Bad→Good リファクタリングを行ってください。

制約・前提条件

• Datastream（Cloud SQL for MySQL → BigQuery、GA機能。append_only モードで変更履歴をそのまま追記）
• BigQuery Standard SQL、dbt Core 1.9+（incremental_strategy='merge', on_schema_change, model contract が利用可）
• Terraform 1.8+（google プロバイダー 5.x）
• cdc_raw.inventory_changelog のスキーマ: product_id, warehouse_id, quantity, reserved_quantity, updated_at（ソース側の更新時刻）に加え、Datastreamが自動付与するメタデータ列 _metadata_source_timestamp（ソースDBでのコミット時刻）, _metadata_log_file/_metadata_log_position（MySQL binlogのファイル名・位置。同一秒内の複数変更の順序を一意に決定できる）, _metadata_deleted（削除イベントなら true）
• Datastreamは変更を1行も更新・削除せず、全て新しい行として追記し続ける（append_only モード）。同一 (product_id, warehouse_id) の変更履歴が複数行たまる
• reserved_quantity 列は最近、在庫管理サービス側チームがMOpsチームへの事前連携なしに追加した新しいカラム
• 欠品アラートLINE通知は mart_inventory_realtime を5分ごとに参照して実行される。データが古い・不正確だと誤通知（在庫があるのに欠品通知/欠品なのに通知なし）につながる

悪いコード (Before)

📋 コピー{{ config(materialized='table') }}
-- 問題③: 変更履歴全体を毎回フル再構築

-- 問題①: アプリ側のupdated_atでソート
--        （クロックずれ・再送で順序が狂う）
-- 問題②: _metadata_deletedを一切フィルタしていない
--        （削除済みの在庫がそのまま残る）
SELECT
    product_id,
    warehouse_id,
    quantity,
    reserved_quantity,
    updated_at
FROM `my-project.cdc_raw.inventory_changelog`
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY product_id, warehouse_id
    ORDER BY updated_at DESC   -- 問題①
) = 1
-- 問題④: 重複排除の粒度がQUALIFYの1回のみで
--        mergeによる冪等性の担保が無い
-- 問題⑤: WHERE句が無く、変更履歴テーブル
--        全体をスキャンしている
-- 問題⑥: reserved_quantity列が将来変更されても
--        検知できない（on_schema_change未設定）

📋 コピーresource "google_datastream_stream" "inventory_cdc" {
  stream_id     = "inventory-cdc-stream"
  location      = "asia-northeast1"
  desired_state = "RUNNING"

  source_config {
    source_connection_profile =
      google_datastream_connection_profile.mysql_source.id
    mysql_source_config {
      include_objects {
        mysql_databases {
          database = "inventory"
          mysql_tables { table = "stock" }
        }
      }
    }
  }

  destination_config {
    destination_connection_profile =
      google_datastream_connection_profile.bq_dest.id
    bigquery_destination_config {
      single_target_dataset { dataset_id = "cdc_raw" }
      append_only {}
      data_freshness = "900s"
    }
  }
}
# 問題⑦: レプリケーション遅延・ストリーム停止を
#        検知するCloud Monitoringアラートが無い

ヒント（段階的開示）

-- 最新版の解決（骨格）
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY product_id, warehouse_id
  ORDER BY _metadata_source_timestamp DESC,
           _metadata_log_file DESC,
           _metadata_log_position DESC
) = 1

# dbt incremental設定の骨格
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key=['product_id', 'warehouse_id'],
    partition_by={'field': 'change_date', 'data_type': 'date'},
    cluster_by=['product_id', 'warehouse_id'],
    on_schema_change='append_new_columns'
) }}

問題点分析（7点）

CDC順序解決の仕組み — Bad vs Good の比較

模範解答

📋 コピー# infra/datastream_governance.tf

resource "google_datastream_stream" "inventory_cdc" {
  stream_id     = "inventory-cdc-stream"
  location      = "asia-northeast1"
  desired_state = "RUNNING"

  source_config {
    source_connection_profile = google_datastream_connection_profile.mysql_source.id
    mysql_source_config {
      include_objects {
        mysql_databases {
          database = "inventory"
          mysql_tables { table = "stock" }
        }
      }
    }
  }

  destination_config {
    destination_connection_profile = google_datastream_connection_profile.bq_dest.id
    bigquery_destination_config {
      single_target_dataset { dataset_id = "cdc_raw" }
      append_only {}
      data_freshness = "300s"   # 欠品アラート通知の5分周期に合わせSLA明示
    }
  }
}

# 修正⑦: Datastreamのレプリケーション遅延を監視し、閾値超過でアラート
resource "google_monitoring_alert_policy" "datastream_latency" {
  project      = var.project_id
  display_name = "inventory-cdc-stream レプリケーション遅延超過"
  combiner     = "OR"
  conditions {
    display_name = "total_latencies が 600秒（欠品アラートSLAの2倍）を超過"
    condition_threshold {
      filter          = <<-EOT
        resource.type="datastream.googleapis.com/Stream"
        resource.labels.stream_id="inventory-cdc-stream"
        metric.type="datastream.googleapis.com/stream/total_latencies"
      EOT
      comparison      = "COMPARISON_GT"
      threshold_value = 600
      duration        = "120s"
      aggregations {
        alignment_period   = "300s"
        per_series_aligner = "ALIGN_PERCENTILE_99"
      }
    }
  }
  notification_channels = [var.oncall_notification_channel_id]
}

# 修正⑦: ストリーム自体が異常停止した場合の検知
resource "google_monitoring_alert_policy" "datastream_stream_down" {
  project      = var.project_id
  display_name = "inventory-cdc-stream ストリーム異常停止"
  combiner     = "OR"
  conditions {
    display_name = "イベント流入が10分間停止"
    condition_threshold {
      filter          = <<-EOT
        resource.type="datastream.googleapis.com/Stream"
        resource.labels.stream_id="inventory-cdc-stream"
        metric.type="datastream.googleapis.com/stream/event_count"
      EOT
      comparison      = "COMPARISON_LT"
      threshold_value = 1
      duration        = "600s"
    }
  }
  notification_channels = [var.oncall_notification_channel_id]
}

📋 コピー{{ config(materialized='table') }}

SELECT
    product_id, warehouse_id, quantity,
    reserved_quantity, updated_at
FROM `my-project.cdc_raw.inventory_changelog`
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY product_id, warehouse_id
    ORDER BY updated_at DESC
) = 1
-- 削除フィルタなし・パーティション絞り込みなし
-- on_schema_changeなし

📋 コピー-- models/staging/stg_inventory_changelog.sql
-- 修正③⑤: incremental化しパーティション+クラスタリングで
--         スキャン範囲を「新しく追記された変更履歴」に限定
{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key=['product_id', 'warehouse_id'],
    partition_by={'field': 'change_date', 'data_type': 'date',
                  'granularity': 'day'},
    cluster_by=['product_id', 'warehouse_id'],
    on_schema_change='append_new_columns'   -- 修正⑥
) }}

WITH changelog AS (
    SELECT
        product_id, warehouse_id, quantity, reserved_quantity,
        _metadata_source_timestamp, _metadata_log_file,
        _metadata_log_position, _metadata_deleted,
        DATE(_metadata_source_timestamp) AS change_date
    FROM {{ source('cdc_raw', 'inventory_changelog') }}
    {% if is_incremental() %}
    -- 修正⑤: 前回実行以降のパーティションのみスキャン
    WHERE DATE(_metadata_source_timestamp) >= (
        SELECT DATE_SUB(MAX(change_date), INTERVAL 1 DAY) FROM {{ this }}
    )
    {% endif %}
),
latest_per_key AS (
    -- 修正①: ソースDBのコミット時刻+binlog位置で
    --        実際の発生順序を一意に解決する
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY product_id, warehouse_id
            ORDER BY _metadata_source_timestamp DESC,
                     _metadata_log_file DESC,
                     _metadata_log_position DESC
        ) AS rn
    FROM changelog
)
SELECT
    product_id, warehouse_id, quantity, reserved_quantity,
    _metadata_source_timestamp AS last_changed_at, change_date
FROM latest_per_key
WHERE rn = 1
  AND _metadata_deleted = FALSE   -- 修正②: 削除イベントを除外

📋 コピー{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key=['product_id', 'warehouse_id'],
    partition_by={'field': 'change_date', 'data_type': 'date', 'granularity': 'day'},
    cluster_by=['product_id', 'warehouse_id']
) }}

SELECT
    product_id,
    warehouse_id,
    quantity - reserved_quantity AS available_quantity,  -- 欠品判定に使う実質在庫数
    quantity,
    reserved_quantity,
    last_changed_at,
    change_date
FROM {{ ref('stg_inventory_changelog') }}
{% if is_incremental() %}
WHERE change_date >= (SELECT DATE_SUB(MAX(change_date), INTERVAL 1 DAY) FROM {{ this }})
{% endif %}

📋 コピー# models/staging/_sources.yml
version: 2

sources:
  - name: cdc_raw
    database: my-project
    schema: cdc_raw
    tables:
      - name: inventory_changelog
        description: "Datastream (Cloud SQL for MySQL → BigQuery) による在庫変更履歴。append_onlyモードで全変更を追記。"
        loaded_at_field: _metadata_source_timestamp
        freshness:
          warn_after: { count: 10, period: minute }
          error_after: { count: 20, period: minute }   # 欠品アラート5分周期に対する遅延許容の上限

# models/marts/_marts.yml
models:
  - name: mart_inventory_realtime
    description: "商品×倉庫の現在庫数マート。欠品アラートLINE通知の判定に使用。"
    config:
      contract:
        enforced: true
    columns:
      - name: product_id
        data_type: string
        constraints: [{ type: not_null }]
      - name: warehouse_id
        data_type: string
        constraints: [{ type: not_null }]
      - name: available_quantity
        data_type: int64
      - name: last_changed_at
        data_type: timestamp
    tests:
      - dbt_utils.unique_combination_of_columns:   # 修正④の検証: 重複キーが残っていないか
          combination_of_columns: [product_id, warehouse_id]

実行例（input → output）

input（cdc_raw.inventory_changelog、商品P100・倉庫W1の変更履歴、簡略化）:

product_id  warehouse_id  quantity  updated_at(app側)         _metadata_source_timestamp  _metadata_log_file  _metadata_log_position  _metadata_deleted
P100        W1            50        10:00:00                   10:00:00.100                binlog.000123       4820                     false
P100        W1            30        10:00:00 (再送で同一値!)   10:00:00.310                binlog.000123       4915                     false
P100        W1            NULL      10:00:01                   10:00:00.900                binlog.000123       4860                     true   (削除イベント。DB上では上記2件より先にcommitされていたが到着が遅れた)

output（Before・問題のある実装）:

updated_at でソートすると quantity=30（アプリ側時刻の粒度・再送で到着順が不安定）と削除イベントの前後関係が崩れ、削除フィルタも無いため quantity=30 が「現在庫」として誤って採用され続けるリスクがある。

output（After・改善後）:

_metadata_log_position（binlog位置）でソートすると、ソースDB内の絶対順序は 4820 → 4860(削除) → 4915 と確定する。最新版は log_position=4915（quantity=30、_metadata_deleted=false）となり、正しく「現在庫30個」として mart_inventory_realtime に反映される。仮にDELETEが最後に発生していた場合は _metadata_deleted=FALSE フィルタにより当該商品はマートから正しく除外される。

ポイント解説

実務への応用

ECサイト MOps チームが扱う在庫データは、倉庫管理サービス・注文管理サービス・返品サービスなど複数のマイクロサービスにまたがって更新される。CDCでリアルタイムに近い連携をする設計は、バッチ集計に比べて鮮度は高いが、その分「変更履歴から正しい状態をどう組み立てるか」という設計の巧拙が直接ビジネス影響（誤った欠品通知・過剰発注・機会損失）に跳ね返る。特に本問のような「アラート・自動発注のトリガーとして使われるデータ」は、鮮度だけでなく正確性（重複排除・削除反映・順序保証）の3点セットが揃って初めて信頼できるデータ基盤になる。

今日のまとめ

次のステップ

• 発展問題: 現在の設計は「5分ごとにバッチでmartを再計算する」擬似リアルタイムだが、真にリアルタイムな欠品アラートが必要な場合、Datastream → Pub/Sub → Dataflow（Streaming）で直接ストリーム処理し、BigQueryへの書き込みを介さずにアラート判定する設計に拡張するとどう変わるか、コスト・複雑性とのトレードオフを整理せよ。また、_metadata_log_file/_metadata_log_position はMySQL固有のメタデータだが、ソースがPostgreSQLの場合（LSN: Log Sequence Number）ではどう読み替えるべきかを調べよ。
• 参考: Datastream（Cloud SQL for MySQL → BigQuery, append_only モード）公式ドキュメント / dbt incremental_strategy: merge / dbt on_schema_change / dbt source freshness / BigQuery パーティション分割テーブル・クラスタ化テーブル / Change Data Capture（CDC）設計パターン

自己評価（あとで記入）