Apache Kafka - カフカはデータベースの夢をみるか (後編)

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

はじめに

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

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

この技術によってリレーショナルデータベースを置き換えるとか、機能差異がどこにどれくらいあるのかといった点については言及しません。あくまでKafkaがデータベースになるという「夢」を見たとして、ではそれはどんな夢なのか、その夢の中でksqlDBがどのような役割を果たしているのかを説明します。（リレーショナルDBとかNewSQLに戦争を仕掛ける類のエントリではありません）

本エントリは先日 (2021/09/24) 実施されたApache Kafka Meetup Japan #9での登壇内容をなぞらえたものとなっています。

参考

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) が別途必要となります。

ksqlDBはこのend-to-endの処理をクエリで構成する事を目的としています。Source、Sinkとの接続、Stream Processing、そしてMaterialized Viewを一つのクエリ言語で表現し構成することが出来ます。

もう一つ特徴的なのはSinkの処理です。一般的には処理の最後はSinkへのデータ投入です。ksqlDBでもゴールをSinkとする事ができますが、合わせてその時点での状態 (Materialized View) をクエリで取得する事が出来ます。ksqlDBではこの2つの異なるクエリの形態を__Push Query__と__Pull Query__と呼び、ストリーム処理結果の消費パターンの違いも吸収してよりシンプルに活用出来る仕組みとなっています。

接続 (Connector) に関してはその定義の性質上、同じクエリ言語で表現はしますが記載方法は一般的なSQLとだいぶ異なります。（Kafka ConnectのConnector定義に近い）

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

前述した様に、ksqlDBはSQL構文で記載したストリーム処理をKafka Streamsアプリとして実行しています。一から設計したミドルウェアと異なり、Kafkaデータにアクセスする為の抽象化レイヤーです。ランタイムも内部にKafka Streamsアプリケーションを含んだ1つのアプリケーションとして構成されます。ksqlDBはミドルウェアとしてでは無く「アプリケーション」として捉える方が妥当な側面もあります。

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

その後、Kafkaの見る夢にはksqlDBが登場する様になります。

Kafkaの最終ゴールあくまで「企業全体にデータを行き届ける中枢神経」となる事であり、データベースのストレージになる事では無いのではと思います。しかしながら、KafkaとSamzaが描いたDatabase inside outの世界観を拡張する可能性がksqlDBにはあります。この夢は、Kafka上でリレーショナルデータベースを模倣する事では無く、Kafkaらしいストリーミングデータベースとしての夢です。

その思想を解釈する手段としての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エコシステムにおける様々な技術のチュートリアルシリーズをご用意しております。