** 本エントリはDistributed computing Advent Calendar 2021の12/8エントリです。

はじめに

Apache Kafkaのメッセージブローカーとは一線を画する思想やデータの扱い方については前編で触れましたが、データベースで言うところのストレージ領域の話に留まっていました。今回はクエリエンジンにあたるksqlDBについて、Kafkaとの組み合わせでどのような事が可能なのかについて記載します。

ksqlDBは「ストリーミングDB」と意図的に呼ぶこともありますが、そもそも「KSQL」と呼ばれていた技術にあえて「DB」を付け加えて再構成しています。これはこの技術の、ひいてはKafkaエコシステム全体における位置付けにも大きく影響しています。やや掴みづらい技術ですが、今回はKafka全体をデータベースとして見立てて考察したいと思います。

参考

Apache Kafka Meetup Japan #9
登壇資料：「カフカはデータベースの夢をみるか」

ksqlDBとは？

ksqlDBはKafkaを利用する事を前提としたストリーム処理エンジンです。ストリーム処理基盤という言い方を避けましたが、これはksqlDB自体がJavaのアプリケーションであり、セルフオーケストレーションする事により機能する事が理由です。またksqlDBがデータ処理を実行するのではないこと、クエリをインプットとして実処理はKafka Streamsで行うこと、そして何よりksqlDBはKafka Streamsのさらに上に形成される抽象化レイヤーであるということから、「基盤」という言い方は適切ではないと感じます。

ストリーム処理とはSourceとSink、そしてその中間を流れるStreamに対する処理全体を指します。KafkaもしくはksqlDBに限らずストリーム処理自体は一般的なものではありますが、通常のデータストア（DB等）へのアクセスをベースとした処理モデルと比べると構成がやや複雑になります。

処理構成としては大きく分割して3つのコンポーネントにより構成されます：

Souce – Sourceで発生するEventを抽出しStream化する処理

Stream Processing – Streamに対する加工処理

Sink – Streamの加工結果をSinkに投入する処理

それぞれ異なる役割であり、異なるコンポーネントを採用するケースもあります。中でもStream Processingはその処理トポロジーをDAG (Direct Acyclic Graph) で表現する事が一般的であり、処理構成は複雑になります。DAGトポロジーの定義や実行ステート管理、そして実際の処理ロジックをコントロールするフレームワークが必要となり、また処理を実行するランタイム (YarnやKubernetes) が別途必要となります。

Kafka Producer/ConsumerとKafka StramsとksqlDB

KafkaへのアクセスにおいてKafkaネイティブな方法は3つあります：

• Kafka Producer/Consumer

• Kafka Streams

ksqlDB

しかしこれらは全く異なる技術ではありません。
Kafka Producer/ConsumerはEvent処理1つ1つを、ループ処理も含めて実装する方法です。ミドルウェアへの標準的なアクセスライブラリではありますが、Kafkaの思想上、重要な処理の多くを担ったライブラリともなっています。Kafka Streamsはストリーム処理のトポロジー定義とステート管理（つまりデータストアを含む）を行う分散アプリケーションとして稼働しますが、メッセージの処理はProducer/Consumerを利用しています。そしてksqlDBはインプットとして受領したSQLから実行計画を作成し、処理はKafka Streamsのトポロジーに変換の上実行されます。

この抽象化度合いがもたらす効果ですが、ステートフルな処理をConsumer APIを利用して実装すると：

ConsumerRecords<String, String> records = consumer.poll(100);
Map<String, Integer> counts = new DefaultMap<String, Integer>();
for (ConsumerRecord<String, Integer> record : records) {
  String key = record.key();
  int c = counts.get(key)
  c += record.value()
  counts.put(key, c)
}
for (Map.Entry<String, Integer> entry : counts.entrySet()) {
  int stateCount;
  int attempts;
  while (attempts++ < MAX_RETRIES) {
   try {
     stateCount = stateStore.getValue(entry.getKey())
     stateStore.setValue(entry.getKey(), entry.getValue() + stateCount)
     break;
   } catch (StateStoreException e) {
     RetryUtils.backoff(attempts);
   }
 }
}

それをKafka Streamsで実装すると：

builder.stream("input-stream",
    Consumed.with(Serdes.String(),
    Serdes.String()))
    .groupBy((key, value) -> value)
    .count()
    .toStream()
    .to("counts",
        Produced.with(Serdes.String(),
        Serdes.Long()));

さらにksqlDBで表現すると：

SELECT x, count(*) 
  FROM stream 
  GROUP BY x 
  EMIT CHANGES;

となります。

Database inside out – クエリエンジンとしてのksqlDB

しかしながら、KSQLがksqlDBと名前を変えてまで進化したことには理由があります。
Apache Kafkaは元々Linkedin内部で開発されたものですが、Kafka以外にApache Samza も開発されています。Samzaの構想は「Database inside Out」(データベースの再構築と分散化) というコンセプトを謳っており、その中ではKafkaその他ストリーム基盤上にストリームETLのパイプラインを提供するものでした。Kafkaにとってデータベースの夢をみるにはSamzaが必要でした。一方Samzaの夢はストリーミングETLであり、データベースになる事ではありませんでした。

その思想を解釈する手段としてのSQLであり、最終的には我々が分散データに対する考え方が変わった先にある新しい形を模索していくのがKafkaとksqlDBだと考えています。この夢はNewSQLの様な分散データベースが描く形とは異なりますが、Kafkaを前提としたKafkaらしいデータベースの夢である様に思います。

おわりに

本エントリでは前編に続きKafkaと、新たにksqlDBの世界観についてご紹介しました。

Kafka StreamsとksqlDBは切っても切れない関係にあります。ksqlDBは開発者から多くの事を隠蔽し、より手軽にストリーム処理を扱えるツールです。一方SQLから変換された処理はKafka Streamsによるストリーム処理として実行される為、リレーショナルな世界のSQLとは実際の働きは大きく異なります。これは制約では無く、新しい「ストリーミングDB」という考え方であると感じて頂けると幸いです。

おまけ

「ちょっとksqlDB触ってみようかな？」とお感じになった方は是非フルマネージドのConfluent Cloudをお試しください。インタラクティブチュートリアルやハンズオンデモ等、多くのリソースも合わせてご利用いただけます。
Confluent Cloud
トライアルでは400USDのクレジットが利用できますが、101KSQLDBのプロモコードでさらに101USD追加でご利用できます。

developer.confluent.io ではApache Kafka 101を始めとして、ksqlDBやKafka Streams等Kafkaエコシステムにおける様々な技術のチュートリアルシリーズをご用意しております。