Kafkaのウィンドウ処理：Tumbling / Hopping / Sessionを正しく使い分ける

ウィンドウ処理は「いつのデータをどこまでまとめるか」を決める時間軸の設計。Kafkaでは基本的にイベント時刻ベースで考えます。

Tumbling（固定長・非重複）、Hopping（固定長・重複）、Session（可変長・マージ）を区別して選ぶのが第一歩。

ウィンドウ処理の基本とタイムセマンティクス

Kafkaのウィンドウ処理は、連続するストリームを時間で区切って集計・結合するための仕組みです。Kafka Streams DSLとksqlDBはいずれも原則としてイベント時刻（レコードのタイムスタンプ）を用います。ストリーム時間は、観測したイベント時刻の最大値で進みます。

遅延イベント（到着順が前後するデータ）を扱うには、ウィンドウ終端後にどの程度まで受け入れるか（Grace）を設定します。ウィンドウは「ストリーム時間が ウィンドウ終端＋Grace に達した時点」で閉じられ、それ以降に属するレコードは遅着として破棄されます。

正しく時刻を扱うには、TimestampExtractor（Kafka Streams）やTIMESTAMP列（ksqlDB）を適切に指定し、期待するイベント時刻を付与することが重要です。

• イベント時刻が基本。処理時刻ベースの近似は試験でも実務でも推奨されません。
• ストリーム時間は最大観測時刻で進む。よって高い時刻のレコードが1件でも到来すると、過去ウィンドウの閉鎖が進む可能性があります。
• 遅延受け入れはGraceで明示。Graceが短すぎると正確性を失い、長すぎるとレイテンシとメモリ消費が増えます。

Tumbling / Hopping / Session の違い（比較表）

3種のウィンドウは境界の作り方と重なり方が決定的に異なります。要件（重複カウント可否、ユーザー行動のかたまり検出、レポートの粒度）に合わせて選択します。特に試験（CCDAK）では、定義と使いどころを問う設問が繰り返し出題されます。

Sessionは非活動ギャップで区切る可変長。遅延イベントを契機に別々のセッションがマージされることがある、という点が実務・試験ともに重要です。

• Tumbling: 固定長・非重複。分単位のKPIやウィンドウごとに唯一の結果が欲しいケースに最適。
• Hopping: 固定長・重複。スライディング分析（移動平均、移動カウント）に向く。重複計上を理解して使う。
• Session: 可変長・マージあり。ユーザー行動の「まとまり」を捉える用途に強い。

遅延イベントとGrace、ストリーム時間の挙動（図解）

ウィンドウは、ストリーム時間がウィンドウ終端＋Graceに到達するまで「開いたまま」です。ストリーム時間は観測したイベント時刻の最大値で進むため、将来時刻のレコードが到着すると過去ウィンドウが一気に閉じることがあります。

以下の図は、Tumblingウィンドウ [0,5), [5,10) と、Grace=2 の例です。ts=4 の遅延イベントがストリーム時間6の時点で到着しても、5+2=7 まではウィンドウ[0,5)が開いているため取り込まれます。

• ウィンドウ閉鎖条件: stream-time >= window-end + grace
• 遅延の受け入れは「閉鎖前まで」。閉鎖後に到着したレコードは破棄される（メトリクスで観測可能）。
• 高い時刻のレコードはstream-timeを押し上げる。設計時はデータ分布を把握する。

ウィンドウ終端とGraceのイメージ（Tumbling, Grace=2）

time --->
0    2    4    6    8   10
|----|----|----|----|----|
[ W1:0-5 )                [ W2:5-10 )
|-------------------------|-------------------------

イベント到着:
 e1: ts=3, arrive@3   -> 属する: W1（オンタイム）
 e2: ts=6, arrive@6   -> 属する: W2（オンタイム）=> stream-time = 6
 e3: ts=4, arrive@6   -> 遅延だが W1 に取り込み可能（stream-time 6 < 5+Grace 7）
 e4: ts=1, arrive@9   -> W1は既に閉鎖（stream-time>=7）=> 遅着として破棄

実装パターン（Kafka Streams DSL と ksqlDB）

Kafka Streams DSLでは、TimeWindows.ofSizeAndGrace と SessionWindows.ofInactivityGapAndGrace を用いてウィンドウ長とGraceを明示します。HoppingはadvanceByでスライド幅を指定します。Sessionではマージ関数（Aggregator Merger）が必須です。

ksqlDBでは、WINDOW句でTUMBLING/HOPPING/SESSIONを指定し、SIZE/ADVANCE BY/GRACE PERIODを合わせて宣言します。最終結果のみを取りたい場合はEMIT FINALを併用します（中間更新を抑制）。実際のバージョンにより句の細かな位置やサポートは変わることがあるため、運用前に該当バージョンのドキュメントを確認してください。

• Kafka Streams: ofSizeAndGrace / ofInactivityGapAndGrace を使う。保持はウィンドウ長+Grace以上に。
• HoppingはadvanceByでスライド設定。重複集計を前提にダウンストリームで扱う。
• ksqlDB: WINDOW句とGRACE PERIOD、必要に応じてEMIT FINAL。

Kafka Streams DSL と ksqlDB のサンプル

// Kafka Streams (Java)
KStream<String, String> source = builder.stream("input");

// 1) Tumbling 1分 + Grace 5分（イベント時刻ベース）
KTable<Windowed<String>, Long> tumbling = source
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(1), Duration.ofMinutes(5)))
    .count();

// 必要ならウィンドウ閉鎖後のみ出力（中間を抑制）
KTable<Windowed<String>, Long> tumblingFinal = tumbling
    .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()));

// 2) Hopping（長さ5分・1分スライド）+ Grace 5分
KTable<Windowed<String>, Long> hopping = source
    .groupByKey()
    .windowedBy(
        TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(5))
                   .advanceBy(Duration.ofMinutes(1))
    )
    .count();

// 3) Session（非活動ギャップ5分・Grace 10分）
KTable<Windowed<String>, Long> sessions = source
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(10)))
    .aggregate(
        () -> 0L,
        (key, value, agg) -> agg + 1,
        (agg1, agg2) -> agg1 + agg2, // セッションマージ時の結合関数
        Materialized.with(Serdes.String(), Serdes.Long())
    );

// ksqlDB
-- ストリーム定義（イベント時刻としてtsを使用）
CREATE STREAM clicks (user_id VARCHAR, ts BIGINT)
  WITH (KAFKA_TOPIC='clicks', VALUE_FORMAT='JSON', TIMESTAMP='ts');

-- Tumbling 1分 + Grace 5分、最終結果のみ
CREATE TABLE clicks_tumbling AS
  SELECT user_id, COUNT(*) AS c
  FROM clicks
  WINDOW TUMBLING (SIZE 1 MINUTE, GRACE PERIOD 5 MINUTES)
  GROUP BY user_id
  EMIT FINAL;

-- Hopping（長さ5分・1分スライド）+ Grace 5分
CREATE TABLE clicks_hopping AS
  SELECT user_id, COUNT(*) AS c
  FROM clicks
  WINDOW HOPPING (SIZE 5 MINUTES, ADVANCE BY 1 MINUTE, GRACE PERIOD 5 MINUTES)
  GROUP BY user_id
  EMIT FINAL;

-- Session（ギャップ60秒・Grace 5分）
CREATE TABLE clicks_session AS
  SELECT user_id, COUNT(*) AS c
  FROM clicks
  WINDOW SESSION (60 SECONDS, GRACE PERIOD 5 MINUTES)
  GROUP BY user_id
  EMIT FINAL;

-- 注: 実運用ではバージョンによりWINDOW/GRACE句の詳細が異なる場合があります。

状態ストアの保持と抑制（Suppression）・出力戦略

ウィンドウ集計は内部に状態（ウィンドウストア）を保持します。保持期間は少なくとも「ウィンドウ長＋Grace」以上が必要です。Kafka StreamsのofSizeAndGraceやofInactivityGapAndGraceを使う場合、この前提を満たす設定を行うのが実務上の定石です。

抑制（Suppression）はウィンドウ閉鎖まで中間結果を溜め、閉鎖時に最終結果のみを下流へ流します。untilWindowClosesは分かりやすい一方、バッファが増えメモリプレッシャーになるため、負荷・レイテンシ要件と合わせて評価します。ksqlDBではEMIT FINALが近い役割を担います。

• 保持は「長さ＋Grace」が原則。足りないと遅延イベントを受け入れられない。
• Suppression/EMIT FINALはレイテンシとメモリのトレードオフ。KPI最終値のみが必要な場合に有効。
• 高カーディナリティのキーではストアサイズが急増するため、キー設計と圧縮トピック運用を併せて検討する。

CCDAK試験対策の要点と落とし穴

出題の定番は、ウィンドウ種別の定義、イベント時刻と処理時刻の違い、Graceによる遅延許容、セッションマージの挙動です。特にSessionのマージ条件（非活動ギャップをまたぐ遅延により別セッションが結合する）と、ウィンドウの閉鎖条件（stream-time >= window-end + grace）は暗記レベルで押さえましょう。

Hoppingは重複カウントが前提である点、Suppression/EMIT FINALの使いどころ、保持期間が足りないと遅延を取り込めない点は実務でも誤解が多く、試験でもひっかけに使われます。

• TimestampExtractor/TIMESTAMP指定を忘れない。処理時刻で近似しない。
• Sessionはマージ時に再集計される。マージ関数が必要。
• ウィンドウ閉鎖はstream-time基準。レコード到着時刻ではない。

問題で確認

よくある質問