取込契約と粒度｜データエンジニアリング実践

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

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

この授業のゴール

• 取込契約（スキーマ・意味・粒度・キー・時刻・配送保証・変更手順）の項目を、具体的な値で書ける
• 受け取ったイベントを契約で検査し、合格・隔離・重複に分けて「入力件数＝合格＋隔離＋重複」を確かめられる
• 粒度の違う表をそのまま足して起きる二重計上を、キーの一意性検査で見つけられる
• 契約の変更が後方互換か破壊的かを判定し、必要な手順（版の更新・移行期間）を選べる

仕事でどう使うか

データ基盤の仕事の多くは、他のチームが作ったデータを受け取るところから始まります。アプリ開発チームが購入イベントの項目名を amount から price に変えた翌朝、売上ダッシュボードが0円になる。明細ごとに送っていたイベントを注文ごとに変えたら、売上が半分になる。どちらもエラーは出ません。こうした事故を防ぐ取り決めが 取込契約（データ契約、data contract）です。

この授業から6回で、遅延と重複を持つ購入イベントを取り込み、利用者へ毎日提供する小さな基盤を作ります。L01はその入口です。

前提

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

• 測定単位と定義（金額は税込か、時刻はどの時差か）→ 測定単位と定義
• 重複と識別子（何が同じ1件かを決めるキー）→ 「重複と識別子」（有料版）
• スキーマと互換（項目の追加・削除が利用者に与える影響）→ 「スキーマと互換」（有料版）、検査の書き方は 「スキーマ検証」（有料版）
• ファクト表の粒度（1行が何を表すか）→ 「スター型とファクト粒度」（有料版）
• イベントデータ（起きた事実を消さずに積む記録）→ 「イベントデータ」（有料版）

この授業の Python は、画面内でそのまま実行できます。外部ライブラリは使わず、標準ライブラリの re（正規表現：文字列の形を調べる仕組み）だけを使います。

身近な状況：誰も間違えていないのに、売上が2倍になった

架空の通販アプリ「そらマート」では、アプリ開発チームが購入イベントを送り、データ基盤チームが受け取って日次売上を集計しています。ある日から、売上が前週の約2倍になりました。

調べると、アプリ側で次の変更がありました。

2商品の注文（合計1,200円）で考えます。

基盤チームは今までどおり SUM(amount_jpy) を計算し、2,400円と報告しました。正しくは1,200円です。アプリ側は「明細ごとに送るほうが便利」と考え、基盤側は「いつもどおり足しただけ」。どちらも自分の範囲では正しいのに、1行が何を表すか（粒度）の合意が文書になっていなかったことが原因です。

この授業では、このような食い違いを受け取る前に見つけるための取込契約と、受け取った後に機械で検査する方法を学びます。

直感的な説明：宅配の「受け取り条件」を先に決める

取込契約は、宅配便の受け取り条件に似ています。

• 箱の大きさと中身の書き方（スキーマ）：項目名・型・必須かどうか
• 1箱に何を入れるか（粒度）：1箱＝1注文なのか、1商品なのか
• 伝票番号（キー）：同じ荷物が2回届いても、伝票番号で1つとわかる
• いつ発送したか（イベント時刻）といつ届いたか（取込時刻）を別々に書く
• 再配達の約束（配送保証）：「少なくとも1回は届ける。2回届くこともある」
• 条件を変えるときの連絡（変更手順）：何日前に、誰に知らせるか
• 条件に合わない荷物の扱い（違反時の処理）：捨てずに保管棚へ置き、送り主に連絡する

受け取る側は、届いた荷物を条件と照らして合格・保管棚（隔離）・二重配達（重複）に分けます。このとき、捨てた荷物があると後で数が合わなくなるので、「届いた数＝合格＋隔離＋重複」が必ず成り立つように扱います。

たとえ話の限界：宅配と違い、データは一度受け取ると下流の多くの集計に一斉に流れます。1件の違反が、ダッシュボード・機械学習の特徴量・請求の数字に同時に広がる点が、契約を厳密にする理由です。

正確な定義：取込契約の項目と、粒度・キー・配送保証

取込契約（data contract）は、データを作る側（提供者）と使う側（利用者）が合意する、データの形と意味と運用の約束です。最低限、次の項目を具体的な値で書きます。

（数値は架空の例です。実際の値は業務の要求と提供者の能力から合意して決めます。）

粒度の検査は、キーの一意性で行います。粒度が「決済1回」でキーが event_id なら、重複除去後の行数は event_id の種類数と一致するはずです。さらに「同じ order_id の行が複数ある」ことが契約上ありえないなら、order_id の重複も違反として検出できます。

配送保証の3分類を区別します。

変更の互換性：任意項目の追加は、既存の利用者が無視できるので後方互換です。項目名の変更・削除・型の変更・意味の変更（税抜→税込、粒度の変更）は破壊的変更で、版を上げて移行期間を設けます。そらマートの事故は、項目名は同じでも意味（粒度）が変わった破壊的変更でした。

段階付き例題：6件のイベントを契約で検査する

入力：アプリから届いた次の6件（到着順）。

ステップ1：1件ずつ契約と照らす。e2 は金額が文字列なので型違反、e3 は時刻の書式違反、e4 は契約にない種類 cancel なので値の違反です。

ステップ2：違反した行は隔離する。捨てずに、元の行と理由を残します。後で提供者に「どの行が、なぜ」と伝えるためです。

ステップ3：合格した行から重複を除く。4番目の e1 は1番目と同じ event_id なので重複です。e5 は金額1,200円で e1 と同じですが、別の決済なので残します（金額が同じことは重複の理由になりません）。

ステップ4：件数を照合する。入力6＝合格2（e1, e5）＋隔離3（e2, e3, e4）＋重複1。合格分の売上は 1,200＋1,200＝2,400円です。

ステップ5：意味を読む。隔離が6件中3件（50%）は、契約の「1%を超えたら提供を保留」を大きく超えます。売上2,400円を確定値として出さず、提供者へ連絡するのが正しい判断です。

import re
CONTRACT = {
    "name": "purchase_events", "version": 1,
    "grain": "1行 = 決済1回（注文の明細ではない）",
    "key": "event_id",
    "required": ["event_id", "event_type", "order_id", "amount_jpy", "event_time"],
    "event_types": {"purchase", "refund"},
}
TIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(Z|[+-]\d{2}:\d{2})$")

def check_record(rec):
    """1件のイベント(dict)を契約と照らし、違反理由のリストを返す。空なら合格"""
    reasons = []
    for f in CONTRACT["required"]:
        if rec.get(f) in (None, ""):
            reasons.append("missing:" + f)
    a = rec.get("amount_jpy")
    if a is not None:
        if not isinstance(a, int) or isinstance(a, bool):
            reasons.append("type:amount_jpy")
        elif a < 0:
            reasons.append("range:amount_jpy")
    t = rec.get("event_type")
    if t not in (None, "") and t not in CONTRACT["event_types"]:
        reasons.append("enum:event_type")
    ts = rec.get("event_time")
    if ts not in (None, "") and not TIME_RE.match(str(ts)):
        reasons.append("format:event_time")
    return reasons

records = [
    {"event_id": "e1", "event_type": "purchase", "order_id": "o1", "amount_jpy": 1200, "event_time": "2026-05-01T10:00:00+09:00"},
    {"event_id": "e2", "event_type": "purchase", "order_id": "o2", "amount_jpy": "800",  "event_time": "2026-05-01T10:05:00+09:00"},
    {"event_id": "e3", "event_type": "purchase", "order_id": "o3", "amount_jpy": 500,  "event_time": "2026/05/01 10:07"},
    {"event_id": "e1", "event_type": "purchase", "order_id": "o1", "amount_jpy": 1200, "event_time": "2026-05-01T10:00:00+09:00"},
    {"event_id": "e4", "event_type": "cancel",   "order_id": "o4", "amount_jpy": 300,  "event_time": "2026-05-01T10:09:00+09:00"},
    {"event_id": "e5", "event_type": "purchase", "order_id": "o5", "amount_jpy": 1200, "event_time": "2026-05-01T10:12:00+09:00"},
]

accepted, quarantined, dup = [], [], 0
seen = set()
for rec in records:
    reasons = check_record(rec)
    if reasons:
        quarantined.append((rec["event_id"], reasons))
    elif rec["event_id"] in seen:
        dup += 1
    else:
        seen.add(rec["event_id"]); accepted.append(rec)
print("合格:", [r["event_id"] for r in accepted])
print("隔離:", quarantined)
print("重複:", dup)
print("照合:", len(records), "=", len(accepted), "+", len(quarantined), "+", dup)
print("売上:", sum(r["amount_jpy"] for r in accepted), "円")
print("隔離率:", round(len(quarantined) / len(records) * 100, 1), "%")

check_record(rec) は辞書 rec を受け取り、違反理由の文字列のリスト（例 ["type:amount_jpy"]）を返す関数です。空のリストなら合格です。TIME_RE.match(文字列) は、時刻が「年-月-日T時:分:秒＋時差」の形のときだけ一致します。

操作して理解する：粒度が変わったことを機械で見つける

冒頭の事故（1行＝明細に変わった）を、キーの一意性で検出します。下のコードは、同じ order_id が何回出てくるかを数え、契約「1行＝決済1回（同じ注文は1行）」に反する注文を報告します。

from collections import Counter
april = [
    {"event_id": "e1", "order_id": "o1", "amount_jpy": 1200},
    {"event_id": "e2", "order_id": "o2", "amount_jpy": 800},
]
may = [
    {"event_id": "e10-1", "order_id": "o10", "line_no": 1, "amount_jpy": 1200},
    {"event_id": "e10-2", "order_id": "o10", "line_no": 2, "amount_jpy": 1200},
    {"event_id": "e11-1", "order_id": "o11", "line_no": 1, "amount_jpy": 500},
]
def grain_report(rows):
    c = Counter(r["order_id"] for r in rows)
    multi = {k: v for k, v in c.items() if v > 1}
    new_fields = sorted(set().union(*[r.keys() for r in rows]) - {"event_id", "order_id", "amount_jpy"})
    return {"rows": len(rows), "orders": len(c), "multi_row_orders": multi, "unknown_fields": new_fields}
print("4月:", grain_report(april))
print("5月:", grain_report(may))
print("5月の単純合計:", sum(r["amount_jpy"] for r in may), "円")
print("注文ごとに1回だけ数えた合計:", sum({r["order_id"]: r["amount_jpy"] for r in may}.values()), "円")

出力では、5月に multi_row_orders が {'o10': 2}、unknown_fields が ['line_no'] になります。単純合計は2,900円、注文ごとに1回だけ数えると1,700円です。

試してみましょう。

1. may の o10 の2行目を消すと、multi_row_orders は空になります。粒度の検査は「重複がない」ことしか確かめられないので、契約の文書と、知らない項目（line_no）の出現の両方を監視します。
2. 「注文ごとに1回だけ数える」直し方は、各行の金額が注文全体の金額だという前提に依存します。明細ごとの金額が入っていれば逆に過小になります。どちらが正しいかはデータからは決められず、提供者に確認するしかありません。これが、契約を文書で持つ理由です。

発展：契約をどこで、どう強制するか

契約の検査は、置く場所によって効果が変わります。

実務では3か所を組み合わせます。

スキーマレジストリ（schema registry）は、イベントの型定義を版ごとに登録し、新しい版が古い版と互換かを自動判定する仕組みです。互換の向きには「新しい読み手が古いデータを読める（後方互換）」「古い読み手が新しいデータを読める（前方互換）」があり、どちらを要求するかは誰が先に更新するかで決めます。JSON Schema、Avro、Protocol Buffers などの形式が使われます（具体的な製品の機能は執筆時点（2026年）の一般的な説明です。最新は各製品の公式情報で確認してください）。

行単位の隔離と全体の停止の使い分け：1行の書式違反は隔離で十分ですが、「必須項目が全行で欠けている」「粒度が変わった」は全体の意味が変わっているので、提供を止めるほうが安全です。L05で、隔離率の閾値による提供判断を扱います。

限界：契約は「何を約束したか」を明確にしますが、提供者が約束どおりに送っているか（例：金額が本当に税込か）は検査だけでは保証できません。定期的に元の業務システムの数字と照合する（L04）ことが必要です。

よくある失敗

失敗1：契約違反の行を黙って捨て、件数が合わなくなる

rows = [{"id": "a", "amount": 100}, {"id": "b", "amount": "x"}, {"id": "c", "amount": 300}]
# 誤り：変換できない行を try/except で無言で飛ばす
clean = []
for r in rows:
    try:
        clean.append({"id": r["id"], "amount": int(r["amount"])})
    except ValueError:
        pass
print("誤り：", len(clean), "件, 合計", sum(r["amount"] for r in clean))
# 正しい：理由とともに隔離し、件数を照合する
ok, bad = [], []
for r in rows:
    (ok if isinstance(r["amount"], int) else bad).append(r)
print("正しい：合格", len(ok), "隔離", len(bad), "入力", len(rows), "一致:", len(ok) + len(bad) == len(rows))

誤りの版は「2件、合計400円」とだけ出力し、b が消えたことは誰にもわかりません。翌月に b が1万円の注文だったと判明しても、どこで消えたか追えません。直し方：違反は理由つきで隔離表に残し、「入力＝合格＋隔離＋重複」を毎回照合する。

失敗2：重複を内容（金額や顧客）で判定し、本物の購入を消す

rows = [
    {"event_id": "e1", "customer": "c1", "amount": 1200},
    {"event_id": "e1", "customer": "c1", "amount": 1200},   # 再送
    {"event_id": "e2", "customer": "c1", "amount": 1200},   # 同じ客が同じ金額をもう一度購入
]
by_content = {(r["customer"], r["amount"]) for r in rows}
by_key = {r["event_id"]: r for r in rows}
print("内容で重複除去:", len(by_content), "件", sum(a for _, a in by_content), "円")
print("キーで重複除去:", len(by_key), "件", sum(r["amount"] for r in by_key.values()), "円")

内容で判定すると1件1,200円、キーで判定すると2件2,400円です。同じ人が同じ金額の商品を2回買うことは普通に起きます。直し方：重複の判定は契約で決めたキー（event_id）だけで行う。提供者が再送でも同じ event_id を送ることを契約に書く。

失敗3：粒度の違う表をそのまま足して二重計上する

注文の粒度の金額（注文合計）が、明細の粒度の行に繰り返し入っている表を、そのまま sum します。

lines = [  # 1行 = 明細。order_total は注文全体の金額を繰り返している
    {"order_id": "o1", "line_no": 1, "order_total": 1200},
    {"order_id": "o1", "line_no": 2, "order_total": 1200},
    {"order_id": "o2", "line_no": 1, "order_total": 500},
]
wrong = sum(r["order_total"] for r in lines)
right = sum({r["order_id"]: r["order_total"] for r in lines}.values())
print("誤り:", wrong, "円  正しい:", right, "円")

誤りは2,900円、正しくは1,700円です。直し方：足す量がどの粒度で定義されているか（注文合計は注文の粒度）を契約に書き、集計前に「キーの一意性」を検査する。一意でなければ、まず注文の粒度へまとめてから足す（「結合前の集計」（有料版））。

失敗4：時差のない時刻を受け入れ、日付の境界でずれる

"2026-05-01 00:30" のように時差がない時刻は、日本時間なのか協定世界時（UTC）なのか決められません。日本時間の5月1日0時30分は、UTCでは4月30日15時30分です。日本時間として解釈してUTCの日付で集計すれば月をまたいで4月の売上に入り、UTCとして解釈すれば5月1日の売上になります。解釈の仕方で月が変わってしまいます。

直し方：契約で「時差つきのISO 8601形式」を必須にし、時差のない時刻は format:event_time で隔離する。基盤の内部ではUTCにそろえて保存し、日次集計の「1日」は日本時間かUTCかを提供仕様に明記する。

まとめ

• 取込契約は、スキーマ・意味・粒度・キー・時刻・配送保証・遅延の上限・変更手順・違反時の処理・責任者を具体的な値で合意した文書。
• 粒度（1行の意味）は型では検出できない。キーの一意性と未知の項目の出現で機械的に監視する。
• 違反行は捨てずに理由つきで隔離し、「入力＝合格＋隔離＋重複」を毎回照合する。
• 少なくとも1回の配送では重複が届く。重複はキーで判定し、内容（金額・顧客）では判定しない。
• 名前・型が同じでも意味が変われば破壊的変更。版を上げ、通知と移行期間を設ける。

用語

取込契約（データ契約）（data contract）

データの提供者と利用者が合意する、スキーマ・意味・粒度・キー・配送保証・変更手順・違反時の処理などの約束。

配送保証（delivery guarantee）

メッセージが何回届くかの約束。高々1回・少なくとも1回・ちょうど1回の3種類で表す。

隔離（検疫）（quarantine）

契約違反の行を捨てずに別の表へ理由つきで保管し、後で確認・再処理できるようにすること。

破壊的変更（breaking change）

既存の利用者の処理や結果の意味を壊す変更。項目の改名・削除・型の変更・意味や粒度の変更など。

スキーマレジストリ（schema registry）

イベントの型定義を版ごとに登録し、新旧の版の互換性を自動で判定する仕組み。

よくある質問

取込契約は、なぜ文書にする必要があるのですか？ 型の検査だけでは足りませんか？

型の検査では「金額が税込か」「1行が決済か明細か」といった意味を確かめられません。冒頭の事故は型も項目名も同じでした。意味は文書で合意し、型や一意性は機械で検査する、という役割分担が必要です。

データ辞書やスキーマ定義とは何が違うのですか？

データ辞書（データ辞書）は項目の意味の説明、スキーマは型の定義です。取込契約はそれらに加えて、粒度・キー・配送保証・遅延の上限・変更手順・違反時の処理・責任者という運用の約束を含みます。誰が何をしたら契約違反かが決まる点が違います。

違反した行は、基盤側で直して取り込んではいけないのですか？

書式の統一（時差の変換など）を契約で「基盤側が行う」と決めているなら構いません。決めていない修正を黙って行うと、提供者は誤りに気づかず直さないままになり、修正規則が基盤側にだけ増えていきます。原則は隔離して提供者に伝え、修正したら版つきの規則として記録します。

配送保証が「ちょうど1回」の基盤なら、重複除去は不要ですか？

不要にはなりません。「ちょうど1回」はある区間（例：メッセージ基盤から書き込み先まで）の約束であることが多く、アプリが送信に失敗して再送するなど、その手前で生まれた重複は防げません。キーによる重複除去は安価な保険として残します。

隔離が多いとき、提供を止めるか、合格分だけで提供するか、どう判断しますか？

隔離率の閾値と、隔離がどこに偏っているかで判断します。全体の0.1%で特定の端末だけなら合格分で提供し注記する、5%を超えるか売上の大きい部分に偏るなら保留する、といった基準を契約に事前に書いておきます。L05で詳しく扱います。

提供者が契約を守ってくれないときはどうしますか？

違反の件数・例・影響（どの指標がいくらずれたか）を数字で示し、責任者に連絡します。契約に違反時の連絡手順と対応期限を書いておくと、個人間の交渉にならずに済みます（「部門間の合意」（有料版））。

確認問題

次に学ぶこと

次の CDCとイベント時刻 では、業務システムのデータベースの変更（追加・更新・削除）をイベントとして取り込む CDC と、イベント時刻・取込時刻の扱いを学びます。

戻って確認するなら、重複とキーは 「重複と識別子」（有料版）、スキーマの互換は 「スキーマと互換」（有料版）、受入基準の作り方は 「受入基準」（有料版） です。モジュール全体の流れは データエンジニアリング実践 を参照してください。