CDCとイベント時刻｜データエンジニアリング実践

このページはプロモーション（広告）を含みます。

データエンジニアリング実践（中級・読む30分＋演習40分）

この授業のゴール

• 全件スナップショット・更新時刻による差分抽出・ログベースCDCの違いを、取りこぼす変更の種類で説明できる
• CDCの変更ログを、到着順ではなく元DBでの確定順（LSN）で適用し、重複・順序の入れ替わり・削除に強い最新状態を作れる
• イベント時刻・確定時刻・取込時刻を区別し、「ある時点の状態」を確定時刻で再現できる
• 2つのスナップショットの差から追加・更新・削除を検出し、その方法の限界を説明できる

仕事でどう使うか

購入イベントのように「起きたことを送ってもらう」データだけでなく、業務システムのデータベース（注文表・顧客表など）の変化を分析基盤へ写したい場面がよくあります。返金で状態が変わった注文、削除された顧客、修正された金額。これらを正しく写せないと、分析基盤の売上は業務システムとずれ続けます。

この授業では、注文表の変更を写す方法を3つ比べ、最も取りこぼしの少ないCDC（Change Data Capture：変更データの捕捉）の変更ログを、重複や順序の入れ替わりがあっても正しく適用する方法を学びます。

前提

次の言葉を使います。不安があれば先に戻ってください。

• 取込契約・キー・配送保証 → 取込契約と粒度
• イベントと状態（イベントを順に適用すると状態ができる）→ 「イベントデータ」（有料版）
• 履歴と有効期間（今の値とそのときの値）→ 「履歴と変更」（有料版）
• トランザクションとコミット（変更が確定する単位）→ 「BEGIN・COMMIT・ROLLBACK」（有料版）、冪等性は 「冪等性」（有料版）
• LEFT JOIN と NULL の比較 → 「LEFT JOINと0件」（有料版）、NULLと3値論理

時刻はすべてUTC（協定世界時）の YYYY-MM-DDTHH:MM:SSZ 形式の文字列で扱います。同じ形式なら文字列の大小比較で前後を判定できます。

身近な状況：夜に1回コピーするだけでは、返金が見えない

そらマートの業務システムには注文表があり、分析チームは毎晩0時に注文表を丸ごとコピー（全件スナップショット）していました。5月1日の注文 o1 の動きは次のとおりです。

夜間コピーには「最後の状態」しか写りません。「5/1 12時時点では1,000円の支払済みだった」「1日のうちに一度は売上だった」ことは、コピーからは復元できません。さらに、注文 o2 が業務システムから削除されると、コピーの差を取らない限り分析基盤には o2 が残り続けます。

別の担当者は「更新時刻 updated_at が前回より新しい行だけ取る」差分抽出を提案しました。これなら速いのですが、削除された行は表に存在しないので取れません。また、長いトランザクションの途中で書かれた行は、updated_at が古いまま後から確定するため、取りこぼすことがあります。

業務システムのデータベースは、すべての変更を確定順にトランザクションログ（変更の記録）へ書いています。このログを読んで変更を1件ずつイベントとして送るのがログベースCDCです。

直感的な説明：写真と、出入りの記録簿

3つの方法を、ビルの入館管理にたとえます。

• 全件スナップショット：毎晩0時に、ビル内にいる人の写真を撮る。昼に来て夕方に帰った人は写らない。
• 更新時刻による差分抽出：「前回以降に名札を書き換えた人」を探す。帰った人（削除）は名札ごと消えているので見つからない。
• ログベースCDC：受付の出入り記録簿を順に読む。入館・名札の書き換え・退館がすべて、起きた順に番号つきで残っている。

記録簿の番号が LSN（Log Sequence Number：ログの通し番号）です。記録簿が分析基盤へ届く途中で、同じ行が2回届いたり（再送）、順番が入れ替わったりすることがあります。それでも番号を見れば本当の順序がわかるので、「今持っている番号より小さい記録は古いので無視する」という規則だけで、正しい最新状態を作れます。

たとえ話の限界：入館記録と違い、データベースの変更は複数の行をまとめて確定（コミット）することがあります。1つのトランザクションの中の変更を途中まで適用すると、業務上ありえない状態（注文はあるのに明細がない等）が一時的に見えます。厳密に扱う方法は発展で触れます。

正確な定義：CDCの変更レコードと、3つの時刻

CDC（Change Data Capture）は、データベースの行の追加・更新・削除を、確定した順に変更レコードとして取り出す仕組みです。変更レコードは一般に次の項目を持ちます（名前は製品により異なります）。

3つの時刻を区別します。

変更の適用規則（キーごとに最新を保つ）：

1. 変更レコードの主キーで、現在の状態を探す。
2. 変更の LSN が、現在の状態の LSN 以下なら無視する（重複・古い変更）。
3. op が c か u なら after の値で置き換え、LSN を記録する。
4. op が d なら、行を消すのではなく 墓標（tombstone：削除済みの印）を LSN とともに残す。墓標がないと、後から遅れて届いた古い更新で、削除した行が復活してしまいます。

この規則は冪等（何回適用しても1回と同じ結果）で、到着順にも依存しません。

時点の状態：時刻 $T$ の状態は、commit_ts が $T$ 以下の変更だけに上の規則を適用すれば得られます。数式で書くと、キー $k$ の時点 $T$ の値は

$$\text{state}_T(k) = \text{after}\left(\arg\max_{c \in C_k,\ \text{commit\_ts}(c) \le T} \text{lsn}(c)\right)$$

です。$C_k$ はキー $k$ の変更レコードの集合、$\arg\max$ は「LSN が最大になる変更を選ぶ」という意味です。o1 で $T$＝5/1 12:00 なら、候補は LSN 101（10:00）と 105（10:30）で、最大の105の値「paid・1,000円」が答えです。

段階付き例題：乱れて届いた変更ログから状態を作る

入力：分析基盤に届いた変更ログ（到着順）。4番目の LSN 105 は遅れて届き、5番目は LSN 103 の再送です。

ステップ1：1, 2 で o1（LSN 101）、o2（LSN 103）ができる。

ステップ2：3 で o1 を refunded に（LSN 107）。

ステップ3：4 は LSN 105 ≤ 107 なので無視。到着順に上書きすると、返金済みの注文が支払済みに戻ってしまいます。

ステップ4：5 は LSN 103 ≤ 103 なので無視（重複）。

ステップ5：6 で o2 を墓標にする（LSN 109）。

出力と意味：最新状態では支払済みの注文はなく、支払済み合計は0円です。5/1 12:00 時点の状態を作ると、o1 は1,000円・paid、o2 は800円・paid で、合計 1,800円 です。同じログから「今」と「ある時点」の両方を再現できることが、CDCを確定時刻つきで保存する価値です。

変更レコードの after は「その時点の行全体」を表すので、返金のレコードにも修正後の金額1,000円が入っています。

# CDCの変更ログ（到着順）。lsn は元DBでの確定順を表す番号
changes = [
    {"lsn": 101, "op": "c", "order_id": "o1", "after": {"status": "paid", "amount": 1200}, "commit_ts": "2026-05-01T10:00:00Z"},
    {"lsn": 103, "op": "c", "order_id": "o2", "after": {"status": "paid", "amount": 800},  "commit_ts": "2026-05-01T10:02:00Z"},
    {"lsn": 107, "op": "u", "order_id": "o1", "after": {"status": "refunded", "amount": 1000}, "commit_ts": "2026-05-01T23:50:00Z"},
    {"lsn": 105, "op": "u", "order_id": "o1", "after": {"status": "paid", "amount": 1000}, "commit_ts": "2026-05-01T10:30:00Z"},
    {"lsn": 103, "op": "c", "order_id": "o2", "after": {"status": "paid", "amount": 800},  "commit_ts": "2026-05-01T10:02:00Z"},
    {"lsn": 109, "op": "d", "order_id": "o2", "after": None, "commit_ts": "2026-05-02T00:10:00Z"},
]

def apply_changes(changes, until=None):
    """lsn の大きい変更だけを採用して、注文ごとの最新状態を作る。until を渡すとその時刻までの状態"""
    state = {}
    for ch in changes:
        if until is not None and ch["commit_ts"] > until:
            continue
        cur = state.get(ch["order_id"])
        if cur is not None and ch["lsn"] <= cur["_lsn"]:
            continue                      # 古い変更・重複は無視
        if ch["op"] == "d":
            state[ch["order_id"]] = {"_deleted": True, "_lsn": ch["lsn"]}
        else:
            state[ch["order_id"]] = {**ch["after"], "_deleted": False, "_lsn": ch["lsn"]}
    return state

now = apply_changes(changes)
noon = apply_changes(changes, until="2026-05-01T12:00:00Z")
paid = lambda st: sum(v["amount"] for v in st.values() if not v["_deleted"] and v["status"] == "paid")
print("最新:", now)
print("最新の支払済み合計:", paid(now), "円")
print("5/1 12:00時点:", noon)
print("5/1 12:00時点の支払済み合計:", paid(noon), "円")
print("2回適用しても同じ:", apply_changes(changes + changes) == now)

apply_changes(changes, until=None) は変更レコードのリストを受け取り、{注文ID: 状態の辞書} を返します。状態の辞書には業務の値のほか _deleted（墓標なら True）と _lsn（採用した変更の番号）が入ります。

操作して理解する：差分抽出が取りこぼすものを確かめる

同じ業務の動きを、(A) 更新時刻による差分抽出と (B) CDCで写し、結果を比べます。業務システムの注文表が、前回の抽出（5/1 0:00）以降に次のように変わったとします。

# 前回までに分析基盤へ写した状態
warehouse = {"o1": {"amount": 1200, "updated_at": "2026-04-30T09:00:00Z"},
             "o2": {"amount": 800,  "updated_at": "2026-04-30T10:00:00Z"},
             "o3": {"amount": 500,  "updated_at": "2026-04-30T11:00:00Z"}}
# 5/2 0:00 時点の業務システムの表（o2 は削除済み。o3 は長いトランザクションで 4/30 23:59 の更新時刻のまま 5/1 に確定）
source = {"o1": {"amount": 1000, "updated_at": "2026-05-01T10:30:00Z"},
          "o3": {"amount": 450,  "updated_at": "2026-04-30T23:59:00Z"},
          "o4": {"amount": 300,  "updated_at": "2026-05-01T12:00:00Z"}}
last_run = "2026-05-01T00:00:00Z"
inc = dict(warehouse)
for k, v in source.items():
    if v["updated_at"] > last_run:
        inc[k] = v
print("(A) 差分抽出:", {k: v["amount"] for k, v in inc.items()}, "合計", sum(v["amount"] for v in inc.values()))
print("    業務システム:", {k: v["amount"] for k, v in source.items()}, "合計", sum(v["amount"] for v in source.values()))
missing = sorted(set(inc) ^ set(source)) + [k for k in source if k in inc and inc[k]["amount"] != source[k]["amount"]]
print("    ずれている注文:", missing)

出力では、差分抽出の合計が2,600円、業務システムが1,750円で、o2（削除）と o3（古い更新時刻で後から確定）がずれています。どちらもエラーにはなりません。

考えてみましょう。

1. last_run を1時間前（2026-04-30T23:00:00Z）にずらして重ねて取ると o3 は拾えますが、o2 の削除は拾えません。重ねて取る方法は遅れて確定する行への対策にはなっても、削除への対策にはなりません。
2. CDCなら、o2 は op="d" のレコードとして、o3 は確定順の LSN つきで届きます。その代わり、ログを読む権限の設定、ログの保持期間（読み終わる前に消されると取りこぼす）、元DBへの負荷を運用で管理する必要があります。

発展：トランザクションの境界、スキーマの変化、初回の全件取得

トランザクションの境界：注文と明細を同じトランザクションで書いた場合、CDCのレコードは複数届きます。分析基盤で注文だけ先に適用されると、一時的に「明細のない注文」が見えます。対策は、(1) 変更レコードに含まれるトランザクションIDで、全部そろってから適用する、(2) 集計はある確定時刻までの変更がそろった時点（L03の watermark）でだけ行う、のどちらかです。

初回の全件取得（スナップショット）とログの接続：CDCを始めるときは、まず全件をコピーし、そのコピーを取った時点の LSN からログを読み始めます。この接続点がずれると、二重適用か取りこぼしが起きます。上の適用規則（LSN が大きいものだけ採用）は二重適用に強いので、「少し前の LSN から読み始める」側に寄せるのが安全です。

スキーマの変化：業務システムで列が追加されると、after に新しい項目が現れます。L01の取込契約の「後方互換な変更」として受け入れるか、隔離するかを事前に決めておきます。

キーの変化：主キーの値そのものが更新されると、多くのCDCでは「古いキーの削除＋新しいキーの作成」として表現されます。適用規則はキー単位なので、この2件を別のキーとして正しく処理できます。

限界：LSN はその1つのデータベースの中での順序です。複数のデータベース（注文DBと在庫DBなど）の変更の前後関係は、LSN では比べられません。時刻（commit_ts）で比べるしかなく、サーバーの時計のずれの分だけ不確かさが残ります。CDCを提供する具体的な製品・サービスの仕様は、執筆時点（2026年）の一般的な説明です。最新は各製品の公式情報で確認してください。

よくある失敗

失敗1：変更を到着順に上書きし、返金が支払済みに戻る

changes = [
    {"lsn": 107, "order_id": "o1", "status": "refunded"},
    {"lsn": 105, "order_id": "o1", "status": "paid"},   # 遅れて届いた古い変更
]
wrong = {}
for ch in changes:
    wrong[ch["order_id"]] = ch["status"]
right = {}
for ch in changes:
    if ch["order_id"] not in right or ch["lsn"] > right[ch["order_id"]][0]:
        right[ch["order_id"]] = (ch["lsn"], ch["status"])
print("到着順:", wrong, " LSN順:", {k: v[1] for k, v in right.items()})

到着順では o1 が paid、LSN で判定すると refunded です。ネットワークや再試行で、変更の到着順は簡単に入れ替わります。直し方：状態ごとに採用した LSN を保存し、それ以下の変更は無視する。

失敗2：削除を本当に消してしまい、古い更新で行が復活する

state = {"o2": {"lsn": 103, "amount": 800}}
# lsn 109 の削除が届いた → 行を消す（誤り）
del state["o2"]
# その後、遅れていた lsn 104 の更新が届く
late = {"lsn": 104, "order_id": "o2", "amount": 750}
if late["order_id"] not in state or late["lsn"] > state[late["order_id"]]["lsn"]:
    state[late["order_id"]] = {"lsn": late["lsn"], "amount": late["amount"]}
print("墓標なし:", state)
# 正しい：墓標を残すと、lsn 104 < 109 なので無視できる
state2 = {"o2": {"lsn": 109, "deleted": True}}
if late["lsn"] > state2["o2"]["lsn"]:
    state2["o2"] = {"lsn": late["lsn"], "amount": late["amount"]}
print("墓標あり:", state2)

墓標がないと、削除した注文が750円で復活します。直し方：削除は _deleted=True と LSN を持つ墓標として残す。墓標は一定期間（遅れて届く変更の上限より長く）保持してから掃除する。

失敗3：取込時刻で日次売上を区切り、日付がずれる

5/1 23:58（UTC）の注文が、通信の遅れで5/2 0:03 に分析基盤に届きました。取込時刻で日付を決めると、この注文は5/2の売上になります。翌日に5/1の売上を見直すと、業務システムの5/1の売上と一致しません。

直し方：「何日の売上か」は注文のイベント時刻（order_ts など行の中の値）で決める。取込時刻は「どれだけ遅れて届いたか」の監視に使う。さらに、日本時間の日付で集計するなら、UTCから時差（＋9時間）を足してから日付を取る、と提供仕様に書く。

失敗4：スナップショットの差を「その日の全変更」と考える

5/1 10:00 に作成され、5/1 15:00 に削除された注文は、5/1 0:00 と 5/2 0:00 のどちらのスナップショットにも存在しません。差を取っても何も起きなかったことになります。誤注文の件数やキャンセル率を測りたい場合、スナップショットの差では過小になります。

直し方：途中の変化が業務上の意味を持つ（取消・返金・状態の遷移を数える）なら、CDCかイベントで取り込む。スナップショットの差は「日末の状態の変化」を表すと、提供仕様に明記する。

まとめ

• 全件スナップショットは途中の変化を、更新時刻による差分抽出は削除を取りこぼす。ログベースCDCは変更を確定順に取り出せる。
• CDCの適用規則は「キーごとに採用済みLSN以下の変更は無視」。重複と順序の入れ替わりの両方に強く、冪等になる。
• 削除は墓標として残す。墓標がないと遅れた古い更新で行が復活する。
• イベント時刻は集計の区切り、確定時刻は時点の再現、取込時刻は遅延の監視に使う。
• ある時点の状態は、確定時刻がその時点以前の変更だけを適用して作る。

用語

CDC（変更データの捕捉）（Change Data Capture）

データベースの行の追加・更新・削除を、確定した順に変更レコードとして取り出す仕組み。

LSN（ログ通し番号）（Log Sequence Number）

データベースのトランザクションログ上の位置を表す、増え続ける番号。同じDB内の変更の確定順を表す。

墓標（tombstone）

削除された行の代わりに残す「削除済み」の印。採用したLSNを持ち、古い更新による復活を防ぐ。

確定時刻（commit timestamp）

変更が元のデータベースでコミット（確定）された時刻。時点の状態を再現するときの基準。

よくある質問

毎晩の全件コピーで困っていないなら、CDCは必要ないのでは？

日末の状態だけが必要で、削除も差で検出でき、表が小さいなら全件コピーで十分です。返金・取消など途中の変化を数えたい、表が大きく全件コピーが間に合わない、数分以内の鮮度が要る、のどれかに当てはまるとCDCを検討します。

イベント（L01の購入イベント）とCDCの変更レコードは何が違うのですか？

イベントは提供者が「業務で起きたこと」を意図して送るもので、意味（購入・返金）が明確です。CDCは表の行の変化を機械的に写したもので、「status が refunded に変わった」ことはわかっても、その業務上の理由は行の値から推測するしかありません。

LSN の代わりに commit_ts（確定時刻）で順序を決めてはいけませんか？

同じ時刻に確定した変更が複数あると順序を決められず、サーバーの時計が戻ることもあります。同じデータベースの中ではLSNのほうが確実です。複数のデータベースをまたぐ順序は、時刻で比べるしかありません。

墓標はいつまで残しておけばよいですか？

遅れて届く変更の最大の遅れより長く残します。契約の遅延上限が72時間なら、余裕を見て7日などと決め、それより古い墓標を定期的に掃除します。早く消しすぎると、遅れた更新で行が復活します。

CDCを始めると業務システムが遅くなりませんか？

ログを読む方式は、全件をSELECTする方式より業務システムへの負荷が小さいのが一般的です。ただし、ログの保持設定を長くする必要があり、読み手が止まるとログが消せずに容量を圧迫することがあります。業務システムの管理者と事前に合意します（「部門間の合意」（有料版））。

機械学習の学習データでは、どの時刻を使えばよいですか？

予測をする時点で「知り得た」値だけを使います。過去の各予測時点について、確定時刻がその時点以前の変更だけで特徴量を作ります。最新状態から作ると、未来に起きた返金などが特徴量に漏れます（特徴量の漏洩。機械学習の問題設定と評価分割・特徴量・前処理・表データ（有料版） の話題につながります）。

確認問題

次に学ぶこと

次の バッチとストリーム では、届いたイベントをいつ集計を締めるかを、バッチとストリームで比べます。遅れて届くイベントを待つ目安である watermark を扱います。

戻って確認するなら、イベントから状態を作る考え方は 「イベントデータ」（有料版）、有効期間つきの履歴表は 「履歴と変更」（有料版） です。