問題タブ [spark-cassandra-connector]

1 つのマスターと 2 つのワーカーを持つ 3 ノードのスタンドアロン Spark クラスターと、2 ノードの cassandra リングを使用しています。これは、私がやろうとしていることのサンプル コードです。

から、System.out.printlnクエリの結果としていくつかのデータがあることはわかっていますが、プロジェクトのホームのoutputディレクトリでは、取得しているファイルは と だけで、ファイルは_SUCCESSあり._SUCCESS.crcませんpart-*。これは予想される動作ですか? そうでない場合、どこが間違っていますか？

Apache Spark の初心者であり、Spark で Cassandra データを取得する際に問題に直面しています。

このクエリは、cassandra サーバー上のデータをフィルタリングしていません。この Java ステートメントが実行されている間、メモリが消費され、最終的に spark java.lang.OutOfMemoryError 例外がスローされます。https://github.com/datastax/spark-cassandra-connector/blob/master/doc/3_selection.mdに記載されているように、クエリはクライアント側ではなく cassandra サーバーでデータを除外する必要があります。

cassandra cqlsh でフィルターを使用してクエリを実行している間、正常に実行されますが、フィルター (where 句) を使用せずにクエリを実行すると、予期されるタイムアウトが発生します。したがって、spark がクライアント側でフィルターを適用していないことは明らかです。

クライアント側でフィルターが適用される理由と、サーバー側でフィルターを適用するためにどのように改善できるか。

Windows プラットフォームで Cassandra クラスターの上に Spark クラスターを構成するにはどうすればよいでしょうか??

Spark cassandraコネクタとcassandraドライバを使用するsparkアプリケーションでストレステストを行っています。私のアプリケーションでは、cassandra ドライバーを使用して C* テーブルから最新の値を選択しています。これは、spark ジョブの送信が spark-job サーバー経由で 1 つずつ発生する限り、正常に機能しています。しかし、複数のジョブ送信 (リクエスト数 = 80) が同時に発生すると、以下の例外が発生します。

スパークジョブサーバーで単一のコンテキストを作成してジョブを実行しています。

私のコード

質問

コードで間違っていることはありますか?

この問題に対処するには？

アプリケーション全体のシングルトン クラスタ オブジェクトを作成して共有する必要がありますか?

sc.cassandraTableJava ドライバーを直接使用する代わりに、メソッドを使用する必要がありますか?

このユース ケースでは、常に kafka トピックをリッスンし、Spark ストリーミング アプリの列値に基づいて 2000 列ファミリー (それぞれ 15 列..時系列データ) に書き込む必要があります。ローカルの Cassandra インストールがセットアップされています。これらの列ファミリを作成するには、3 つのコアと 12 ギガの RAM を使用する CentOS VM で約 1.5 時間かかります。私の Spark ストリーミング アプリでは、これらのストリーム イベントを Cassandra に保存するための前処理を行っています。ストリーミング アプリがこれを完了するのに時間がかかるという問題が発生しています。
このキーに基づいて、300 個のイベントを複数の列ファミリー (約 200 ～ 250) に保存しようとしていましたが、アプリはそれらを保存するのに約 10 分かかります。キーでグループ化されたこれらのイベントを画面に出力するのに1分もかからないため、これは奇妙に思えますが、それらをCassandraに保存する場合にのみ時間がかかります。Cassandra に 300 万件のレコードを保存しても問題はありませんでした。3 分もかかりませんでした (ただし、これは Cassandra の単一の列ファミリーに対するものでした)。

私の要件は、可能な限りリアルタイムであることですが、これはどこにも近いようには思えません。実稼働環境では、3 秒ごとに約 400 のイベントが発生します。

Cassandra の YAML ファイルまたは cassandra-connector 自体に変更を加える必要がある調整はありますか

Cassandra 2.1、Spark 1.1、spark-cassandra-connector 1.1

キーと値のペアの非常に背の高い列ファミリがあります。また、そのCFから選択したいキーのRDDもあります

私がやりたいことは次のようなものです

ただし、マップ内の Spark コンテキストを参照すると、NPE が発生します。完全な tallTable から RDD を作成し、ID に結合することもできますが、これは非常に遅い操作であり、回避したいと考えています。

このように Cassandra から一連のキーを読み取る方法はありますか?

Cassandra の Reading パフォーマンスのベンチマークを行っています。テストセットアップのステップで、1 / 2 / 4 の ec2 インスタンスとデータノードを持つクラスターを作成しました。1 億のエントリ (約 3 GB の csv ファイル) を含む 1 つのテーブルを作成しました。次に、spark-cassandra-connector を使用してデータを RDD に読み込む Spark アプリケーションを起動します。

ただし、動作は次のようにする必要があると考えました。Cassandra (Spark では同じインスタンス数) を使用するインスタンスが多いほど、読み取りが速くなります! 書き込みでは、すべてが正しいようです (クラスターが 2 倍大きい場合、~2 倍速くなります)。

しかし: 私のベンチマークでは、2 または 4 インスタンス クラスターよりも 1 インスタンス クラスターの方が読み取りが常に高速です!!!

私のベンチマーク結果:

クラスターサイズ 4: 書き込み: 1750 秒 / 読み取り: 360 秒

クラスターサイズ 2: 書き込み: 3446 秒 / 読み取り: 420 秒

クラスターサイズ 1: 書き込み: 7595 秒 / 読み取り: 284 秒

追加の試行 - Cassandra-Stress ツールを使用

Cassandra クラスター (サイズ 1 / 2 / 3 / 4 ノード) で「cassandra-stress」ツールを起動したところ、次の結果が得られました。

結果: 4 つまたは 8 つのスレッドの場合、単一ノード クラスターは、より大きなクラスターと同じか、それよりも高速です!!!

ダイアグラムとして
の結果: データセットはクラスタ サイズ (1/2/3/4)、x 軸はスレッド、y 軸は ops/sec です。

--> ここで質問: これらの結果はクラスター全体の結果ですか、それともローカル ノードのテストですか (つまり、リングの 1 つのインスタンスのみの結果ですか)?

誰かが説明できますか？ありがとうございました！

私は Apache Spark の使用方法を学ぼうとしてきましたが、(datastax spark-cassandra-connector を使用して) Cassandra から列のすべての値を合計しようとすると問題が発生します。私が試したすべての結果はjava.lang.OutOfMemoryError: Java heap spaceになります。

スパークマスターに提出するコードは次のとおりです。

現在、クラスターには 1 つの Spark ワーカー ノードしかありません。テーブルのサイズを考えると、一度にすべてをメモリに収めることができない可能性は間違いありません。ただし、spark はコマンドを遅延評価することになっているため、これが問題になるとは思いませんでした。列内のすべての値を合計しても、テーブル全体が一度にメモリに存在する必要はありません。

私はこのトピックの初心者なので、なぜこれがうまくいかないのか、正しく行う方法を教えていただければ幸いです。

ありがとう