データフレームでRDD演算を実行

Question

私は10フィールドのデータセットを持っています。これらのDataFrameでRDD操作を実行する必要があります。

がここ

... map、 flatMap、などのようなRDD操作を実行することが可能であることは私のサンプルコードは次のとおりです。

df.select("COUNTY","VEHICLES").show(); 

これは私のdataframeであると私は、このdataframeがRDDに変換して、いくつかのRDD操作を操作する必要がありますこの新しいRDDでここ

コードiはRDDに変換した後RDD

RDD<Row> java = df.select("COUNTY","VEHICLES").rdd(); 

に変換データフレーム午前方法であり、iはRDDの結果を見ることができないです、私は上記のすべての場合において

java.collect(); 
java.take(10); 
java.foreach(); 

を試みI結果を得ることができませんでした。

私を助けてください。

Answer 1

val myRdd : RDD[String] = ds.rdd 

Dataset to RDDを確認してください。あなたのケースではlazy val rdd: RDD[T]

その呼び出し.rdd後に選択を行うことで、レコードの選択とそれがスパーク1.6についてRDD

Answer 2

に変換ウィルDATAFRAMEを作成する：あなたが見ることができません

あなたはそれが何をするかRDDにDataframeを変換しているように、結果のは、あなたがこれらのいずれかをしようとすると、それが故にRDD[Row]

に変換します：

java.collect(); 
java.take(10); 
java.foreach(); 

結果はArray[Row]となり、結果を得ることはできません。

ソリューション：

あなたはそれぞれの値に行を変換して、ここのようにそれのうちRDDを取得することができます：

val newDF=df.select("COUNTY","VEHICLES") 
val resultantRDD=newDF.rdd.map{row=> 
val county=row.getAs[String]("COUNTY") 
val vehicles=row.getAs[String]("VEHICLES") 
(county,vehicles) 
} 

そして今、あなたが取得するforeachとcollect機能を適用することができます値。

P.S .:コードはScalaで書かれていますが、私がしようとしていることの本質を得ることができます！

Answer 3

スパーク2.0以降、RDD操作を使用するためにtoDS関数を使用してDataFrameをDataSetに変換できます。
マスタリングスパーク2.0についての詳細article

Answer 4

rddからデータを読み取る前にrddを持続させてみてください。

val finalRdd = mbnfinal.rdd 
finalRdd.cache() 
finalRdd.count()