問題タブ [trident]

現在、クラスター化されたトポロジ モードで Apache Storm 0.9.5 を使用して、Amazon Kinesis レコード (spout) を処理し、それらを Redshift データ ウェアハウス (bolt) に保存しています。私たちの Storm クラスターは AWS にデプロイされており、1 つの Nimbus + UI ノード、1 つの ZooKeeper ノード、および 3 つのスーパーバイザー + ログビューアー ノードで構成されています。当社のトポロジ設定は、複数の Kinesis ストリームの処理をサポートし、ストリームごとに以下が含まれます。

• 着信レコードをリッスンする 1 つの Kinesis ストリーム スパウト
• データ ウェアハウスにレコードを挿入するための 1 つの Redshift ボルト

トポロジー：

このシステムの弱点は、入力メッセージの一度だけの処理を保証できないことであり、その結果、同じビジネス キーを持つ複数のレコードがターゲット データベースに挿入されます。問題の規模を把握するために、制御されたテストを実行したところ、すべての入力レコードの約 3 分の 1 が複数回処理のために送信されたことがわかりました。

このスレッド(現在は未回答) に従って、一度だけの処理を保証するために Trident を使用することも検討しましたが、べき等性をシステムに組み込むことがより重要であるという結論に達しました (少なくとも-once セマンティクス) ではなく、複雑さを追加し、パフォーマンスを低下させ、この別の記事で提案されているように状態を生成します。

現在、クラスタリングをサポートする方法で既存のトポロジ内にべき等性を実装する最善の方法についてアドバイスを求めています。これまでのところ、タプル メッセージ ID によって値をキーにする RedisBolt の導入に傾いています。Apache Storm を使用してこれを達成するための既存のパターンはありますか?

ストームトライデントは初めてです。これが私の問題です：

私は Kafka キューからメッセージを消費し、の実装を使用CombinerAggregatorして集計を行い、Cassandra でデータを永続化しています。

これをローカルで実行すると、数値が正しく入力されていることがわかります。しかし、実稼働環境では、ほとんどの場合、これらの集計が 0 になっています。これは間違っています。これらのケースにも値が必要です。また、場合によっては、ゼロ以外の数字が表示されますが、これは正しいものです。

numOfWorkersトポロジーを 4 と2に設定しましparallelismHintた。何がうまくいかないか分かりますか? または、この問題をデバッグする方法を提案してください。

HDFS にメッセージを書き込もうとしているときに、Trident JMS スパウトから Trident hdfs ボルトに JMS テキスト メッセージを送信しようとすると、Storm Trident トポロジが次のようなエラーをスローします。

この種のエラーに遭遇した人はいますか? もしあなたがそれに直面した場合、これは何だと思いますか?

ストームのバージョンは 0.9.1.2.1.3.0-563 です

Storm-Hdfs は 0.9.4 です

ローカル クラスターを使用して Trident トポロジを再実行すると、次のエラーが発生します。

前のトランザクション ID はカウンター値と共に Cassandra に保存されますが、最初はトランザクション ID が常に 1 に設定されているように見えるため、単純な検証ステップ (現在のトランザクション ID が前のトランザクション ID より大きい) は常に失敗します。

Trident に現在のバッチ ID を保持させる方法は?

ありがとう！

2 つの異なるストリームに 2 つの異なるスパウトを持つストーム トライデント トポロジを実行しています。私のスパウトは JMS スパウトであり、HDFS State を使用してタプルを永続化します。

スパウトを 1 つだけ実行すると、正常に動作します。HDFS の JMS キューに投稿されたすべてのレコードを取得しています。

2 つの異なるキューに接続する 2 つのスパウトを使用してトポロジを実行しているときに、QUEUE に投稿したものよりも少ないレコードを取得しています。ここで何か間違ったことをしていますか？私がこれを行っている方法に問題がある場合はお知らせください。

Storm Trident で同じボルトから複数のストリームに放出するにはどうすればよいですか?

いくつかの計算を行うボルトがあり、その結果に基づいて、ある値をあるストリームに渡し、他の値を別のストリームに渡したいと考えています。

Storm (Trident ではない) では、次の方法でそれを実現できます。

ストリームを複数のストリームに分割します。

次に、結果に基づいて次のように発行します。

次に、予想されるストリームをリッスンして残りの作業を行います。

では、Storm Trident を使用して同じ動作を実現するにはどうすればよいでしょうか?

1つのオプションは、同じストリームに対して「each」を呼び出し、同じボルトを実行し、そのストリームに発行したいものに基づいてのみ発行するか、別のオプションは、キーと値のペアを発行し、キーに基づいてストリームをフィルタリングします(type1など) type2、エラーなど)、再び複数のストリームを作成します。しかし、どれも私には良いデザインとは思えません。それを達成するための最良の方法は何ですか？

プロジェクトで Apache Storm の TridentTopology を使用したいと考えています。私は、storm.trident.Stream クラスの .each() 関数を理解するのが難しいと感じています。以下は、参照用のチュートリアルに記載されているサンプルコードです。

メソッド .each() のシグネチャがわかりませんでした。以下は私が理解したものです。私が間違っている場合は修正してください。また、私の知識についてさらに情報を提供してください。

。各（）

• 最初のパラメーターは、スパウトから出力された値に関連付けられたキーであり、スパウトの getOutputFields() メソッドから返されるフィールドを受け取ります。なぜそのパラメーターが使用されるのかはまだわかりません。
• 2 番目のパラメーターは、BaseFunction を拡張するクラスです。タプルを処理します。
• 3 番目のパラメーターの理解は、1 番目のパラメーターと似ています。