問題タブ [trident]

カフカからタプルを取得するトポロジがあります。

トポロジーは大まかにこんな感じです。

ドキュメントから私が理解しているのは、アグリゲーターではすべてのタプルに対して集約メソッドが呼び出され、すべてのタプルが処理されると完全なメソッドが呼び出されるということです。

初期化および集約メソッドにデバッグ sysout を入れました。トポロジーを開始すると、init および aggregate メソッドからの連続した sysout を確認できます。この動作は予期されたものですか? (Kafka にはデータがないことに注意してください)

HBaseState をクエリする Trident トポロジを構築しています。org.apache.storm.hbase パッケージを使用しています。

私の理解 (間違っている場合は訂正してください) は、HBaseQuery が特定の rowKey のすべての列値 (または ProjectionCriteria で指定された値) を読み取り、各列を Fields("columnName","columnValue") で個別に出力することです。

たとえば、ペットのテーブルがあり、rowKey がペットの名前で、"type" の列と "age" の列がある場合、stateQuery は Values("Fido") を含む入力タプルを受け取り、2 つの別々の出力を出力します。タプル:

Values("フィド","タイプ","犬")

Values("フィド","年齢",11)

いくつかの質問：

1. 1 つのクエリで複数の列から値を取得する方法はありますか? つまり、Fields("Name","column1Value","column2Value") で単一の出力を取得できますか?

2. 複数の列から 1 つのタプルに値を取得する方法がある場合、それらが異なる型 (たとえば、1 つは文字列、1 つは整数) である場合でも実行できますか?

最終的に、私の目標は、Fields("Name") で入力タプルを取得し、Fields("Name","Type","Age") で単一の出力タプルを取得できるようにすることです。たとえば、Values("Fido","Dog ",11) および Values("Mr. Kibbles","Cat",4). 上記を使用できない場合、どのように可能ですか？

どんな助けでもTIA！

また、Trident を使用する場合、バッチが 1 つのボルトで処理される場合、バッチは異なるワーカーで実行できますか?

1 つのバッチを複数のサーバーで処理できるようにしたいのですが、どうすれば作成できますか?

つまり、並列度が 2 の場合、ボルトは 2 つの異なるサーバーで実行され、並列度が 3 の場合、ボルトは 3 つの異なるサーバーで実行されます。すべてのタスクを 1 つのサーバーだけで実行したくないので、これは私にとって重要です。それでは遅すぎます。

まず、私は Storm/Trident を初めて使用し、すでに何時間も問題に取り組んでいます。

私が持っているのは、1 つのパーティションを持つ 1 つの Kafka トピックです。プロデューサーは x ミリ秒ごとにこのトピックにタプルを送信します。TransactionalTridentKafkaSpout がこのトピックから読み取り、一部の Trident オペレーターがそれらを処理します。トポロジ全体がローカル モードで実行されています (リモート モードは今のところテストされていません)。

トポロジのメイン コードは次のとおりです。

今私が抱えている問題は、プロデューサーのメッセージ間隔が短いほど、実行されるオペレーターが少なくなることです。
たとえば、プロデューサがそれぞれ 100 ミリ秒の間隔で 200 のタプルを送信する場合、すべてのオペレーターは 200 のタプルすべてを正しく処理しますが、間隔が 20 ミリ秒に設定されている場合、たとえば、オペレーターは次の数のタプルのみを処理/実行します:
CustomCombinerAgg1: 200
CustomCombinerAgg2: 50
CustomBaseFilter1: 50
CustomCombinerAgg3: 150
CustomCombinerAgg4: 180
CustomBaseFilter2: 60

私が理解している限り、(トランザクション) Trident は正確に 1 回の処理を保証し、タプルの新しいバッチは、前のバッチが完全に処理された後にスパウトから取得する必要があります。これはここでは当てはまらないようで、むしろ最初のオペレーターである CustomCombinerAgg1 が速度を指示し、その後のオペレーターは指定された時間内にすべてのタプルを処理できないように見えますか?

私が期待するのは、すべてのタプルに対してすべてのオペレーターが適切に実行され、タプル/バッチがすべてのオペレーターによって処理されると、次のオペレーターがフェッチされることです。これは Trident を使用した場合に当てはまりませんか? 私は何か間違ったことをしていますか？どうすればこの動作を達成できますか?
Trident は、タプルが完全に処理されたことをどのようにして知るのでしょうか? 私の知る限り、Storm ではタプルを ack() する必要がありますが、Trident オペレーターには OutputCollector がないため、ack() を呼び出すことはできませんか? 私の問題は何とかこれに結びついていますか？

ありがとう。