ksqlDB 入門: SQL で書くストリーミング処理

ksqlDB は Apache Kafka の上で動くストリーム処理エンジンで、Kafka Streams を土台にしつつ SQL ライクに継続クエリを書けます。プロデューサ/コンシューマのコードを書かずに、フィルタ、集計、ジョイン、派生トピックの作成を行えます。

本稿では、CCDAK（Confluent Certified Developer for Apache Kafka）の出題範囲に寄せながら、実務で迷いがちなキー設計、タイムセマンティクス、内部トピック、スキーマ管理の要点を具体的に解説します。

ksqlDB の全体像と位置づけ

ksqlDB は Kafka のトピックを入力として継続クエリを実行し、結果を別トピックに書き出す、あるいは内部ストアにマテリアライズして問い合わせ可能にします。永続クエリ（CSAS/CTAS）はサーバ側で常時実行され、障害時は内部トピックと状態ストアから再構築されます。

開発者は SQL で論理を宣言し、キーやパーティション、シリアライゼーション形式、タイムスタンプなどの Kafka 固有事項を明示的に設定します。これらは CCDAK でも頻出で、特にジョイン/集計前の再パーティション（PARTITION BY）を忘れないことが肝要です。

• クエリ種別: 一時（SELECT ... EMIT CHANGES）と永続（CREATE STREAM/TABLE AS SELECT）
• 問い合わせ種別: Push（変更を逐次送出）と Pull（マテリアイズ済みテーブルからキー検索）
• タイムセマンティクス: 既定はイベント時刻。WITH (TIMESTAMP=...) で明示可能
• 内部生成: 再パーティションや集計で内部トピックと状態ストアが自動生成
• 障害復旧: 内部トピックと changelog から状態ストアを再構築

イベントの流れとクエリ実行のイメージ

最小構成のストリーム定義と派生トピック作成

CREATE STREAM clicks (
  user_id STRING KEY,
  page STRING,
  ts TIMESTAMP
) WITH (
  kafka_topic='raw_clicks',
  value_format='JSON',
  timestamp='ts'
);

CREATE STREAM clicks_en AS
  SELECT user_id, page, ts
  FROM clicks
  WHERE page LIKE 'en/%'
  EMIT CHANGES;

STREAM と TABLE、Push/Pull クエリの使い分け

STREAM は不変のイベントの連続（append-only）として解釈され、フィルタやマッピング、ウィンドウ集計の前段に向きます。TABLE は最新値の集合（変更ログの畳み込み結果）を表し、キーごとの最新状態を保持します。

Push クエリは変化を逐次取得する購読で、ダッシュボード更新などに向きます。Pull クエリはマテリアイズ済み TABLE に対するキー検索で、リクエスト/レスポンス型に適します。Pull は主キー制約に基づく等価比較が基本です。

• STREAM→TABLE へは GROUP BY などの集計で遷移
• TABLE 同士や STREAM-TABLE ジョインはキー整合（同一パーティション）必須
• Pull クエリは TABLE のマテリアライズが前提（CTAS または集計）
• Push は SELECT ... EMIT CHANGES、Pull は SELECT ... WHERE key=... の形が基本

STREAM と TABLE、Push/Pull の例

-- 集計で TABLE を作成（CTAS）
CREATE TABLE pv_per_user AS
  SELECT user_id, COUNT(*) AS pv
  FROM clicks
  WINDOW TUMBLING (SIZE 5 MINUTES)
  GROUP BY user_id
  EMIT CHANGES;

-- Push: 変化を逐次取得
SELECT user_id, pv FROM pv_per_user EMIT CHANGES;

-- Pull: 特定ユーザーの現在値を 1 回取得（TABLE が必要）
-- SELECT pv FROM pv_per_user WHERE user_id='u-123';

ステートフル処理: ウィンドウ集計とジョインの基礎

ウィンドウ集計はイベント時刻で時間区切りを設け、TUMBLING（固定幅・非重複）、HOPPING（固定幅・重複）、SESSION（アクティビティの塊）の 3 形態が代表的です。遅延到着イベントに対する許容期間（グレース）や出力タイミングは設計上の重要点です。

ジョインはキーとパーティション整合が前提です。STREAM-STREAM は時間的な WITHIN ウィンドウ指定が必要、STREAM-TABLE は参照系でウィンドウ不要、TABLE-TABLE は最新値同士の結合です。いずれもキーが合わなければ PARTITION BY で再パーティションします。

• TUMBLING: 固定幅、重なりなし
• HOPPING: 固定幅、指定間隔ごとにスライドして重複あり
• SESSION: アイドル時間で区切る
• STREAM-STREAM ジョインは WITHIN が必要、他は不要
• キー不一致時は PARTITION BY で内部トピックを作り直す

ウィンドウ集計とジョインの例

-- 5 分の TUMBLING 集計
CREATE TABLE page_views_5m AS
  SELECT user_id, COUNT(*) AS cnt
  FROM clicks
  WINDOW TUMBLING (SIZE 5 MINUTES)
  GROUP BY user_id
  EMIT CHANGES;

-- STREAM-STREAM ジョイン（到着の近いイベント同士を結合）
CREATE STREAM enriched AS
  SELECT c.user_id, c.page, p.product_id
  FROM clicks c
  JOIN purchases p
    WITHIN 10 MINUTES
    ON c.user_id = p.user_id
  EMIT CHANGES;

スキーマ・シリアライゼーションと Schema Registry

ksqlDB は VALUE_FORMAT と KEY_FORMAT を指定して Avro/JSON/Protobuf/KAFKA（バイト列）などを扱います。Schema Registry を併用するとスキーマの互換性チェックと進化（後方互換など）が運用しやすくなります。既定のサブジェクト名は <topic>-value / <topic>-key です。

集計やジョインではキーが正しく定義されていることが重要です。キーが不適切だとパーティションがバラけ、集計結果が期待通りになりません。PARTITION BY でジョイン・集計前に正しいキーへ並び替えるのが定石です。

• Schema Registry 併用でスキーマ進化と互換性を自動チェック
• キーと値は別フォーマット可（例: KEY=KAFKA, VALUE=AVRO）
• WITH (TIMESTAMP='列名') でイベント時刻列を明示
• PARTITION BY で再パーティションし内部トピックが作成される

スキーマとフォーマット、再パーティションの例

-- Avro で値を管理し、イベント時刻を明示
CREATE STREAM orders (
  order_id STRING KEY,
  user_id STRING,
  amount DECIMAL(10,2),
  event_time TIMESTAMP
) WITH (
  kafka_topic='orders_raw',
  key_format='KAFKA',
  value_format='AVRO',
  timestamp='event_time'
);

-- 集計前にユーザー単位へ再パーティション
CREATE STREAM orders_by_user AS
  SELECT * FROM orders PARTITION BY user_id EMIT CHANGES;

運用の勘所: トピック設計、内部トピック、可用性

永続クエリ（CSAS/CTAS）は内部トピックと状態ストアを自動的に持ちます。スケールさせるには入力トピックのパーティション数とキー分布が均等であることが前提です。内部トピックのレプリケーションやクリーンアップポリシーも運用ポリシーに合わせて見直します。

障害時は内部の changelog から状態を再構築します。保留中のコミットや再実行に備え、少なくとも 3 ブローカー、十分なレプリケーション係数、堅牢なストレージを推奨します。Exactly-once の処理保証は Kafka/ksqlDB 双方で適切に設定したときに有効になります。

• 入力トピックのキー設計と均等な分布が最重要
• 内部トピックのレプリケーション係数とストレージ容量を監視
• CSAS/CTAS はクエリ ID 単位で管理、停止/削除の運用手順を準備
• 処理保証（at-least-once/ exactly-once）は要件とコストで選択

運用時に使う代表コマンド

-- クエリ一覧と説明
SHOW QUERIES;
EXPLAIN <QUERY_ID>;

-- 一時的な動作確認（早期テストに有効）
SET 'auto.offset.reset'='earliest';
PRINT 'raw_clicks' FROM BEGINNING LIMIT 5;

CCDAK 対策の要点と落とし穴

試験では、キーとパーティション整合、ジョイン種別の前提、STREAM と TABLE の違い、Push/Pull の使い分け、スキーマ互換性と進化、内部トピックの役割、処理保証の違いが狙われます。ksqlDB 固有の文法暗記よりも、Kafka の設計原則に沿って理由を説明できることが重要です。

特に、集計・ジョイン前の PARTITION BY、イベント時刻の取り扱い（timestamp 指定）、Pull クエリには TABLE が必要、といった定石は落としがちなので確実に押さえましょう。

• STREAM-STREAM は WITHIN 必須、STREAM-TABLE は参照系でウィンドウ不要
• Pull クエリは主キーでの等価条件が基本
• 再パーティションが必要なときは内部トピックが作られる
• スキーマ進化は互換性ルール（後方/前方/完全）に注意

出題で迷いがちな書き方の最小例

-- 典型的な CSAS と再パーティション
CREATE STREAM s AS SELECT * FROM src EMIT CHANGES;
CREATE STREAM s_by_key AS SELECT * FROM s PARTITION BY some_key EMIT CHANGES;

-- Pull クエリは TABLE が対象
-- SELECT v FROM some_table WHERE id='k';

問題で確認

よくある質問