問題タブ [spark-cassandra-connector]

Kafka + Cassandra を Spark 1.5.1 に接続したい。

ライブラリのバージョン:

アプリへの初期化と使用:

次のように Cassandra にスキーマを作成します。

また、準備jarができたら、いくつかの戦略を作成します。

問題は関連していると思います

使用に縛られますMergeStrategy.last。

何か案は？

例外があります:

Spark でクエリのリストを並列処理しています。私の RDD は、各クエリのデータ ソース属性に基づいてオブジェクト タイプが異なるオブジェクトのコレクションにマップされます。String -> function のマップを持っています。各関数は、cassandra に保存されるオブジェクトのコレクションを提供します。例えば：

ここで、processTrendsResponse は Seq[Trends] を返す関数として定義され、processYahooResponse は Seq[Yahoo] を返す関数として定義されます。トレンドと Yahoo 定義はケース クラスとして定義されます。

Trends と Yahoo の両方のタイプに対応するために、processingMethods Map は Map[String, (String) => Seq[Any]] として定義されました。しかし、スパークで saveToCassandra アクションを実行すると、例外でゲートされます-

前もって感謝します

私は Spark と Cassandra の新しい学習者です。重大なパフォーマンスの問題に直面しています。Spark で 5 秒ごとに Kafka からデータをストリーミングし、JRI を使用して R 言語でデータを分析し、最終的にデータを Cassandra のそれぞれの列ファミリーに保存しています。データを Cassandra に保存するための時間 (ミリ秒単位) は、入力要求の数に応じて急速に増加します [各要求は 200 KB]。

スパークコード:

Spark SQL を使用して、Cassandra データベースからデータを取得しました。

その後、いくつかのフィルタリングを行い、このデータを次のような別の Cassandra テーブルに保存したいと考えています。

DataFrameを新しいテーブルに保存するときに、これらの追加のプロパティを追加するにはどうすればよいですか? また、この例を使用して Cassandra の長い行を分割するベスト プラクティスは何ですか? には 4,000 ～ 6,000 のレコードがあると予想されるDataFrameため、長い行をシャーディングすることは必須ですが、レコードを数えてsharder特定の数のアイテムの を変更することが、Spark または Cassandra でのベスト プラクティスであるかどうかはわかりません。

Cassandra 2.2.3 (localhost で実行) を Spark 1.5.1 および最新バージョンの spark-cassandra-connector (1.5.0-M2) で動作させようとしています。

これが私が使用している基本的なスニペットコードです。キースペースとテーブルはすでに作成されています。

sbt アセンブリでは問題なくコンパイルされますが、アプリを送信すると次のエラーが発生します。

Apache Spark を使用して Cassandra からデータを取得し、データを変換して別の Cassandra テーブルに保存する単純な Java アプリケーションを作成しました。

マシン上にある単一のマスターとスレーブを使用して、スタンドアロン クラスター モードで構成された Apache Spark 1.4.1 を使用しています。

ご覧cacheのcustomersWhoHaventOrderedTheProductとおり、DataFrame を実行した後、 を実行してcountを呼び出しますtoJavaRDD。

私の計算では、これらのアクションは 1 回だけ実行する必要があります。しかし、現在のジョブの Spark UI に移動すると、次のステージが表示されます。

ご覧のとおり、すべてのアクションが 2 回実行されます。

私は何か間違ったことをしていますか？見逃した設定はありますか？

どんなアイデアでも大歓迎です。

編集：

私が電話した後System.out.println(storeCustomerReport.toJavaRDD().toDebugString());

これはデバッグ文字列です:

編集2：

そのため、いくつかの調査と試行錯誤を組み合わせた結果、仕事を最適化することができました。

から RDD を作成し、アクションcustomersWhoHaventOrderedTheProductを呼び出す前にそれをキャッシュします。count()(キャッシュを から に移動しDataFrameましたRDD)。

その後、これRDDを使用してstoreCustomerReport DataFrame.

ステージは次のようになります。

ご覧のとおり、2 つのcountとcacheはなくなりましたが、まだ 2 つの「javaRDD」アクションがあります。toJavaRDDコード内で 1 回しか呼び出していないため、それらがどこから来ているのかわかりません。

Spark に次のコードがあります。

どこ

どこ：

saveToCassandraはオブジェクトのコレクションを期待しSeq[Any]、戻り値の型として両方を含むように使用し、例外Seq[A]でSeq[B]中断saveToCassandraします -scala.ScalaReflectionException: <none>は用語ではありません。この動作の理由は何でしょうか?