バッチとストリーム｜データエンジニアリング実践

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

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

この授業のゴール

• watermark（ここまでのイベント時刻はそろったとみなす目安）を計算し、枠を確定するタイミングと遅延イベントを判定できる
• 待ち時間（許容遅延）を変えたときの、遅延イベント数と結果の遅れのトレードオフを数値で説明できる
• 配送保証（高々1回・少なくとも1回・ちょうど1回）を、読み取り位置の記録と書き込み先の冪等性の組み合わせで説明できる
• 鮮度の要求・遅延の分布・運用費（架空単価）から、バッチ・マイクロバッチ・ストリームを選べる

仕事でどう使うか

「キャンペーン中の売上を15分ごとに見たい」「月次の確定売上は翌月3日までに」。利用者の要求する鮮度（どれだけ新しい数字が要るか）によって、処理方式と費用は大きく変わります。ストリーム処理は数秒〜数分で結果を出せますが、常時動かす費用と、遅れて届くイベントをどう扱うかという難しさを抱えます。

この授業では、L01の契約（少なくとも1回の配送、遅延の上限）を前提に、watermark を使って「いつ枠を締めるか」を決める小さなストリーム処理を作ります。

前提

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

• バッチ・ストリーム・マイクロバッチの違い、イベント時刻で枠を決めること → 「バッチとストリーム」（有料版）
• 遅延到着と再計算の窓、締め → 「遅延到着とバックフィル」（有料版）
• 冪等と再試行 → 「冪等と再試行」（有料版）、「再試行と冪等」（有料版）
• 取込契約の配送保証・遅延の上限 → 取込契約と粒度、イベント時刻と確定時刻 → CDCとイベント時刻

この授業では、時刻を「10:00からの経過分」という整数 event_min で表し、計算を見やすくします。枠は半開区間 $[\text{開始}, \text{開始}+\text{幅})$ で、10分枠なら 0〜9分台が枠0、10〜19分台が枠10です。

身近な状況：「10分ごとの売上」が後から増える

そらマートのタイムセールで、販促担当者は10分ごとの売上をダッシュボードで見ています。ところが、10:00〜10:10 の売上が、10:12 に見ると300円、10:20 に見ると350円、翌朝には430円になっていました。

原因はスマートフォンの通信です。地下鉄で購入した人の購入イベントは、電波が戻るまで端末にたまり、数分〜数時間遅れて届きます。さらに通信が不安定だと、アプリは同じイベントを再送します（L01の「少なくとも1回」の配送）。

担当者が知りたいのは「10:00〜10:10の売上はいくらで、その数字はもう変わらないのか」です。ストリーム処理では、すべてのイベントが届くのを永遠に待つことはできません。どこかで「この枠はもう締める」と決める必要があります。その目安が watermark です。

直感的な説明：集合写真の「もう来ないだろう」

集合写真を撮るとき、集合時刻を過ぎても少しは遅刻者を待ちます。何分待つかが 許容遅延（待ち時間）です。

• 待ち時間を長くすると、遅刻者も写れる（取りこぼしが減る）。ただし、写真ができるのが遅くなる。
• 待ち時間を短くすると、写真はすぐできる。ただし、後から来た人は写れない（別撮りが必要）。

ストリーム処理では、時計の代わりに届いたイベントの時刻を使って「今どこまで来たか」を推定します。「これまでに見た一番新しいイベントが10:18なら、待ち時間5分を引いた10:13より前のイベントは、もうほぼ届いたはずだ」と考えます。この10:13が watermark です。watermark が枠の終わり（10:10）を越えたら、その枠を確定して結果を出します。

確定後に届いたイベント（遅刻者）は、捨てずに遅延イベントとして別に記録し、確定値の訂正（L04で扱う版の更新）や調整表に回します。

たとえ話の限界：集合写真の遅刻者は数えられますが、ストリームでは「まだ届いていないイベントが何件あるか」は誰にもわかりません。watermark は推定であり、保証ではありません。待ち時間は、過去の遅延の分布から「99%が間に合う」ように決めるのが普通です。

正確な定義：枠・watermark・確定・遅延、そして配送保証

記号：イベント $i$ のイベント時刻を $t_i$（分）、枠の幅を $w$（分）、許容遅延を $d$（分）とします。

枠の開始：$s_i = \lfloor t_i / w \rfloor \times w$。$\lfloor\ \rfloor$ は小数点以下の切り捨てです。例：$t=18, w=10$ なら $\lfloor 1.8 \rfloor \times 10 = 10$ で、枠10に入ります。

watermark：これまでに処理したイベントの時刻の最大値から $d$ を引いた値です。

$$W = \max_{j \le \text{今}} t_j - d$$

例：e5（18分）まで処理し $d=5$ なら、$W = 18 - 5 = 13$ です。

確定：枠 $[s, s+w)$ は、$W \ge s + w$ になった時点で確定し、結果を出力します。

遅延イベント：届いた時点で自分の枠がすでに確定している（$s_i + w \le W$）イベントです。確定値には足さず、別に記録します。

結果の遅れ：枠の終わりから確定までには、少なくとも $d$ 分かかります。$d$ を大きくすると遅延イベントは減りますが、結果が遅れます。

日次バッチは「翌日2時まで待つ」という長い待ち時間のストリームだと考えることもできます。

配送保証と書き込み：ストリーム処理は、入力のどこまで読んだか（読み取り位置、オフセット）を記録しながら進みます。障害で再起動すると、最後に記録した位置から読み直します。

「ちょうど1回」は、多くの場合この3行目の組み合わせで実現されます。冪等な書き込みとは、たとえば「枠の開始時刻をキーに、合計を上書きする」ことです。「合計に足す」書き込みは冪等ではありません。

段階付き例題：8件を、待ち時間5分で処理する

冒頭の8件を、枠の幅 $w=10$、待ち時間 $d=5$ で処理します。

出力：確定値は枠0＝350円、枠10＝450円。枠20（120円）は未確定。遅延イベントは e6（80円）、重複は1件。

照合：重複を除いた入力は 100＋200＋150＋50＋300＋80＋120＝1,000円。確定800円＋未確定120円＋遅延80円＝1,000円で一致します。

意味：販促担当者には「10:00〜10:10は350円で確定。ただし確定後に80円の遅延があり、訂正版は430円」と伝えます。待ち時間を10分にすれば e6 は間に合いますが、枠0の確定は「20分以降のイベント（ここでは26分の e7）が届くまで」遅れます。

def run_stream(events, window=10, delay=5):
    """到着順のイベントを処理し、イベント時刻の枠ごとの合計を watermark で確定させる"""
    final, open_, late, dup, seen, W = {}, {}, [], 0, set(), None
    for e in events:
        if e["event_id"] in seen:            # 少なくとも1回の配送 → 重複を除く
            dup += 1
            continue
        seen.add(e["event_id"])
        start = e["event_min"] // window * window
        if W is not None and start + window <= W:
            late.append(e["event_id"])       # 確定済みの枠に届いた → 遅延として別扱い
        else:
            open_[start] = open_.get(start, 0) + e["amount"]
        cand = e["event_min"] - delay
        W = cand if W is None else max(W, cand)
        for s in sorted(open_):
            if s + window <= W:              # 枠の終わりを watermark が越えたら確定
                final[s] = open_.pop(s)
    return {"final": final, "open": open_, "late": late, "duplicates": dup, "watermark": W}

events = [  # 到着順。event_min は 10:00 からの経過分（イベント時刻）
    {"event_id": "e1", "event_min": 2,  "amount": 100},
    {"event_id": "e2", "event_min": 7,  "amount": 200},
    {"event_id": "e3", "event_min": 12, "amount": 150},
    {"event_id": "e1", "event_min": 2,  "amount": 100},   # 再送
    {"event_id": "e4", "event_min": 4,  "amount": 50},    # 少し遅れて到着
    {"event_id": "e5", "event_min": 18, "amount": 300},
    {"event_id": "e6", "event_min": 9,  "amount": 80},    # 大きく遅れて到着
    {"event_id": "e7", "event_min": 26, "amount": 120},
]

for d in (0, 5, 10):
    r = run_stream(events, window=10, delay=d)
    print(f"待ち時間{d:>2}分: 確定={r['final']} 未確定={r['open']} 遅延={r['late']} 重複={r['duplicates']} W={r['watermark']}")

run_stream(events, window, delay) は、到着順のイベントのリストを受け取り、確定・未確定の枠の合計（{枠の開始: 金額}）、遅延イベントのID、重複件数、最後の watermark を辞書で返します。

操作して理解する：待ち時間・遅延の分布・費用を比べる

遅延が実際の分布に従うとき、待ち時間 $d$ を変えると何が起きるかを確かめます。下のコードは、架空の1,000件のイベントについて、到着の遅れを「多くは1分以内、たまに長い」分布で作り、$d$ ごとの遅延率と、結果の遅れを比べます。最後に、架空の単価で月の運用費を比べます。

import numpy as np
rng = np.random.default_rng(0)
n = 1000
t_event = np.sort(rng.uniform(0, 600, n))               # 10時間に1000件
lag = rng.exponential(1.0, n)                            # 遅れ：平均1分
heavy = rng.random(n) < 0.03                             # 3%は10〜120分遅れる
lag[heavy] += rng.uniform(10, 120, heavy.sum())
t_arrive = t_event + lag
order = np.argsort(t_arrive)
ev = [{"event_id": i, "event_min": float(t_event[i]), "amount": 1} for i in order]
def run_stream(events, window=10, delay=5):
    """到着順のイベントを処理し、イベント時刻の枠ごとの合計を watermark で確定させる"""
    final, open_, late, dup, seen, W = {}, {}, [], 0, set(), None
    for e in events:
        if e["event_id"] in seen:            # 少なくとも1回の配送 → 重複を除く
            dup += 1
            continue
        seen.add(e["event_id"])
        start = e["event_min"] // window * window
        if W is not None and start + window <= W:
            late.append(e["event_id"])       # 確定済みの枠に届いた → 遅延として別扱い
        else:
            open_[start] = open_.get(start, 0) + e["amount"]
        cand = e["event_min"] - delay
        W = cand if W is None else max(W, cand)
        for s in sorted(open_):
            if s + window <= W:              # 枠の終わりを watermark が越えたら確定
                final[s] = open_.pop(s)
    return {"final": final, "open": open_, "late": late, "duplicates": dup, "watermark": W}

for d in (1, 5, 15, 60):
    r = run_stream(ev, window=10, delay=d)
    print(f"待ち時間{d:>2}分: 遅延率 {len(r['late'])/n*100:5.1f}%  枠の終わりから確定まで 約{d}分以上")
unit = 40   # 架空単価：計算機1台1時間あたり40円
print("ストリーム（常時1台）:", 1 * 24 * 30 * unit, "円/月")
print("日次バッチ（1日20分）:", round(20 / 60 * 30 * unit), "円/月")

出力の読み方：

1. 待ち時間を1分→60分に延ばすと遅延率は下がりますが、3%の「長く遅れるイベント」の一部は60分でも間に合いません。どの待ち時間でも遅延はゼロにならないので、遅延イベントの扱い（L04の版の更新）が必ず必要です。
2. 運用費は、架空単価でストリーム28,800円/月、日次バッチ400円/月で、約70倍です。「15分ごとの数字」が意思決定（在庫の追加発注など）を本当に変えるかを確かめてから選びます。
3. rng = np.random.default_rng(0) の 0 を変えると、乱数が変わります。遅延率が大きく変わらないことを確かめましょう。

発展：watermark の決め方と、ストリームで難しいこと

watermark の決め方：この授業の「最大時刻−固定の待ち時間」は最も単純な方式です。実際の基盤では、入力の区画（パーティション）ごとに watermark を計算し、その最小値を全体の watermark にします。1つの区画だけ止まると、全体の watermark が進まず結果が出ない問題（アイドルな区画）があり、一定時間イベントが来ない区画を除外する設定が用意されていることがあります。

トリガー：確定を待たずに途中経過を出す（例：1分ごとに暫定値を出し、確定時に最終値を出す）方式もあります。この場合、利用者に「暫定」「確定」「訂正」の区別を必ず見せます。

状態の大きさ：重複除去のための seen（見た event_id の集合）は、放っておくと無限に大きくなります。実務では「watermark より十分古い event_id は忘れる」ように保持期間を決めます。保持期間より遅れて届いた再送は重複として検出できないので、L04の再処理（日次の全件照合）で補います。

結合：2つのストリーム（注文と決済など）を結合するには、両方の到着を待つ時間の上限が必要です。上限を過ぎた片方だけのイベントの扱いを決めておかないと、状態が増え続けます。

ブラウザでの模擬の範囲：この授業のシミュレーターは1台・1区画・メモリ上の小さな処理で、watermark・遅延・重複の考え方を確かめるものです。分散したストリーム処理基盤の性能・障害時の挙動・区画ごとの watermark は、この模擬では確かめていません。実際の製品（Apache Flink、Spark Structured Streaming、各クラウドのストリーム処理サービスなど）の設定名や既定値は、執筆時点（2026年）の一般的な説明です。最新は各製品の公式情報で確認してください。

よくある失敗

失敗1：処理した時刻（到着時刻）で枠を決める

events = [{"id": "e6", "event_min": 9, "arrive_min": 14, "amount": 80},
          {"id": "e5", "event_min": 18, "arrive_min": 18, "amount": 300}]
by_arrive, by_event = {}, {}
for e in events:
    a = e["arrive_min"] // 10 * 10; by_arrive[a] = by_arrive.get(a, 0) + e["amount"]
    s = e["event_min"] // 10 * 10;  by_event[s] = by_event.get(s, 0) + e["amount"]
print("到着時刻で集計:", by_arrive, " イベント時刻で集計:", by_event)

到着時刻で集計すると、10:09の購入80円が10:10台の売上になります。通信状況によって売上の時間帯が変わり、再処理するたびに値が変わります。直し方：枠はイベント時刻で決め、到着時刻との差は遅延の監視指標にする。

失敗2：確定後に届いたイベントを黙って捨てる（または黙って足す）

遅延イベントを捨てると、ストリームの合計は業務システムより常に少なくなり、差の理由を誰も説明できません。逆に、確定済みの値に黙って足すと、10:12 に「350円で確定」と見た人と、翌朝「430円」と見た人の数字が食い違い、どちらが正しいかわかりません。

confirmed = {0: 350}
late = [{"id": "e6", "window": 0, "amount": 80}]
# 正しい：確定値は残し、遅延を版つきで記録して訂正版を作る
adjust = [{"window": l["window"], "delta": l["amount"], "reason": "late:" + l["id"]} for l in late]
v2 = {w: v + sum(a["delta"] for a in adjust if a["window"] == w) for w, v in confirmed.items()}
print("版1（確定）:", confirmed, " 調整:", adjust, " 版2（訂正）:", v2)

直し方：遅延イベントは理由つきで記録し、確定値は版を上げて訂正する（L04）。利用者には版と訂正の有無を見せる。

失敗3：少なくとも1回の配送なのに、合計に「足す」書き込みをする

sink_add, sink_put = {}, {}
batch = [("w0", 350)]
def write_add(sink, rows):
    for k, v in rows: sink[k] = sink.get(k, 0) + v
def write_put(sink, rows):
    for k, v in rows: sink[k] = v
for attempt in range(2):          # 書き込み後、位置を記録する前に落ちて、再起動でもう一度処理
    write_add(sink_add, batch); write_put(sink_put, batch)
print("足す書き込み:", sink_add, " 上書き（冪等）:", sink_put)

足す書き込みでは700円、上書きでは350円です。直し方：書き込み先をキー（枠の開始・版）による上書きか、一意制約つきの upsert にする。読み取り位置は書き込みが成功した後に記録する。

失敗4：待ち時間を勘で決める

「念のため待ち時間0分で速く出す」と決めると、例題では e4 と e6 の2件が遅延になり、確定値の訂正が毎回発生します。逆に「念のため24時間待つ」と、ストリームにした意味がなくなります。

直し方：過去のイベント時刻と取込時刻の差（遅れ）の分布を測り、「95%や99%が間に合う」値を候補にする。その待ち時間で確定値が出る時刻が、利用者の鮮度の要求（例：枠の終わりから15分以内）を満たすかを確かめる。満たせないなら、暫定値＋訂正の運用を利用者と合意する。

まとめ

• 枠はイベント時刻で決め、watermark＝最大のイベント時刻−待ち時間 が枠の終わりを越えたら確定する。
• 確定後に届いたイベントは遅延として理由つきで記録し、確定値は版を上げて訂正する。捨てても黙って足してもいけない。
• 待ち時間を延ばすと遅延は減るが結果は遅れる。遅れの分布から決め、遅延はゼロにならない前提で設計する。
• 実質的なちょうど1回は「書き込み後に読み取り位置を記録」＋「冪等な書き込み」で作る。
• 方式は鮮度の要求・遅延の分布・運用費で選ぶ。シミュレーターは論理の確認であり、本番の性能の保証ではない。

用語

watermark（ウォーターマーク）（watermark）

「この時刻より前のイベントはほぼ届いた」とみなす推定の時刻。最大のイベント時刻−許容遅延などで計算し、枠を確定するきっかけにする。

許容遅延（待ち時間）（allowed lateness）

枠の終わりから確定までに、遅れて届くイベントを待つ時間。

遅延イベント（late event）

自分の属する枠が確定した後に届いたイベント。確定値とは別に記録して訂正に使う。

読み取り位置（オフセット）（offset / checkpoint）

入力のどこまで処理したかの記録。再起動時はここから読み直す。記録する時点で配送保証が決まる。

実質的なちょうど1回（effectively-once）

重複して処理されても、冪等な書き込みによって結果が1回処理したときと同じになる性質。

よくある質問

watermark は、なぜ時計の時刻ではなくイベント時刻で計算するのですか？

時計で計算すると、入力が止まったり、過去のデータを再処理したりしたときに、イベントの実際の時刻と関係なく枠が締まってしまいます。イベント時刻で計算すれば、再処理しても同じ結果になります（再現性）。ただし入力が完全に止まると watermark が進まないので、その対策が別に必要です。

DA04-L05 の「再計算の窓」と watermark は何が違うのですか？

どちらも「遅れて届くデータをどこまで待つか」の考え方です。再計算の窓はバッチで「直近k日分を毎回作り直す」方法、watermark はストリームで「この時刻より前はそろったとみなして確定する」方法です。バッチの窓は後から何度でも作り直せる前提、watermark は一度出した確定値を版で訂正する前提、という違いがあります。

ちょうど1回（exactly-once）をうたう製品を使えば、重複除去は要りませんか？

製品の保証は、多くの場合「その製品の入力から出力まで」の範囲です。アプリが送信に失敗して再送した重複など、範囲の外で生まれた重複は残ります。event_id による重複除去と、冪等な書き込みは残しておくのが安全です。

ストリームとバッチを両方持つのは無駄ではありませんか？

速報はストリーム、確定はバッチで全件を作り直す、という二本立ては実務でよく使われます。費用と二重の実装の管理が負担になるので、同じ処理のコードでバッチとストリームの両方を動かせる設計や、ストリームの遅延イベントを版で訂正する設計（L04）で一本化することもあります。

待ち時間はどのくらいにすればよいですか？

取込時刻−イベント時刻（遅れ）の分布を過去のデータで測り、例えば99%が間に合う値を候補にします。その値で、利用者が求める鮮度（枠の終わりから何分以内に確定値が欲しいか）を満たせるかを確かめます。満たせなければ、暫定値と確定値を分けて提供する合意をします。

この授業のシミュレーターの結果を、本番の基盤の性能見積もりに使えますか？

使えません。このシミュレーターは1台のメモリ上の処理で、watermark・遅延・重複・照合の論理を確かめるためのものです。本番の処理量・区画ごとの watermark・障害時の再起動時間は、実際の基盤で負荷試験をして確かめます。

確認問題

次に学ぶこと

次の 整合性と再処理 では、遅延イベントや再送があっても、確定値と業務システムの数字が一致していることを照合し、元のログから再処理して版つきで訂正する方法を学びます。

戻って確認するなら、バッチとストリームの基本は 「バッチとストリーム」（有料版）、遅延到着とバックフィルは 「遅延到着とバックフィル」（有料版）、費用の考え方は 「性能・遅延・費用」（有料版） です。