RDD转Dataset方法

Spark支持两种RDD转化为Dataset的方法。

第一种：Inferring the Schema using reflection

方法：使用反射机制将包含特定类型的对象的RDD转换成Dataset
细节：case class 用来定义数据表的schema ，case class的参数名通过反射会作为列名，且可以包含复杂的数据类型，如：Seq、Array等。RDD通过implicit转换为DataFrame

schema（发音 “skee-muh” 或者“skee-mah”，中文叫模式）是数据库的组织和结构，schemas 和schemata都可以作为复数形式。模式中包含了schema对象，可以是表(table)、列(column)、数据类型(data type)、视图(view)、存储过程(stored procedures)、关系(relationships)、主键(primary key)、外键(foreign key)等。数据库模式可以用一个可视化的图来表示，它显示了数据库对象及其相互之间的关系

import spark.implicits._
//
val peopleDF = spark.sparkContext
  .textFile("examples/src/main/resources/people.txt")
  .map(_.split(","))
  .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
  .toDF()
// 临时视图people，可以看成一张数据表
peopleDF.createOrReplaceTempView("people")
// 
val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")

第二种：Programmatically Specifying the Schema

当使用case class不能提前定义好时，比如：一个文本，可能会因为用户不同而映射有不同的fields。程序化的创建DataFrame有三步：
1、创建元素是Row对象的RDD。
2、使用StructType创建schema，保证与第1步中Row的结构相匹配。
3、通过sparkSession提供的createDataFrame方法，讲schema应用于RDD上。

import org.apache.spark.sql.types._
// Create an RDD
val peopleRDD = spark.sparkContext.textFile("examples/src/main/resources/people.txt")
// The schema is encoded in a string
val schemaString = "name age"
// Generate the schema based on the string of schema
val fields = schemaString.split(" ")
  .map(fieldName => StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)
// Convert records of the RDD (people) to Rows
val rowRDD = peopleRDD
  .map(_.split(","))
  .map(attributes => Row(attributes(0), attributes(1).trim))
// Apply the schema to the RDD
val peopleDF = spark.createDataFrame(rowRDD, schema)
// Creates a temporary view using the DataFrame
peopleDF.createOrReplaceTempView("people")
// SQL can be run over a temporary view created using DataFrames
val results = spark.sql("SELECT name FROM people")