問題タブ [spark-cassandra-connector]

Spark Cassandra Connector を介して以下を適用しています。

私が常に変更する唯一のパラメータが「link_id」であることを考えると、上記の一連のクエリをより効率的に適用する方法があるかどうかを尋ねたいと思います。

「link_id」の値は、Cassandra の「records」テーブルの唯一のパーティション キーです。Cassandra v.2.0.13、Spark v.1.2.1、Spark-Cassandra Connector v.1.2.1 を使用しています。

これらのクエリを適用し、「link_speed_records」を SparkRDD として取得するために、Cassandra セッションを開くことが可能かどうかを考えていました。

データ分析のためにspark-Cassandra-connectorでApache Sparkを使用しています。そのために、300万を超えるレコードを含むCassandraからテーブルのJavaRDDを作成しました。

フィルタリング後に行は返されません。だから私が申し込むと、

カウントが 0 になるまでに約 2 分かかりました。

datastax が提供する spark-cassandra-connector 1.1.0 を使用しています。興味深い問題に気付きましたが、なぜこのようなことが起こっているのかわかりません: cassandra コネクタをブロードキャストしてエグゼキュータで使用しようとすると、構成が無効であることを示す例外が発生し、0.0.0 で Cassandra に接続できません。

スタックトレースの例:

しかし、ブロードキャストせずに使用すると、すべて正常に動作します。

私にとって奇妙なことは、ドライバー側では値が適切な構成をブロードキャストしましたが、エグゼキューター側ではそうではありませんでした。

運転席側:

エグゼキュータ側:

なぜそのような方法で動作しているのか、エグゼキューター側で使用できる方法で Cassandra コネクタをブロードキャストする方法を誰かが説明できますか?

更新1.2.3 バージョンのコネクタにも同じ問題があります。

スパークのバージョン: 1.4.0 カサンドラのバージョン: 2.1.8

Spark と Cassandra の両方をブリッジするために、datastax Spark Cassandra コネクタを使用しています。6 つの異なるワーカーで実行されている Spark で 6 つのノードを使用しています。これを支援する 2 つの Cassandra ノードがあります。

列ファミリー (CassandraUtil.javaFunctions(sc).cassandraTable("keyspace","columnfamily").count()) 内の行数のカウントを実行するサンプル アプリケーションを試しました。

ここで、この 1 つのジョブをマスターにディスパッチすると、Spark クラスターの 2 つのワーカー ノードでジョブが実行されました (イベント タイムラインから取得)。

質問

1. 私は単一の仕事を派遣しました。なぜ二人の作業員で行われたのですか？一人の労働者が主人のように振る舞うようなものですか？
2. 1 人のワーカーで逆シリアル化時間が非常に長いことがわかりました。他のワーカーはかなり速く仕事を完了しました (1 人は 40 秒、2 人は 1 秒かかりました)。これに光を当てることができますか？
3. どちらのワーカーも Cassandra との接続を確立したようで、結果を返しています。したがって、私の見解では、どちらも同じ仕事をしています。これに光を当てることができますか？
4. Cassandra を使用したこの分散領域のどこに RDD の実装が適合するのか、私はまだ疑問に思っています。誰かがこれに光を当てることができますか? 複数のワーカーは、Cassandra のどのパーティションで作業する必要があるかをどのように知るのでしょうか? たとえば、6 つのワーカー間で 10,000 のパーティションを分割できる場合はどうすればよいでしょうか? フェッチはすべて1人のワーカーで行い、処理は6人で行うようなものですか？その場合でも、実行ロジックはすべてのワーカー (Cassandra からのフェッチとプロセス) で同じままです。Spark はこれをどのように行いますか?
5. Cassandra で Spark を使用する本当の利点を知りたいです。それはメモリ管理レベルですか、それとも他の利点がありますか?

編集

走行中の写真を追加しました。10個の異なるパーティションがあります。これは単純なカウント操作です。

私の質問はまだパズルのままです。

添付ファイルをご覧いただければ、お分かりいただけると思います。これは、私のスパークマスターに送信された単一のジョブ用でした。2 つの異なるエグゼキュータでどのように動作するのか不思議です。両方のエグゼキュータが同じバイト数を返しています。つまり、どちらも cassandra から 10 個のパーティションすべてを取得したことがわかります。これが起こる方法である場合、spark は cassandra よりも何を提供してくれますか? または、10 個のパーティションが 2 つの異なるワーカーによってフェッチされるように、別の方法でフェッチする必要がありますか?

ジョブをスパーク マスターにトリガーするアプリケーションがあります。しかし、ジョブを実行している IP アドレスを確認すると、spark ワーカー IP ではなく、アプリケーション IP が表示されます。したがって、私が理解していることから、RDD を呼び出すと、動作する Spark ワーカーが生成されます。

しかし、私の質問はこれです。

ワーカーが 2 に対して何かをしているのに、1 に対しては何もしていないのを見ました。

これは、Cassandra からのフェッチと、そこからの RDD の作成がすべてアプリケーションで行われることを意味するのでしょうか?

その場合、2 は 2 つのワーカーに対してジョブをトリガーします。その場合、Cassandra から再度フェッチしてカウントを処理しますか?

誰かがこれを明確にできますか??

編集

1. 提供された回答によると、カウント呼び出しがワーカーの機能をトリガーする場合、ローカルでRDDを作成するexecuteSQLの使用は何ですか? クエリを実行して、データの Cassandra データセットを作成しますか? その場合、Cassandra からのクエリは 2 回行われますか?

2. Spark が Cassandra の 10 パーティションの計算を 4 つのワーカーに自動的に分散する場合、誰が結果を集計しますか? マスターは配信をしているだけです。それで、それも集約されますか？

3. RDD をキャッシュせずに別のカウント操作を行うとどうなりますか? Spark は、特定のパーティションに対して以前に使用されたのと同じワーカーを使用しようとし、そのノードの結果の RDD に追加しようとします。このパーティション データを再度取得するには、Cassandra にクエリを実行する必要があると思いますか? これについて明確に説明できますか？

4. RDD をキャッシュするとどうなりますか? RDD はワーカーに格納され、すべての操作に使用されますか? その場合、データセットをメモリに保存して処理するのとどう違うのでしょうか? これについても権利があるかどうか教えてください。

常に 6 つのタスクを実行する 5 ノードの Cassandra クラスターから 500m 行を読み取る単純な Spark ジョブがあり、各タスクのサイズが原因で書き込みの問題が発生しています。input_split_size を調整してみましたが、効果がないようです。現時点では、テーブル スキャンの再分割を余儀なくされていますが、これはコストがかかるため理想的ではありません。

いくつかの投稿を読んだ後、起動スクリプト (以下) で num-executors を増やそうとしましたが、効果はありませんでした。

Cassandra テーブル スキャンでタスクの数を設定する方法がない場合は、それで問題ありません..しかし、ここで何かが欠けているのではないかと常に感じています。

Spark ワーカーは、それぞれに 2 TB SSD を備えた 8 コア、64 GB サーバーである C* ノード上に存在します。

起動スクリプト:

編集 - Piotrの答えに続いて:

sc.cassandraTable の ReadConf.splitCount を次のように設定しましたが、生成されるタスクの数は変わりません。つまり、テーブル スキャンを再分割する必要があります。私はこれについて間違った考えをしていて、再分割が必要であると考え始めています。現在、このジョブには約 1.5 時間かかっています。テーブル スキャンをそれぞれ約 10 MB の 1000 のタスクに再分割すると、書き込み時間が数分に短縮されました。

私はspark cassandra Javaコネクタを使用してテーブルを照会しているspark/cassandraセットアップを持っています。これまでのところ、1 つの Spark マスター ノード (2 コア) と 1 つのワーカー ノード (4 コア) があります。どちらも conf/ の下に次の spark-env.sh があります。

ここに私のスパーク実行コードがあります:

ここで、最初のノードでマスター スパークを開始し、次に 2 番目のノードでワーカーを開始してから、上記のコードを実行します。ワーカーにエグゼキュータ スレッドを作成しますが、アプリケーション側のログに次のメッセージが表示されます。

同じ設定を維持したまま、マスター サーバーで spark/sbin/start-all.sh を実行すると、最初のノードにマスター インスタンスとワーカー インスタンスが作成されます。繰り返しますが、同じコードを実行し、割り当てられたワーカーがこの新しいワーカーである場合、完全に正常に動作します。

マスター ノードとは異なるノードで実行されている元のワーカーの問題は何ですか?

Spark/Spark Cassandra Connector は初めてです。チームで初めてスパークを試しており、スパーク cassandra コネクタを使用して cassandra データベースに接続しています。

データベースの重いテーブルを使用するクエリを作成しましたが、テーブルへのクエリがすべてのレコードをフェッチするまで Spark タスクが開始されないことがわかりました。

データベースからすべてのレコードを取得するだけで 3 時間以上かかります。

使用するDBからデータを取得します。

すべてのデータのダウンロードが完了していなくても、spark に作業を開始するように指示する方法はありますか?

fetch により多くのスレッドを使用するように spark-cassandra-connector に指示するオプションはありますか?

ありがとう、ココ。

キースペースのテーブルを更新しているときに、scala の spark cassandra コネクタの問題に直面しています

ここに私のコードがあります

このコードを実行すると、このようなエラーが発生します

なぜこれが起こっているのですか？どうすればこれを修正できますか?