問題タブ [rdd]

私のクエリは、ファイルから入力を読み取り、scala を使用してファイルのデータ行を List[Map[Int,String]] に変換することです。ここでは、入力としてデータセットを指定します。私のコードは、

しかし、このコードを.csvファイルからの入力を受け入れるように変更するにはどうすればよいですか?

アプリケーションを実行しましSpark clusterたが、次のerrorような結果になりました：

誰かがそれを修正する方法を教えてもらえますか?

多くの列を持つデータ テーブルの row_numbers の完全なリストを生成する必要があります。

SQL では、これは次のようになります。

ここで、Spark に (K, V) という形式の RDD があるとしましょう。V=(col1, col2, col3) の場合、私のエントリは次のようになります。

sortBy()、sortWith()、sortByKey()、zipWithIndexなどのコマンドを使用してこれらを注文し、正しいrow_numberを持つ新しいRDDを作成したい

(かっこは気にしないので、代わりに (K, (col1,col2,col3,rownum)) の形式にすることもできます)

どうすればいいですか？

これが私の最初の試みです：

また、関数 sortBy を RDD に直接適用することはできませんが、最初に collect() を実行する必要があることに注意してください。その後、出力も RDD ではなく、配列になります。

もう少し進行状況がありますが、まだ分割されていません。

各パーティションの要素がすべて単一の大きなオブジェクトへのアクセスを共有するスパークRDDを永続化しようとしています。ただし、このオブジェクトはメモリに数回格納されるようです。私の問題を、200 要素しかない単一のパーティションのおもちゃのケースにまで減らします。

ログに示されているように、これは予想される量のメモリを消費します。

storage.MemoryStore: ブロック rdd_1_0 が値としてメモリに保存されます (推定サイズ 38.2 MB、空き容量 5.7 GB)

ただし、200 は、これが該当する要素の最大数です。nElements=201利回りの設定:

storage.MemoryStore: ブロック rdd_1_0 が値としてメモリに保存されます (推定サイズ 76.7 MB、空き容量 5.7 GB)

これは何が原因ですか？このマジック ナンバー 200 はどこから来て、どうすれば増やすことができますか?

明確化のための編集：

関数に println を追加すると、関数が実際に 1 回だけ呼び出されることがわかります。さらに、次を実行します。

..10000000 個の要素すべてが実際に同じオブジェクトを指していることを明らかにしているため、データ構造は本質的に正しく動作します。問題が発生するのは、nExamples がはるかに大きい (たとえば 20000) 場合であり、永続化できません。

storage.MemoryStore: rdd_1_0 をメモリにキャッシュするのに十分なスペースがありません! (これまでに計算された 6.1 GB)

設定すると、推定サイズ 1907.4 MBnExamples=500を示すメモリ内の rdd が正常に保持されますが、メモリ使用量の実際の増加はこれよりもはるかに少ないことがわかります。

文字列配列である列を値を連結して文字列に変換しようとしていますが、その特定の列を文字列の配列として関連付けることができない方法があります:

そして、私は次のようになります：

4/11/23 10:40:16 エラー executor.Executor: ステージ 2.0 (TID 5) のタスク 1.0 での例外 java.lang.ClassCastException: scala.collection.mutable.ArrayBuffer は [Ljava.lang.String; にキャストできません。

私は何を間違っていますか？

ありがとう、
ダニエル

Scala Spark アプリでテストをビルドしましたが、テストの実行中に Intellij で以下の例外が発生しました。SparkContext を使用しない他のテストは正常に実行されています。「sbt test-only」を使用して端末でテストを実行すると、sparkcontext を使用したテストは機能しますか? sparkcontext を使用したテスト用に intellij を特別に構成する必要がありますか?

例外またはエラーにより実行が中止されました: org.apache.spark.rdd.ShuffledRDD.(Lorg/apache/spark/rdd/RDD;Lorg/apache/spark/Partitioner;)V java.lang.NoSuchMethodError: org.apache .spark.rdd.ShuffledRDD.(Lorg/apache/spark/rdd/RDD;Lorg/apache/spark/Partitioner;)V at org.apache.spark.graphx.impl.RoutingTableMessageRDDFunctions.copartitionWithVertices(RoutingTablePartition.scala:36) org.apache.spark.graphx.VertexRDD$.org$apache$spark$graphx$VertexRDD$$createRoutingTables(VertexRDD.scala:457) at org.apache.spark.graphx.VertexRDD$.fromEdges(VertexRDD.scala:440) org.apache.spark.graphx.impl.GraphImpl$.fromEdgeRDD(GraphImpl.scala:336) で org.apache.spark.graphx.impl.GraphImpl$.fromEdgePartitions(GraphImpl.scala:282) で org.apache.spark .graphx.GraphLoader$.edgeListFile(GraphLoader.scala:91)

私のRDDには、定数値を持つ列があるかもしれません。つまり、一部の列の分散がゼロになる場合があります。私の目的は、RDD からそのような列をすべて削除することです (そして、最終的に残りの列の共分散行列を計算します)。どうやってやるの？

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

私は頭を包み込むことができない繊細なスパークの問題を抱えています。

2 つの RDD があります (Cassandra から来ています)。RDD1 にはデータが含まれActions、RDD2 にはHistoricデータが含まれます。両方とも、一致/結合できる ID を持っています。しかし問題は、2 つのテーブルが N:N の関係にあることです。Actionsには同じ ID を持つ複数の行が含まれていHistoricます。両方のテーブルの日付の例を次に示します。

Actions 時間は実際にはタイムスタンプです

Historic set_at は実際にはタイムスタンプです

このような結果が得られるように、これら 2 つのテーブルをどのように結合できますか?

巨大なデータセットに対して多くの反復を行わなければ、適切な解決策を思いつくことはできません。私は常にセットから範囲を作成することを考えなければならず、計算を行うために（11:00 - 12:15）などの範囲Historicに収まるかどうかを何らかの方法で確認する必要があります。Actionsしかし、それは私にはかなり遅いようです。それを行うより効率的な方法はありますか？この種の問題は人気があるように思えますが、これに関するヒントはまだ見つかりませんでした。スパークでこの問題をどのように解決しますか?

これまでの私の現在の試み（途中で完了したコード）