問題タブ [trident]

このプロジェクトでは、Storm と Trident を試しており、そのために Clojure と Marceline を使用しています。Marceline のページにある wordcount の例を拡張して、文のスパウトがローカルのスパウトではなく DRPC 呼び出しから来るようにしようとしています。DRPC ストリームがクライアントに返すには結果が必要であるという事実に起因すると思われる問題がありますが、DRPC 呼び出しが効果的に null を返し、永続化されたデータを更新するだけにしたいと考えています。

コードには 2 つの代替方法があります。固定バッチ スパウトを使用する方法は問題なくロードされますが、代わりに DRPC ストリームを使用してコードをロードしようとすると、次のエラーが発生します。

このエラーは、クライアントに何かを返すために DRPC ストリームが出力をサブスクライブしようとしている必要があるという事実から来ていると思いますが、サブスクライブするpersistent-aggregateそのような出力は提供しません。

では、DRPC ストリームによって永続化されたデータが更新されるようにトポロジを設定するにはどうすればよいでしょうか?

マイナーアップデート:これは不可能なようです:( https://issues.apache.org/jira/browse/STORM-38

私はストーム トライデント トポロジを構築しており、ある時点で partitionPersist ステージを使用してデータを solr に保存しています。State クラスを実装し、StateUpdater を使用して solr を更新するロジックをいくつか書きました。重複をsolrに保存しないようにするために、どの段階でもトランザクションID（txid）を追跡する必要がありますか、それともトライデントがこれを処理してくれますか？

partitionPersist ステージと状態クエリを使用して、ストーム トライデント トポロジにウィンドウ結合を実装しました。クエリを実行しているときに、メモリ内の状態からアイテムを削除したいと思います。これは可能ですか？

シンプルなハッシュマップ状態と状態クエリを持つストーム トライデント トポロジがあります。状態の更新中に状態クエリが呼び出されることはありますか?

私は Java と Trident を初めて使用します。ツイートを取得するためにプロジェクトをインポートしましたが、何かを取得したいtuple.getValue(0); 最初のツイートのみを意味するコードから取得したときに、このコードが複数のツイートを取得する方法 ?!

ハッシュセットまたはハッシュマップ内のすべてのツイートを取得して、各ツイートの特徴的な単語の総数を取得するという問題

このメソッドは、ツイートで方程式を実行するために使用されます

このコードは最初のツイートを取得してから本文を取得し、それを文字列の配列に変換します。解決する必要があることはわかっていますが、うまく記述できませんでした

私の考え：forループのようにする

ストリーム処理用のストームを設定しています。トライデント スパウトは kafka トピックからデータをフェッチしていますが、kafka からメッセージをフェッチしている間、連続して BufferUnderFlowException を取得しています。

java.lang.RuntimeException: java.lang.RuntimeException: java.nio.BufferUnderflowException at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:128) ~[storm-core-0.9.5.jar:0.9.5] backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:99) ~[storm-core-0.9.5.jar:0.9.5] 原因: java.lang.RuntimeException: java.nio.BufferUnderflowException at storm. kafka.KafkaUtils.fetchMessages(KafkaUtils.java:177) ~[stormjar.jar:na] at storm.kafka.trident.TridentKafkaEmitter.fetchMessages(TridentKafkaEmitter.java:132) ~[stormjar.jar:na] at storm.kafka. trident.TridentKafkaEmitter.doEmitNewPartitionBatch(TridentKafkaEmitter.java:113) ~[stormjar.jar:na]

原因: java.nio.BufferUnderflowException: null at java.nio.HeapByteBuffer.get(HeapByteBuffer.java:151) ~[na:1.8.0_05] at java.nio.ByteBuffer.get(ByteBuffer.java:715) ~[ na:1.8.0_05] kafka.api.ApiUtils$.readShortString(ApiUtils.scala:40) で ~[stormjar.jar:na]

私が使用しているストームのバージョンは 0.9.5 で、kafka のバージョンは 2.11 です。ここで何が問題を引き起こしているのか理解できません。

trident でトポロジを動的に作成する方法はありますか? 誰か例を教えてください。