整合性と再処理｜データエンジニアリング実践

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

データエンジニアリング実践（上級・読む35分＋演習45分）

この授業のゴール

• 提供している集計表を、業務システムの確定記録と日ごとの件数・金額で照合し、一致・不一致・欠落を判定できる
• 追記のみの原本から対象日を再計算し、結果が変わったときだけ版を上げる冪等な再処理を書ける
• 版の履歴から「ある時刻に利用者が見ていた数字」を答えられる
• 件数と金額の照合では見逃す誤り（打ち消し合う誤り）を説明し、ID単位の照合を使い分けられる

仕事でどう使うか

ストリームでもバッチでも、提供した数字はいつか誰かに照合されます。経理が業務システムの確定売上と比べる、経営会議の資料と先週のダッシュボードが食い違う、監査で「この数字は何に基づくか」と聞かれる。そのときに「どの版の数字で、元のどの記録から、どのコードで作ったか」を答えられることが、継続提供できる基盤の条件です。

次の例を、この授業の中心に置きます。e1＝100円、e2＝200円、e1 の再送が届いても合計は400円ではなく300円。翌朝 e3＝50円が遅れて届いたら、350円の版2として追えるようにする。

前提

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

• 冪等と置換ロード → 「冪等と再試行」（有料版）、CDCの冪等な適用 → CDCとイベント時刻
• 遅延イベントと確定値の訂正 → バッチとストリーム、遅延到着とバックフィル → 「遅延到着とバックフィル」（有料版）、「バックフィル」（有料版）
• 件数と金額の検証 → 「件数と金額の検証」（有料版）、ID単位の照合 → 「パイプラインテスト」（有料版）
• データセットの版と来歴 → 「データセット版」（有料版）、「来歴とリネージ」（有料版）

この授業の時刻は、読みやすさのため日本時間の 'MM-DD HH:MM' 形式の文字列で書きます（同じ形式なので文字列の大小で前後を比べられます）。実際の基盤ではUTCの完全な日時で保存します。

身近な状況：「昨日見た300円」と「今日の350円」はどちらが正しい？

5月1日の売上について、3人がそれぞれ違うことを言っています。

取込の原本（届いたものを届いた順にすべて残した記録）は次のとおりです。

新人の450円は再送を二重に数えた誤りです。販促担当の300円は、5/1 23:00 の時点で届いていた分については正しい値で、経理の350円は e3 が届いた後の値です。300円と350円はどちらも正しく、違うのは時点です。問題は、ダッシュボードが300円を黙って350円に書き換えると、「昨日300円と報告した」理由を誰も説明できなくなることです。

この授業では、(1) 原本から何度でも同じ規則で再処理できるようにし、(2) 結果が変わったら版を上げて履歴を残し、(3) 業務システムと毎日照合して、ずれを自動で見つける仕組みを作ります。

直感的な説明：レシートの束と、日付入りの集計表

お店の売上管理にたとえます。

• 原本は、届いたレシートを一枚も捨てずに箱に入れたもの。同じレシートのコピーが混ざっていても捨てない（後で重複だと判断できるように）。
• 再処理は、箱のレシートを最初から数え直すこと。数え方の規則（同じレシート番号は1回だけ）が同じなら、何度数えても同じ答えになる。
• 版は、集計表の右上に「5/1 23:00 集計・第1版」と書いておくこと。数え直して数字が変わったら、古い表は残したまま「5/2 9:00 集計・第2版（理由：遅れて届いたレシート e3）」を新しく作る。
• 照合は、お店の集計表を、本部の帳簿と毎日突き合わせること。

この形にしておけば、「販促担当が見たのは第1版の300円、経理が見たのは第2版の350円」と、両方の数字を説明できます。

たとえ話の限界：レシートの箱は小さいですが、実際の原本は毎日数百万件になることがあります。全期間を毎回数え直すのは費用がかかるので、再処理は日（区画）単位で、影響のある日だけ行います（「バックフィル」（有料版） のバックフィル）。

正確な定義：原本・再処理・版・照合

原本（raw / landing）：取込の入口で受け取ったデータを、加工せず、追記のみ（append-only：上書きも削除もしない）で保存したもの。重複も契約違反の行もそのまま残します。原本があれば、規則を直したときに過去を作り直せます。

再処理（reprocessing / replay）：原本から、対象の区画（例：売上日＝5/1）の結果を、決められた規則で計算し直すこと。冪等であるためには次を満たします。

1. 入力は原本だけ（前回の結果に足し込まない）。
2. 重複除去などの規則が決まっている（event_id で1件）。
3. 書き込みは区画ごとの置換か、版の追加（既存の版を書き換えない）。
4. 「いつまでに届いた分を使ったか」（計算時点）を記録する。

版（version）：ある区画の結果を、計算するたびに番号つきで残したもの。最低限、次の列を持ちます。

結果が変わらない再処理では版を上げません。こうすると、同じ再処理を何度流しても版の履歴は変わらず（冪等）、「版が上がった＝数字が変わった」と読めます。

時点の問い合わせ：時刻 $t$ に提供されていた版は、$\text{computed\_at} \le t$ の版のうち最後のものです。

照合（reconciliation）：提供している結果を、独立した正（業務システムの確定記録など）と比べることです。区画ごとに件数 $n$ と金額 $T$ を比べ、次のように判定します。

件数と金額の両方を比べるのは、片方だけでは見逃す誤りがあるからです（失敗4）。

段階付き例題：300円の版1から、350円の版2へ

ステップ1：5/1 23:59 に日次処理。この時点までに届いたのは e1・e2・e1（再送）。event_id で重複除去すると2件、$100 + 200 = 300$ 円。版1として記録します。

ステップ2：5/2 6:00 に念のため再実行。届いた分は変わらないので結果も2件300円。版は増やしません。

ステップ3：5/2 8:30 に e3 が到着。L03 の watermark でいえば遅延イベントです。

ステップ4：5/2 9:00 に再処理。原本から5/1を数え直すと3件、$100 + 200 + 50 = 350$ 円。版1と違うので版2を追加し、理由に late:e3 と書きます。版1は300円のまま残します。

ステップ5：時点の問い合わせ。「5/2 8:00 に提供されていた5/1の売上は？」→ 版1の300円。「5/2 10:00 なら？」→ 版2の350円。

ステップ6：照合。業務システムの確定記録の5/1は3件350円。版2と一致するので ok。

# 取込の原本（追記のみ）。arrived は届いた時刻（日本時間, 'MM-DD HH:MM'）
raw = [
    {"event_id": "e1", "day": "05-01", "amount": 100, "arrived": "05-01 10:00"},
    {"event_id": "e2", "day": "05-01", "amount": 200, "arrived": "05-01 10:05"},
    {"event_id": "e1", "day": "05-01", "amount": 100, "arrived": "05-01 10:06"},   # 再送
    {"event_id": "e3", "day": "05-01", "amount": 50,  "arrived": "05-02 08:30"},   # 翌朝に遅れて到着
    {"event_id": "e4", "day": "05-02", "amount": 400, "arrived": "05-02 09:00"},
]

def compute_day(raw, day, until=None):
    """原本から、その日の売上を event_id で重複除去して計算する。until までに届いた分だけ使う"""
    uniq = {}
    for r in raw:
        if r["day"] == day and (until is None or r["arrived"] <= until):
            uniq.setdefault(r["event_id"], r["amount"])
    return {"n": len(uniq), "total": sum(uniq.values())}

def reprocess(raw, day, versions, computed_at, reason):
    """再計算し、結果が最新版と違うときだけ新しい版を追加する（冪等）"""
    res = compute_day(raw, day, until=computed_at)
    mine = [v for v in versions if v["day"] == day]
    if mine and mine[-1]["n"] == res["n"] and mine[-1]["total"] == res["total"]:
        return versions
    new = {"day": day, "version": len(mine) + 1, **res, "computed_at": computed_at, "reason": reason}
    return versions + [new]

def as_of(versions, day, t):
    """時刻 t の時点で提供されていた版を返す（なければ None）"""
    cands = [v for v in versions if v["day"] == day and v["computed_at"] <= t]
    return cands[-1] if cands else None

v = []
v = reprocess(raw, "05-01", v, "05-01 23:59", "daily")
v = reprocess(raw, "05-01", v, "05-02 06:00", "rerun")      # 変化なし → 版は増えない
v = reprocess(raw, "05-01", v, "05-02 09:00", "late:e3")
v = reprocess(raw, "05-02", v, "05-02 09:30", "daily")
for x in v:
    print(x)
print("5/2 8:00 時点の5/1:", as_of(v, "05-01", "05-02 08:00")["total"], "円")
print("5/2 10:00 時点の5/1:", as_of(v, "05-01", "05-02 10:00")["total"], "円")
print("誤り（原本を単純合計）:", sum(r["amount"] for r in raw if r["day"] == "05-01"), "円")

関数の説明：compute_day(raw, day, until) は原本から売上日 day の件数と金額を返します（until までに届いた分だけ）。reprocess(raw, day, versions, computed_at, reason) は版のリストを受け取り、必要なら1版追加した新しいリストを返します。as_of(versions, day, t) は時刻 t の時点で提供されていた版（辞書）か None を返します。

操作して理解する：SQLで日次照合を自動化する

業務システムの確定記録 ledger と、基盤が提供している日次表 daily_sales を照合します。下のSQLは、区画（日）ごとに件数と金額を比べ、判定を付けます。

-- db: none
CREATE TABLE ledger (event_id TEXT PRIMARY KEY, day TEXT, amount INTEGER);
CREATE TABLE daily_sales (day TEXT PRIMARY KEY, n_events INTEGER, total INTEGER, version INTEGER);
INSERT INTO ledger VALUES ('e1','05-01',100),('e2','05-01',200),('e3','05-01',50),('e4','05-02',400),('e5','05-03',120);
INSERT INTO daily_sales VALUES ('05-01',2,300,1),('05-02',1,400,1),('05-04',1,80,1);
WITH src AS (SELECT day, COUNT(*) AS n, SUM(amount) AS total FROM ledger GROUP BY day),
keys AS (SELECT day FROM src UNION SELECT day FROM daily_sales)
SELECT k.day, s.n AS ledger_n, s.total AS ledger_total,
       d.n_events AS served_n, d.total AS served_total, d.version,
       CASE WHEN s.day IS NULL THEN 'extra'
            WHEN d.day IS NULL THEN 'missing'
            WHEN s.n = d.n_events AND s.total = d.total THEN 'ok'
            ELSE 'mismatch' END AS status
FROM keys AS k
LEFT JOIN src AS s ON k.day = s.day
LEFT JOIN daily_sales AS d ON k.day = d.day
ORDER BY k.day;

結果は、5/1＝mismatch（版1の300円のまま。e3 の遅延を反映していない）、5/2＝ok、5/3＝missing（提供側に区画がない）、5/4＝extra（業務システムにない区画がある）です。

試してみましょう。

1. daily_sales の5/1を ('05-01',3,350,2) に変えると ok になります。これが再処理で版2を作った後の状態です。
2. keys を使わず src を左に置いた LEFT JOIN だけにすると、5/4 の extra が消えます。正の側にない区画（テストデータの混入、日付の誤り）を見つけるには、両方の区画を集めてから結合します。
3. 照合で mismatch が出たら、次はID単位で差を調べます（ledger と原本の event_id を比べる）。区画単位の照合は「どこがずれたか」、ID単位の照合は「何がずれたか」を教えてくれます。

発展：区画の一括切り替え、規則の版、どこまで遡るか

読み手に途中の状態を見せない：再処理の途中（5/1の一部だけ書き換わった状態）を利用者が見ると、存在しない数字を見ることになります。新しい版を別の場所に書き終えてから、提供する版の参照を一度に切り替える（ビューの付け替え、区画の入れ替え）方法が一般的です。この授業の「版を追加する」方式は、最新版を選ぶ問い合わせを使えば自然にこの性質を持ちます。

規則の版：重複除去の規則や税の扱いを直したときも、版を上げます。この場合は reason に rule:v2 と書き、code_version を記録します。規則の変更で過去の数字が変わることを利用者に通知する手順を、L05のSLAに含めます。

どこまで遡るか：遅延イベントは契約上の上限（例：72時間）までは普通に起こるので、自動で再処理します。上限を超えて届いたものや、締め済み（経理が月次決算を確定した）期間は、確定値を書き換えずに調整表に記録し、翌期の報告で扱うのが一般的です（「遅延到着とバックフィル」（有料版））。どちらにするかは、利用者（経理・経営）との合意事項です。

原本の保持期間と費用：原本を永久に残すと保存費用が増え、個人情報の削除要求（「保存・削除・利用目的」（有料版））とも衝突します。「再処理が必要になり得る期間＋監査で求められる期間」を保持期間とし、それを過ぎた原本は集計済みの版だけを残して削除する、といった方針を文書にします。

限界：照合の「正」とした業務システムの記録自体が誤っていることもあります。2つの独立した経路の数字が一致することは、両方が同じ誤りを持たない限り強い証拠ですが、絶対の保証ではありません。

よくある失敗

失敗1：ずれた数字を手で直し、理由が残らない

-- db: none
CREATE TABLE daily_sales (day TEXT PRIMARY KEY, total INTEGER);
INSERT INTO daily_sales VALUES ('05-01', 300);
-- 誤り：経理の数字に合わせて手で上書き
UPDATE daily_sales SET total = total + 50 WHERE day = '05-01';
SELECT * FROM daily_sales;

数字は350円になりますが、「誰が・いつ・なぜ50円足したか」が残りません。翌日の自動処理が原本から300円を再計算すると、手作業の修正は消えます（e3 が原本に届いていない場合）。直し方：数字は必ず原本からの再処理で作る。原本に足りないデータがあるなら、原本に追記して再処理し、版と理由を残す。

失敗2：再処理の結果を「追記」して二重にする

table = [{"day": "05-01", "total": 300}]
# 誤り：再処理の結果をそのまま追記
table.append({"day": "05-01", "total": 350})
print("追記した表の5/1の合計:", sum(r["total"] for r in table if r["day"] == "05-01"), "円")
# 正しい：版を付けて、最新版だけを提供する
versions = [{"day": "05-01", "version": 1, "total": 300}, {"day": "05-01", "version": 2, "total": 350}]
latest = max((v for v in versions if v["day"] == "05-01"), key=lambda v: v["version"])
print("最新版:", latest)

版の列がないまま追記すると、SUM で650円になります。直し方：区画を丸ごと置換するか、版の列を持たせて「区画ごとの最新版」だけを提供するビューを作る。

失敗3：前回の結果に差分を足し込む再処理

「e3 の50円が遅れて届いたので、前回の300円に50円を足す」という処理は、一見速くて正しそうです。しかし、この処理を2回流すと400円になり、e3 が実は e1 の再送だった場合にも50円を足してしまいます。

直し方：再処理の入力は原本だけにし、前回の結果を使わない。区画の大きさが大きくて全件の数え直しが重い場合でも、まず原本からの再計算を正とし、差分の足し込みは「原本からの再計算と一致すること」を定期的に照合したうえで使う。

失敗4：件数と金額が一致したので「正しい」と判断する

ledger = {"a": 100, "b": 200, "c": 300}
served = {"a": 100, "b": 200, "x": 300}   # c が欠落し、正にない x が混入（同じ300円）
print("件数:", len(ledger), len(served), " 金額:", sum(ledger.values()), sum(served.values()))
print("ID単位の差: 欠落", sorted(set(ledger) - set(served)), " 混入", sorted(set(served) - set(ledger)))

件数3・金額600円で一致しますが、中身は違います。欠落と混入が同じ金額で打ち消し合うことは、定額の商品や定額の会費では珍しくありません。直し方：毎日の照合は件数・金額で行い（安くて速い）、定期的に、または mismatch が出たときに、ID単位の差（欠落・混入・重複）を調べる（「パイプラインテスト」（有料版））。

まとめ

• 原本は追記のみで残す。重複も違反行も捨てずに残し、再処理の唯一の入力にする。
• 再処理は原本から区画単位で数え直す。前回の結果に足し込まない。
• 結果が変わったときだけ版を追加し、計算時点と理由を残す。古い版は書き換えない。
• 時刻 t の数字は「computed_at ≤ t の最後の版」。300円と350円はどちらも、その時点では正しい。
• 照合は区画ごとに件数と金額の両方で行い、missing・mismatch・extra を区別する。打ち消し合う誤りにはID単位の照合を使う。

用語

原本（ランディング）（raw / landing zone）

取込で受け取ったデータを加工せず、追記のみで保存した記録。再処理の入力。

再処理（reprocessing / replay）

原本から、決められた規則で対象区画の結果を計算し直すこと。冪等に設計する。

照合（reconciliation）

提供している結果を、独立した正（業務システムの確定記録など）と区画ごと・ID単位で比べ、差を検出すること。

版（バージョン）（version）

区画の結果を計算するたびに、番号・計算時点・理由とともに残したもの。結果が変わったときだけ増やす。

よくある質問

原本を残すのは保存費用の無駄ではありませんか？

原本がないと、規則の誤りを直しても過去を作り直せず、照合で不一致が出ても原因を調べられません。保存費用は圧縮・安い保存先への移動・保持期間で抑えます（「ファイル形式と圧縮」（有料版）、「転送・計算・保存費用」（有料版））。原本を失う費用（数字を説明できない、監査に答えられない）と比べて判断します。

版を残すと、利用者はどの版を見ればよいのか迷いませんか？

利用者には通常「区画ごとの最新版」だけを見せ、版の番号と計算日時、訂正の有無を小さく表示します。過去の版は、説明や監査が必要なときに参照する記録です。

置換（区画の上書き）と版の追加は、どう使い分けますか？

誰がいつどの数字を見たかを説明する必要がある（経営報告・請求・監査に使う）なら版の追加、そうでない中間の表（作業用の集計）なら置換で十分です。置換でも、処理ログに「いつ何件・いくらに変わったか」を残すと説明がしやすくなります。

照合の「正」はいつも業務システムですか？

多くの場合、売上なら会計・決済の確定記録が正です。ただし業務システムにも遅延や誤りがあります。正をどれにするか、正と食い違ったときにどちらを調べるかを、提供仕様に書いておきます（L05）。

再処理はどれくらい遡れるようにしておくべきですか？

最低でも契約の遅延上限（例：72時間）は自動で再処理し、それ以上は手動のバックフィル手順で対応する、という二段構えが一般的です。締め済みの期間は数字を変えず、調整表で扱うかどうかを利用者と合意します。

照合で不一致が出たら、提供を止めるべきですか？

差の大きさと利用者への影響で決めます。例えば「差が0.1%以内なら注記して提供、それを超えたら保留して調査」のように、事前に閾値を決めておきます。次の 品質と提供SLA で、品質の検査と提供の判断を扱います。

確認問題

次に学ぶこと

次の 品質と提供SLA では、照合や契約検査の結果をもとに「今日の数字を提供してよいか」を判定し、鮮度・完全性の目標（SLO）と提供の約束（SLA）を数値で管理します。

戻って確認するなら、冪等なロードは 「冪等と再試行」（有料版）、バックフィルの計画は 「バックフィル」（有料版）、変更時の回帰確認は 「変更時の回帰確認」（有料版） です。