実践Kafka Streams 〜イベント駆動型アーキテクチャを添えて〜

JJUG CCC SPRING 2025 登壇資料

Transcript

1. 実践Kafka Streams 〜イベント駆動型アーキテクチャを添えて〜 joker1007 (Repro株式会社) JJUG CCC SPRING 2025 1
2. 自己紹介 橋立友宏 (@joker1007) Repro株式会社 チーフアーキテ クト 元々はRailsエンジニアだった が、最近はJavaばかり書いてい る 日本酒とクラフトビールが好き Oxygen Not Included を再開し て生活が終わりつつある 2
3. Asakusaから来ました 活動中のRubyコミュニティで日本最古 (自分は本当に浅草に住んでます) 3
4. 最近の仕事 チーフアーキテクトとして、サービス全体の中長期的な技術選定、設計アドバイス、全体アー キテクチャのデザインなどをしている。 今回話す内容は事業内容が密接に関わっている。 4
5. Reproの事業 マーケティングオートメーションサービスを提供している。 マーケティングオートメーションとは、 ユーザーの行動を分析・分類し 複数のチャネルで適切なキャンペーンを配信することで 顧客とエンドユーザーのコミュニケーションを支援する SaaS事業だけでなく、サービスグロースの総合的な支援も提供しています。 5
6. ReproとKafkaの親和性 Reproのサービスには以下の様な特徴がある。 エンドユーザーの数が膨大なので、大量のトラフィックが届く トラフィックの多い箇所だと、秒間6万件ぐらいは捌く必要がある 直接的なビジネストランザクションではないので、個別のレイテンシはシビアで はない 一方で定期的なバッチ処理ではリードタイムが長過ぎる バックグラウンド通信が大半を占めるので、パイプラインの非同期処理化が容易 エンドユーザーの一つの行動が複数の処理と変化を引き起こす 例えばどういうものがあるか…… 6
7. ユーザーセグメンテーションの更新 ある行動をしたユーザーが、どういったユーザーの分類に含まれるのか。 久しぶりにスマホアプリを起動したユーザー 1週間以上アプリを起動していないユーザー 直近で商品を購入したユーザー ユーザーの行動ログの発生によって、これらの分類に含まれる様に変化したり、逆に条件から 外れる様に変化したりする。 7
8. トリガー配信 ある行動をしたユーザーに、プッシュ型の配信を送る。 キャンペーン応募条件を満たす行動をしたユーザーに素早く案内通知を送る カートに入れたまま購入に至っていないユーザーにリマインド通知を送る これらの配信は、複数のチャネルで行われる可能性がある。プッシュ通知だったりLINEのメ ッセージだったりメールだったりする。 8
9. ユーザーの行動というイベントを起点に考える エンドユーザーの行動をイベントして受け取り、イベントが複数のマイクロサービスを起動す る。 処理結果をスケーラブルなデータストアに永続化し、必要な時に参照を行う。 読み込みと書き込みの責務が明確なら、データの整合性はコントロールできる。 9
10. イベント駆動型アーキテクチャへ イベント駆動型アーキテクチャは以下の様な特徴を持つ。 イベントの発火点は他のサービスのことを知らなくて良い イベントバスに情報が集約され、各サービスはそこと自分の管轄範囲の情報だけ を持つ イベントに関連するサービスを自由に追加可能で、イベントバスへの負荷を除い て他サービスに影響を与えない イベントバスに依存するが、スケーラビリティは比較的確保しやすい バッチ処理よりも遥かに短いリードタイム これらの特性がReproのサービスと非常に相性が良い。 10
11. イベントバスとしてのKafkaの採用 イベント駆動型アーキテクチャにおいてはイベントバスの選択が非常に重要。 Reproでは長く開発が継続しているOSSであり、各種ミドルウェアの接続サポートの多い Kafkaを採用した。 11
12. ストリームアプリケーションの構築 Kafkaに接続するスケーラブルなアプリケーションを構築するにはストリームアプリケーショ ンの構築ノウハウを学ぶ必要がある。 一般的なWebアプリケーションやRPCを基本としたマイクロサービスとは異なった概念を理解 しなればならない。 理屈とフレームワークの両面から理解を深めていく。 12
13. Kafka Streamsフレームワーク Apache Kafkaプロジェクト公式でメンテナンスされているストリームアプリケーションフレ ームワーク。 Javaで普通にCLIアプリケーションを書く様にストリームアプリケーションが書ける。 13
14. Kafka Streamsの特徴 Kafkaクライアント自体のクラスタ管理機能を利用するので、Kafka Broker以外の ミドルウェアやリソースマネージャーに依存しない 典型的な処理を記述できるDSLとその基盤となる自由度の高いProcessor APIを持 つ 集計処理などStatefulな処理に対応するローカルなデータストアと障害復旧機能を 持つ 弊社では、こういった特徴の中でも特に基盤のシンプルさや学習コストの低さを重視して Kafka Streamsを採用した。 14
15. Kafka Streams概要 Kafkaの1つ以上のトピックからデータをconsume し、必要な加工や集計を行い、また別のトピック に書き出す、という処理を基本形とする。 consume、データ処理、produce、の各処理をノ ードと呼び、そのノードをグラフ状に繋げたトポ ロジーというモデルを構成することでアプリケー ションを記述する。 15
16. Kafka Streams DSLの基本概念 KStream トピックにレコードが到達し処理をしたら、それだけで完結し状態を持たないもの。 KTable トピックに到達したレコードはKStreamと同様にレコード単位で処理されるが、後から参照 できる様にローカルのデータストアに蓄積していて、キーバリューストアのテーブルの様に扱 えるもの。同一のキーを持つものは最新のものだけ参照できる。 GlobalKTable Tableと同様の動きをするが、全てのConsumerが同じ内容を読み込む。トピックパーティシ ョンが複数ある場合は全てのパーティションをそれぞれのConsumerが読み込むことになる 16
17. DSLのサンプルを軽く紹介 17
18. KStream, KTableの作成 StreamsBuilder builder = new StreamsBuilder(); KStream<String, Long> wordCounts = builder.stream( "word-counts-input-topic", /* input topic */ Consumed.with( Serdes.String(), /* key serde */ Serdes.Long() /* value serde */ ); KTable<String, Long> wordCounts = builder.table( "word-counts-input-topic", /* input topic */ Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as( "word-counts-store" /* table/store name */) .withKeySerde(Serdes.String()) /* key serde */ .withValueSerde(Serdes.Long()) /* value serde */ ); 18
19. 値のフィルタ・加工 KStream<String, Long> onlyPositives = stream.filter((key, value) -> value > 0); KStream<String, Integer> transformed = stream.map( (key, value) -> KeyValue.pair(value.toLowerCase(), value.length())); // キーを変更するとrepartition flagが立つ KStream<byte[], String> uppercased = stream.mapValues(value -> value.toUpperCase()); 19
20. 処理の分岐と結合 Map<String, KStream<String, Long>> branches = stream.split(Named.as("Branch-")) .branch((key, value) -> key.startsWith("A"), /* first predicate */ Branched.as("A")) .branch((key, value) -> key.startsWith("B"), /* second predicate */ Branched.as("B")) .defaultBranch(Branched.as("C")) /* default branch */ ); KStream<byte[], String> merged = stream1.merge(stream2); 20
21. ワードカウントを題材とした集約の例 KStream<String, Long> wordCounts = textLines .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .groupBy((key, word) -> word) // Count the occurrences of each word (record key). // // This will change the stream type from `KGroupedStream<String, String>` to // `KTable<String, Long>` (word -> count). .count() // Convert the `KTable<String, Long>` into a `KStream<String, Long>`. .toStream(); 21
22. 別トピックへの書き出し stream.to("my-stream-output-topic", Produced.with(Serdes.String(), Serdes.Long()); 22
23. DSLの裏側 単純に値を変換するmapValuesの実装 class KStreamMapValues<KIn, VIn, VOut> implements FixedKeyProcessorSupplier<KIn, VIn, VOut> { private final ValueMapperWithKey<KIn, VIn, VOut> mapper; public KStreamMapValues(final ValueMapperWithKey<KIn, VIn, VOut> mapper) { this.mapper = mapper; } @Override public FixedKeyProcessor<KIn, VIn, VOut> get() { return new KStreamMapProcessor(); } private class KStreamMapProcessor extends ContextualFixedKeyProcessor<KIn, VIn, VOut> { @Override public void process(final FixedKeyRecord<KIn, VIn> record) { final VOut newValue = mapper.apply(record.key(), record.value()); context().forward(record.withValue(newValue)); } } 23
24. 低レベルAPIとしてのProcessor API Kafka Streamsはプログラマーが細かな制御を行うためのAPIとしてProcessor APIというも のを提供している。 see. https://kafka.apache.org/20/documentation/streams/developer- guide/processor-api フレームワークが提供しているDSL自体も、Processor APIによって実装されており実装はそ れ程複雑ではない。 自分でProcessor APIを利用する時の参考にもなるので、ソースコードは読んでおくと良い。 24
25. Processor APIのInterface public interface Processor<KIn, VIn, KOut, VOut> { default void init(final ProcessorContext<KOut, VOut> context) {} void process(Record<KIn, VIn> record); default void close() {} } この様に非常にシンプルであり、基本的にやりたい処理をprocessメソッドに実装するだけ。 後続の処理に流したい場合は、 ProcessorContext#forward を呼ぶ。 25
26. Statefulな処理の実装方法 以下は、汎用的なAggregateを行うDSLの実装を抜粋したもの。 final ValueAndTimestamp<VAgg> oldAggAndTimestamp = store.get(record.key()); VAgg oldAgg = getValueOrNull(oldAggAndTimestamp); final VAgg newAgg; final long newTimestamp; if (oldAgg == null) { oldAgg = initializer.apply(); newTimestamp = record.timestamp(); } else { oldAgg = oldAggAndTimestamp.value(); newTimestamp = Math.max(record.timestamp(), oldAggAndTimestamp.timestamp()); } newAgg = aggregator.apply(record.key(), record.value(), oldAgg); final long putReturnCode = store.put(record.key(), newAgg, newTimestamp); 26
27. Processor APIとRocksDB Kafka StreamsはStatefulな処理を実装するためのバックエンドとしてRocksDBをフレーム ワーク内に組み込んでいる。 RocksDBはLSM Treeをベースとした高速なKVS。 処理レコードのキーをRocksDBのキーとしてシリアライズした値を格納しておき、後続に同一 キーのレコードが到達したらRocksDBから値をfetch、値を加工してから再度RocksDBにput する。 Statefulな処理は基本的にこの様に実装される。 複雑な処理を実装する場合は、このRocksDBの使い方と性質の理解が重要である。 27
28. Kafka Streams 応用編 28
29. ストリームアプリケーションのパフォーマンス Kafkaを利用した分散ストリームアプリケーションを実装する際に、特に重要なことは以下の 3つ。 1. レイテンシ 2. パーティションキーの選択 3. 処理データ量の見積りとパーティションサイズの決定 この中でも特にレイテンシに関してはシビアな感覚を持った方がいいと感じている。 29
30. ストリームアプリケーションにおけるレイテンシの重 要性 そもそもこういったアプリケーションを実装したい理由は、短いリードタイムで大量のデータ を処理するスケーラブルかつハイパフォーマンスなパイプラインを構築したいから。 そして、KafkaとKafka Streamsはその特性上、スケーラビリティはトピックのパーティショ ン数によって担保され、基本的に個別のパーティションのレコードはシーケンシャルに処理さ れる。 特に処理順番が重要なシステムではそうしなければならない。 そのため、1レコードを捌くのにかかる時間とパーティション数で一定時間辺りのスループッ トの上限が決まってしまう。 30
31. ストリームアプリケーションのレイテンシ基準 現状のReproのレイテンシ基準は概ね以下の通り。 シンプルな処理なら0.1msec以下 複雑な処理でも1〜2msec以下 1msecはかなり遅い部類で、これだと1パーティションのレコードは秒間1000件しか処理でき ない。 100パーティションあっても秒間10万件程度で頭打ちになる。 現状のReproの基準だと余裕があるとは言えない。 31
32. レイテンシを短縮するには とにかくできるだけI/Oをしない、ネットワーク通信を減らすことを意識する。 具体的には…… 外部のDBと通信しない、RDBだけでなくRedisなども re-partitionの回避 offset commitタイミングのコントロール RocksDBのチューニング 32
33. 外部DBに依存しないペイロード設計 クラウド上で構成したアプリケーションでは、Redisなどと通信するだけで1msecぐらいのレ イテンシがある。(遅い) ペイロードサイズとのバランスは考える必要があるが、レコードは圧縮も可能なので基本的に は処理に必要な情報は1レコードに全てまとめた方が処理効率は良い。 どうしてもRDBにおけるJOINの様な動作が必要な場合、同一のパーティションキーとパーテ ィション数にコントロールしたトピック同士なら、安全に一つのノードに集約できるので低レ イテンシで処理できる。 33
34. パーティションキーの設計 JOINやグルーピングしての集約において、パーティションキーが異なっていると中間トピッ クを噛まして再分散させる必要が出てきてパフォーマンスに悪影響を及ぼす。 こういった再分散(re-paritioning)を避けるために、一緒に扱いたい情報同士のパーティショ ンキーは事前に揃えておく必要がある。 集約処理を実装した後にKafkaトピックのパーティションを変更するのは非常に難しい。 Reproの場合は大抵はエンドユーザーごとに一意になるユニークIDを選択している。 ちなみに、Kafkaにおいてレコードの一意性を表すレコードキーとパーティションキーは独立 していて、それぞれ個別に設定できる。 34
35. offset commitのコントロール Kafka Streamsではconsumerのoffset commitは一定間隔で同期リクエストとして実行され る。 大抵はat least onceで動作させることになるので、短いスパンでcommitした方がリトライ時 の重複は少なくなる一方で、同期リクエストのオーバーヘッドが発生する。 処理自体が羃等であれば、デフォルトの設定(30秒)より多少長くしても問題無い。 35
36. RocksDBのパフォーマンス RocksDBはオンメモリのmemtableとbuffer cacheを活用し、高速な書き込みとルックアッ プを実現している 1msec以下のレイテンシを基本にしつつ集約処理を行うには、こういった組込み型の高速な KVSを活用する必要があるため、フレームワークがサポートしてくれている。 Reproでの主なチューニングポイントは以下の通り。 memtableのサイズやbuffer cacheサイズ コンパクションタイミング indexとbloom filterをメモリ上にpin止め 36
37. RocksDBのアーキテクチャ概要 37
38. RocksDBのコンフィグ例 BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig(); tableConfig.setBlockCache(cache); tableConfig.setCacheIndexAndFilterBlocks(true); options.setWriteBufferManager(writeBufferManager); tableConfig.setCacheIndexAndFilterBlocksWithHighPriority(true); tableConfig.setPinTopLevelIndexAndFilter(true); tableConfig.setBlockSize(4 * 1024L); tableConfig.setFilterPolicy(filter); options.setWriteBufferSize(getMemtableSize()); options.setMaxWriteBufferNumber(4); options.setMinWriteBufferNumberToMerge(2); options.setTableFormatConfig(tableConfig); options.setTargetFileSizeBase(256L * 1024 * 1024); options.setLevel0FileNumCompactionTrigger(10); 38
39. Processor APIによるRocksDBの高度な活用法 DSLのバックエンドとしてRocksDBが活用されているが、開発者自身でProcessor APIを利用 すればより細かな制御が可能になる。 39
40. レコードキーと異なるキーの利用 DSLバックエンドとしてのRocksDBは基本的にはレコードのキーをそのままキーとして利用 する。 一方でProcessor APIでは何をキーにしてRocksDBに格納するかは完全に自由である。 また当然のことだが、シリアライズ可能なものであれば何でも格納できる。 これを利用すると、エンドユーザーIDごとに分散されたノード上で、顧客ID毎に中間集約を行 い、それを最終的な集約トピックにまとめることで、データの再配置を削減する、などの細か な制御が可能になる。 40
41. その他のRocksDBの活用例 RocksDBはキーに使用したバイト列の順番でソートされるので、prefixをコントロールして scanするとセカンダリインデックスの様なことができる。 また、flushタイミングをコントロールすれば、複雑なデータ構造をシリアライズして格納で きる。 応用すると以下の様なことが実現できる。 セカンダリキーを利用した1:NのJOIN (昔は出来なかったが、今はDSLでもサポー トした) 確率的データ構造(sketch)と組み合わせたユニークカウントの近似値計算 キャンセル可能な処理を対象にした遅延キューイング 41
42. RocksDBのデータロスト対策 RocksDBは各処理ノード上のローカルストレージに保持されているので、ノードのterminate に対応できる様にデータロスト対策が必要になる。 Kafka Streamsではchangelogトピックというフレームワーク管理のtopicがRocksDBとセッ トで生成され、RocksDBに書き込んだ内容がflushされると同時にKafkaトピックにも同じ内 容のレコードが書き込まれる様になっている。 RocksDB消失時には、フレームワークが自動的にchangelogトピックからデータを取得しな おし、そのレコードを利用してRocksDBをレストアする処理が走る。 42
43. レストアによるStop the world フレームワークによるレストア処理は非常に便利だが、一方で無視できない問題も発生する。 それは、レストア処理実行中は、そのパーティションのストリーム処理がストップしてしまう ことだ。 例えばエンドユーザー毎に集計したデータ量が膨大であれば、1パーティションのレストアに もかなりの時間を要する。その間処理が停止していると無視できない処理遅延に繋がってしま う。 43
44. ストレージティアリング Reproではこの問題に対処するためにストレージティアリングの概念を導入している。 先に説明した様に外部のデータベースを利用するのは慎重になるべきだが、リクエストが常に 発生しないのであればある程度は許容できる。方針は以下の通り。 ノードローカルのRocksDBを直近のホットデータに限定したキャッシュに近いデ ータとして扱う 古いデータは外部のDB (Cassandraなど) に保存しておいて、必要に応じて取得す る また再利用に備えて外部DBから取得した値もRocksDBに書き込んでおく 上記の処理を透過的に行うイテレーターを実装して、Processor APIで利用してい る 44
45. イベント駆動アーキテクチャのトレードオフ Kafka Streamsの特性以外にも、アーキテクチャのトレードオフとして抱え込んだ課題も存在 する。 システム全体での複雑さは根本的には解消できていないし、サービス間を跨ぐテ ストや全体像の把握はモノリシックなアプリケーションよりは確実にやり辛い。 一部には同期リクエストからそういったストリーム処理を前提に構築されたコン ポーネントを利用したいケースも残っており、逆に工夫が必要になる箇所も存在 している。 処理レイテンシにかなりフォーカスしたシステム分割構成になっているので、サ ービスごとの責任範囲が直感的ではない 等々、難しい問題は多く残っている。 45
46. 疎結合の代償 RPCやWeb APIを利用したマイクロサービスと異なり、他のサービスのことを何も知らなくて も動作するため疎結合なマイクロサービス群を構築しやすい。 その反面システム全体像の把握が難しいし、オンボーディングが大変。 また、全体としてのシステムパフォーマンスの把握が難しい。 46
47. トレーシングの重要性 最近は、OpenTelemetryがこなれてきてサービス間の繋がりをトレースしやすくなった。 システム同士の依存性を可視化できるツールやサービスも増えてきて、一気に動作イメージを 把握し易くなった。 もし、新しくこういったアーキテクチャでシステムを構成するなら、最初からトレーシングの 活用を念頭に置いておくことはとても重要。 47
48. アーキテクチャ転換の所感 こういった難しさはあったものの、Kafka Streamsとストリームアプリケーションの知見の蓄 積、イベント駆動アーキテクチャの採用によって一定の成果は出せたと感じている。 チャネルの追加や新しい機能を作る上で、高いスケーラビリティが求められる箇所を上手く隠 蔽しつつ、既存のシステムに影響を与えずに新しいコンポーネントを開発できる様になった。 多くのコンポーネントを開発してきて、Reproというサービスとの相性は十分に良い選択だっ たと評価している。 全体像の把握し辛さも、分散トレーシングによるシステム構成の可視化、各種メトリックダッ シュボードの作成により対策が進んでいる。 48
49. Reproは開発者を募集中です ハイトラフィックをスピーディに捌くストリームアプリケーションに興味がある方。 大量の活きたデータを支えるデータストアのR&Dや技術選定に関わってみたい方。 新しいアーキテクチャの評価・選定に関わってみたい方。 TypeScriptとAIをゴリゴリ活用している新規事業部などもあります。 Java関係だとQuarkusなんかも活用しています。 (Quarkus + Kafka Streamsの話もどこかでやりたい) 是非Reproに話を聞きにきてください！ ご清聴ありがとうございました。 49