問題タブ [trident]

2 つのストリームがあるとします。

通常のマージでは、次のようなストリーム 3 が生成されます。

タプルが発行された順序を維持しながらストリームをマージしたいので、[2,5]時間 1 で[1,3]発行され、時間 2、[3,2]時間 3、および[2,4]時間 4 で発行された場合、結果のストリームは次のようになります。

とにかくこれを行う方法はありますか？私は完全なトライデント初心者で、最近トライデント ベースのプロジェクトに参加したばかりなので、いくつかのサンプル コードを提供していただければ幸いです。

よろしくお願いいたします。

エリ

私はトライデントフレームワークを学んでいます。Trident には、インターフェイスを使用してタプルのステートフル マッピングを実行できるこのメソッドを含む、バッチ内の集計タプル用Streamのメソッドがいくつかあります。しかし残念なことに、他の 9 つの のオーバーロードのように、引数としてのみ を使用して、マップの状態をさらに永続化するための組み込みの対応物は存在しません。AggregatorpersistentAggregate()Aggregator

では、下位レベルの Trident と Storm の抽象化とツールを組み合わせて、目的の機能を実装するにはどうすればよいでしょうか? Javadoc ドキュメントがほとんどないため、API を調べるのはかなり困難です。

つまり、persistentAggregate()メソッドを使用すると、永続的な状態を更新してストリーム処理を終了できます。

ちなみに、永続的な状態を更新し、さまざまなタプルを発行したい:

Stream.aggregate(Fields, Aggregator, Fields)耐障害性を提供​​しません:

フィールドのリスト、つまり{field1、field2、field3、field4}があるとします。field2でいくつかの操作を実行したとします。たとえば、5などの値で各タプル値をインクリメントしたいとします。

Maven エラーを使用して Storm Trident トポロジをコンパイルすると、次のように発生します。

パッケージ MemoryMapState が存在しません

BigData DB (私の場合は Cassandra) に列名 col1、col2、col3、val1、val2 の行があります。

SQLアプローチでは、col1、col2またはcol2、col1またはその他の可能な方法でグループ化できます。このようにして、ツリー階層を簡単に形成できます。

しかし、現在、グループ化をサポートしていないデータを保存するために Cassandra を使用しています。したがって、グループ化と集計を行うために Storm を使用したいと考えています。集計とグループ化を行うサンプルコードをいくつか書きましたが、それを達成できるかどうかについて意見を述べることができません。

データはこんな感じ

Excel ピボットのように、階層 root->child1->child2->child3-val1,val2 を構築したい場合、階層が col1->col2->col3 の場合、次のようになります。

この場合、データは次のようになります

私のトライデント コードのいくつかの行はこのように見えますが、期待どおりに動作していません。

上記の変換を行うために、Trident API サポートの有無にかかわらず、Storm を使用したいと考えています。誰でもそれを達成する方法を教えてもらえますか? プログラムのアイデアは大歓迎です。

Cassandra 2.0.5、Storm バージョン 0.9.0.1 で Storm の Trident を使用する方法を学んでいます。com.hmsonline storm-cassandra 0.4.0-rc4 contrib も使用しています。

私の目標は、id (int)、名前 (text)、および文 (text) 列を持つテーブルにいくつかのテキスト行を挿入することです。id と name は主キーです。

にはpartitionPersistが必要でStateUpdater、そのために を使用していcom.hmsonline.storm.cassandra.trident.CassandraUpdater<K, C, V>ます。しかし、どうやら、入力として2つではなく1つのキーしか取得できないようです（IDと名前が必要です）。タプル マッパー ( TridentTupleMapper) も 1 つのキーを使用します。

何かが足りないのかもしれませんが、複数の列をキーとして定義するにはどうすればよいですか?

Cassandra 2.0.5、Storm バージョン 0.9.0.1 のテーブルに単純な行を挿入しようとしています。

私のテストは次のとおりです。

ID (int) と文 (text) 列で構成されるテーブルがあります。id は主キーです。

私のスパウトは文を生成し、ID (コードの静的インクリメント) を追加します。

これは私のトポロジです:

MyTridentTupleMapper のコード:

https://github.com/guywald/trident-cassandra-read-write-examples/blob/master/src/test/java/com/guywald/storm/trident/cassandra/MyTridentTupleMapper.java

次の例外が発生します。

なぜこれが返されるのかわかりません。助けていただければ幸いです。