問題タブ [apache-kafka]

私はStormで作業し、ローカル テキスト ファイルを入力ソースとして使用する基本的なプログラムを開発しました。しかし今は、外部システムから継続的に送られてくるデータのストリーミングに取り組まなければなりません。この目的には、Kafka が最適です。

問題は、Spout に Kafka からストリーミング データを取得させる方法です。または、Storm を Kafka と統合する方法。Kafkaからのデータを処理できるようにするにはどうすればよいですか?

kafkaを使用する場合、kafka プロデューサーの kafka.compression.codec プロパティを設定することでコーデックを設定できます。

プロデューサーでスナッピー圧縮を使用するとします。カフカコンシューマーを使用してカフカからメッセージを消費する場合、スナッピーからデータをデコードするために何かをする必要がありますか、それともカフカコンシューマーの組み込み機能ですか?

関連ドキュメントでは、kafka コンシューマーのエンコーディングに関連するプロパティを見つけることができませんでした (プロデューサーのみに関連しています)。

誰かこれクリアできる？

kafka 7.2を使用して、プロデューサーを使用してメッセージを送信すると、メッセージを消費すると、メッセージの先頭に追加のセクションが追加されて到着することがわかります。

たとえば、単純な文字列「King Daniel」を kafka に送信する場合、バイト配列では次のようになります。

しかし、何らかの理由でそれを消費すると、次のようになります。

「………………|King Daniel」という文字列はどれ？

したがって、メッセージの先頭に 12 文字が追加されます。これは何かのヘッダーですか？元のメッセージを取得するにはどうすればよいですか?

これが私の消費者コードです：

をファイルに書き込んでいmsg.message().payload().array()て、このファイルを開くと、最初に 12 文字が追加された元のコンテンツが表示されます。

元のメッセージを正確に取得するにはどうすればよいですか?

Zookeeper から最後のオフセット時間を取得するには? ストーム スパウトを使用してカフカからメッセージを読み取る場合。コンテキスト: Kafka はメッセージを継続的に取得し、コンシューマーはしばらくの間読み取り、その後何らかの理由でシャットダウンし、コンシューマーは最新のメッセージのみを読み取りますが、最後に読み取られたオフセットからは読み取りません。

見つける方法はありますか

• メッセージがまだ消費されていない場所からのオフセットまたはタイムスタンプ

• 消費された最後のタイムスタンプ

または、ストームを使用してカフカで既読メッセージと未読メッセージを区別する

メッセージを消費するためにkafkaスパウトを使用しています。しかし、トポロジを変更してアップロードする必要がある場合、古いメッセージから再開するか、新しいメッセージから開始しますか? Kafka スパウトは、消費する場所からのタイムスタンプを指定しますが、タイムスタンプを知るにはどうすればよいでしょうか?

私はkafka stromの統合に取り組んでいます。エラーで立ち往生しています。を使用して実行しようとすると、ビルドが失敗します
mvn -e -f m2-pom.xml compile exec:java -Dexec.classpathScope=compile -Dexec.mainClass=storm.starter.MainTopology

これは、pom.xml ファイルのスニペットです。

私は試した

私は storm-0.9.0-rc3 と kafka-0.7.2 を使用しています