转换RDD到数据帧在2.0

Question

我想转换RDD在Spark2.0

val conf=new SparkConf().setAppName("dataframes").setMaster("local") 
val sc=new SparkContext(conf) 
val sqlCon=new SQLContext(sc) 
import sqlCon.implicits._ 
val rdd=sc.textFile("/home/cloudera/alpha.dat").persist() 
val row=rdd.first() 
val data=rdd.filter { x => !x.contains(row) } 

data.foreach { x => println(x) } 


case class person(name:String,age:Int,city:String) 
val rdd2=data.map { x => x.split(",") } 
val rdd3=rdd2.map { x => person(x(0),x(1).toInt,x(2)) } 
val df=rdd3.toDF() 


df.printSchema(); 
df.registerTempTable("alpha") 
val df1=sqlCon.sql("select * from alpha") 
df1.foreach { x => println(x) } 

到数据帧，但在toDF我得到一个错误以下（）。 ---> “VAL DF = rdd3.toDF（）”

Multiple markers at this line: 
- Unable to find encoder for type stored in a Dataset. Primitive types (Int, String, etc) and Product types (case 
classes) are supported by importing spark.implicits._ Support for serializing other types will be added in future releases. 
- Implicit conversion found: rdd3 ⇒ rddToDatasetHolder(rdd3): (implicit evidence$4: 
org.apache.spark.sql.Encoder[person])org.apache.spark.sql.DatasetHolder[person] 

如何以上使用toDF（）

Answer 1

有一个简单的错误，我在主方法内定义了case类。删除后，我能够将RDD转换为DataFrame。

package sparksql 

import org.apache.spark.SparkConf 
import org.apache.spark.sql.SparkSession 
import org.apache.spark.sql.Encoders 
import org.apache.spark.SparkContext 

object asw { 

case class Person(name:String,age:Int,city:String) 
def main(args: Array[String]): Unit = { 

val conf=new SparkConf().setMaster("local").setAppName("Dataframe") 
    val sc=new SparkContext(conf) 
val spark=SparkSession.builder().getOrCreate() 
import spark.implicits._ 


val rdd1=sc.textFile("/home/cloudera/alpha.dat") 
val row=rdd1.first() 
val data=rdd1.filter { x => !x.contains(row) } 
val rdd2=data.map { x => x.split(",") } 
val df=rdd2.map { x => Person(x(0),x(1).toInt,x(2)) }.toDF() 
df.createOrReplaceTempView("rdd21") 
spark.sql("select * from rdd21").show() 

} 
} 

Answer 2

Cloudera的&火花2.0转换为数据帧？嗯，并不认为我们支持，但:) :)

无论如何，首先你不需要在RDD上拨打.persist()，所以你可以删除那一点。其次，由于Person是一个案例类，你应该把它的名字大写。

最后，在Spark 2.0中，您不再调用import sqlContext.implicits._隐式构建DataFrame架构，现在您可以拨打import spark.implicits._。这是由您的错误消息暗示。