import包：

import org.apache.spark.{SparkConf, SparkContext}import org.apache.spark.rdd.RDDimport org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}import org.apache.spark.sql.{DataFrame, Row, SQLContext}

样例类：

case class Person(id:Int,name:String,age:Int)

主函数：def main(args: Array[String]): Unit = {  val sparkConf: SparkConf = new SparkConf().setAppName(this.getClass.getName).setMaster("local")  val sparkContext = new SparkContext(sparkConf)  val sqlContext = new SQLContext(sparkContext)  val rdd: RDD[String] = sparkContext.textFile("C:\\Users\\dummy\\Desktop\\person.txt")  val lineRdd: RDD[Array[String]] = rdd.map(_.split(" "))  InferringSchema(lineRdd,sqlContext)  SpecifyingSchema(lineRdd,sqlContext)  sparkContext.stop()}第一种方法：（需要创建样例类）

/**  * 通过反射推断Schema  * @param lineRdd  * @param sqlContext  */def InferringSchema(lineRdd: RDD[Array[String]],sqlContext:SQLContext): Unit ={  //将RDD和case class关联  val personRdd: RDD[Person] = lineRdd.map(x=>Person(x(0).toInt,x(1),x(2).toInt))  //导入隐式转换，如果不导入无法将RDD转换成DataFrame  import sqlContext.implicits._  //将RDD转换成DataFrame  val personDF: DataFrame = personRdd.toDF()  personDF.show()  //注册一张临时表  //personDF.registerTempTable("person")  //val personDF2: DataFrame = sqlContext.sql("select * from person")  //将结果以JSON的方式存储到指定位置  //personDF2.write.json("C:\\Users\\dummy\\Desktop\\out")  //personDF2.show()}

第二种方法：

/**  * 通过StructType直接指定Schema  * @param lineRdd  * @param sqlContext  */def SpecifyingSchema(lineRdd: RDD[Array[String]],sqlContext:SQLContext): Unit ={  //通过StructType直接指定每个字段的schema  val schema=StructType(    List(      /**StructField只需传入前面两个参数即可        * name: String,        * dataType: DataType,        * nullable: Boolean = true,        * metadata: Metadata = Metadata.empty)        */      StructField("id",IntegerType),      StructField("name",StringType),      StructField("age",IntegerType)    )  )  val rowRdd: RDD[Row] = lineRdd.map(x=>Row(x(0).toInt,x(1),x(2).toInt))  val personDF: DataFrame = sqlContext.createDataFrame(rowRdd,schema)  //personDF.show()  personDF.registerTempTable("person")  val personDF2: DataFrame = sqlContext.sql("select * from person")  //personDF2.write.json("C:\\Users\\dummy\\Desktop\\out")  personDF2.show()}对比：