Streamsを試してみます。Streamsについては、こちら。Streams参考にした日本語情報。Streamsオフィシャルサイトで、今回試した範囲でざっと気になるあたり...。Queriesとまあ、いろいろあるのですが、小難しいことは置いておいて、とりあえず動かしてみましょう。Streamsを扱うのに必要な、Maven依存関係はこちら。Countにしてみましょう。元ネタは、オフィシャルサイトのコンセプトのドキュメントにします。Luceneを使うとしましょう。</dependency>あとは、テストライブラリを。KafkaのTopicsは、以下の2つを作成。Countした結果を保存するTopicです。作成したテストコードの雛形は、こちら。ここに、テストを書く!}とりあえず、データを入れてみるまずは、データを入れておきます。Streamsは関係ないのでさらっと。データは、「test-topic」に放り込んでいます。}jsoupでデータをパース後、scriptタグを削除したりしていますが、これはHandlebarsのテンプレートになっていて、ここからHTMLを抜き出してセレクタで評価できる形にしたかったからです。Streamsを使ったコードを書いてみます。Streamsnのドキュメントサンプルをほぼそのまま真似ました。Streams書いたコードは、こちら。}tokenizeしている部分は、こういう感じになっています。LuceneのStandardAnalyzerでアナライズして単語に分解します。Kafkaへの接続情報やSerialize/Deserializeで使うデフォルトのクラスなどを設定します。"100");「wordcount-application」は今回のアプリケーション名で、コミット間隔は100msecにしています。APIのように処理を書いていきます。Countなので、最後はcountなのですが、その際にMaterializedを使用するようです。.count(Materialized.as("counts-store"));そして、このStreamの結果を別のTopicに書き出します。ここでは、最初に用意した「word-count-topic」に書き込むようにします。Produced.valueSerde(Serdes.Long()));toメソッドでTopicに書き出す際に、デフォルトのSerializer/Deserializaerを変更することができます。ドキュメントではProduced#withを使用してキーと値の両方を変更していますが、Produced#valueSerdeのように値に適用するSerializer/Deserializerのみの変更を行うなどもできるようです。「Serdes」ってなんだろうと思っていたのですが、Serialize/Deserializerのそれぞれ先頭3文字から取ったものみたいですね。ここまで定義したら、KafkaStreamsのインスタンスを作成してstart。streams.start();すぐに終わってしまうと困るので、今回はsleepさせておきます...。}本来は実行し続ける感じになるので、ドキュメントではSIGTERM送信時にRuntime#shutdownHookを使ってKafkaStreams#closeを呼び出す例が記載されています。ApplicationKafkaStreamsのインスタンスは、使い終わったらclose。streams.close();なお、このコードを実行すると、作成したTopic以外に2つのTopicが作られるようです。それぞれ、「wordcount-application-counts-store-changelog」と「wordcount-application-counts-store-repartition」という名前。アプリケーションの名前と、count時に指定したstoreの名前からきているようですね。今回は置いておきます...。Queryどうやら、Query(...というより、Iteratorですが)も使用できるとか。Queriesこちらも試してみます。}KafkaStreamsを作成して実行するところまでは先ほどとほぼ同じで、KafkaStreamsのインスタンスと途中で使った「counts-store」を使ってReadOnlyKeyValueStoreのインスタンスを取得します。keyValueStore.all();ここから、KeyValueIteratorを取得することができるので、あとはIteratorとして使えばOKです。今回はReadOnlyKeyValueStore#allを使って全レコードを取得しましたが、キーのみを指定したgetや、範囲を指定したrangeなどもあります。https://kafka.apache.org/10/javadoc/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.htmlその他、SessionStoreやWindowStoreなどもあるようなので、また気が向いたら...。全レコードをListにしてKafkaStreams#close。streams.close();上位10件くらいをちょっと強引に抜き出して、見てみます。9L));直接見てみる?KafkaStreamsを使って書き出したTopic「word-count-topic」を、Consumerを使って直接見たらどうなるか気になったので、ちょっと見てみました。}いたって普通に見れました。Countした結果が入っていますね。「wordcount-application-counts-store-changelog」についても見てみましたが、そう変わらず。}ちょっとこのあたりの話はちゃんと見ていないので、おいおい...。Streamsを試してみました。Kafkaの挙動と合わさって動作確認はちょっとてこずりました...。面白そうなので、少しずつ見ていってみましょう。読者になる検索リンクはてなブログブログをはじめる週刊はてなブログはてなブログPro月別アーカイブはてなブログをはじめよう!Kazuhiraさんは、はてなブログを使っています。ブログを報告するスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。