rdd

-3熱

1答えて

に浮いているように私の火花RDDこの totalDistance=flightsParsed.map(lambda x:x.distance) totalDistance.take(5) [1979.0, 640.0, 1947.0, 1590.0, 874.0] のようになりますが、私はそれを減らす実行すると totalDistance=flightsParsed.map(lambd

1熱

1答えて

スパーク - 新しいrddとしてrddのトップNを取得する方法（ドライバで収集しないで）

トップN値の1つを持つRDDをフィルタリングする方法は不思議です。通常、私はRDDをソートし、そのようRDDをフィルタリングするために放送することができ、N番目の値を見つけるためにドライバ内の配列としてtop Nのアイテムを取る： val topNvalues = sc.broadcast(rdd.map(_.fieldToThreshold).distict.sorted.take(N)) v

-2熱

1答えて

Spark RDD：並列操作のためにデータを共有する方法

スパークについての私の理解は、RDDでreduce操作を実行すると、異なるノードによって並列に操作され、結果がマスターノードによって蓄積されることです。これらの操作は並行して実行されるため、結果は全体として利用可能であり、処理中に実行される更新に依存することはできません。たとえば、共有キャブアプリケーションを設計していると、トリップの開始位置と実際の位置異なるキャブ。私は簡単に旅行開始ポイントから

0熱

1答えて

KVPのRDD値をPySparkの整数に変換する

RDDのキーと値のペアの列[2]の値を文字列から整数に変換しようとしているので、それらを合計して平均を計算できます。私は列には[2]に自動的にマッピングフェーズの間の整数として表示していしようとしましたが、私はエラーを取得しています、「TypeError例外：『タイプ』オブジェクトが属性を持っていない『のGetItem』」 textfile = sc.textFile("hdfs://...

0熱

1答えて

私のRDDはいつペナルティを受けることができますか？

私は、何度か呼び出されるメソッドを持っています。このメソッドは、次のようになります。 def separateGoodAndBad(myRDD: RDD[String]): RDD[String] = { val newRDD = myRDD.map(......) //do stuff newRDD.cache //newRDD has 2 actions performed

1熱

1答えて

pyspark：1つのRDDを複数のRDDに分割する

回答が見つかりません。私にRDDがある場合 rdd = sc.parallelize([('a', [1,2,3]), ('b',[4,5,6])]) 各値はリストです。今、各値はリストの要素の一つであり、キーと一致する場所、それが sc.parallelize([('a',1),('a',2),('a',3),('b',4),('b',5),('b'6)]) なるように、RDDを分割

0熱

1答えて

同じパーティション：データフレームのために、後に、このデータフレーム

のRDDのために、私はデータフレームDFの仕様で複数のウィンドウ関数を使用します。 windowSpec = \ Window .partitionBy(df['customID']) 後、私が使用します。 df.rdd.map((x[0],x[0], x[1], x[2])) and df.mapPartitions(some func) 私のデータフレームのkv

1熱

1答えて

Sparkで文字列の配列を含むRDDからSQLContextデータセットを作成

はRDD[Array[String]]です。私はそれを反復し、隣接する要素を比較したい。これを行うには、RDDからデータセットを作成する必要があります。 import org.apache.spark.sql.SQLContext val sqc = new SQLContext(sc) val lines = sqc.createDataset(data) そして、私は2つの次のエラー

-1熱

1答えて

エラーを解決する方法：値reduceByKeyはorg.apache.spark.rdd.RDD [（Int、Int）]のメンバーではありませんか？

私はapache sparkを学習しており、scala端末上で小さなプログラムを実行しようとしています。私は、次のコマンドを使用して、DFS、糸や履歴サーバを開始しました： start-dfs.sh start-yarn.sh mr-jobhistory-deamon.sh start historyserver 、その後、Scalaのターミナルで、私は次のコマンドを書かれている： va