DataFrame与RDD互操作
官网：https://spark.apache.org/docs/2.2.1/sql-programming-guide.html
1、DataFrame与RDD互操作之一：反射方式 （RDD ==> DataFrame ①建立样例类，②调用toDF方法）
package com.hadoop.spark
import org.apache.spark.sql.SparkSession
/**
* DataFrame和RDD的互操作
*/
object DataFrameRDDApp {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("DataFrameRDDApp").master("local[2]").getOrCreate()
//RDD ==> DataFrame
val rdd = spark.sparkContext.textFile("file:///Users/chandler/Documents/Projects/SparkProjects/people.txt")
//导入隐式转换
import spark.implicits._
val peopledataframe = rdd.map(_.split(",")).map(line => Info(line(0).toInt, line(1), line(2).toInt)).toDF()
peopledataframe.show()
spark.stop()
}
case class Info(id: Int, name: String, age: Int)
}
总结：
*使用反射来推断包含了特定数据类型的RDD的元数据，这个元数据就是个case class
*使用DataFrame API或者SQL方式编程
2、DataFrame与RDD互操作之二(RDD ==> DataFrame )：编程方式
第二种编程方式是比第一种反射方式要复杂的，但是这第二种他允许你构建一个DataFrame／Dataset，什么条件下呢？就是当你事先不知道这些列和他们的类型除非等到运行的时候才知道。
当我们的case class不能提前定义的时候就需要使用这一种方式了，这个方式我们必须要遵从下面三个步骤：
①创建一个RDD，我们用RowS来创建
②定义一个Schema，我们用StructType来定义
③把这个Schema作用到RDD的RowS上面通过createDataFrame这个方法来实现，当然这个方法是通过SaprkSession来提供的
def programe(spark: SparkSession): Unit = {
//RDD ==> DataFrame
val rdd = spark.sparkContext.textFile("file:///Users/chandler/Documents/Projects/SparkProjects/people.txt")
//1、创建一个RDD，我们用RowS来创建
val peopleRDD = rdd.map(_.split(",")).map(line => Row(line(0).toInt, line(1), line(2).toInt))
//2、定义一个Schema，我们用StructType来定义
val structType = StructType(Array(StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("age", IntegerType, true)))
//3、把这个Schema作用到RDD的RowS上面通过createDataFrame这个方法来实现，当然这个方法是通过SaprkSession来提供的
val peopledataframe = spark.createDataFrame(peopleRDD, structType)
peopledataframe.printSchema()
peopledataframe.show()
}
总结：
DataFrame和RDD互操作的两个方式：
1、反射：case class 前提：事先需要知道你的字段、字段类型
2、编程：Row 如果第一种情况不能满足你的要求（事先不知道列）
3、选型：优先考虑第一种
3、Dataset简述
官网说了，一个Dataset是一个分布式的数据集，而且它是一个新的接口，这个新的接口是在Spark1.6版本里面才被添加进来的，所以要注意DataFrame是先出来的，然后在1.6版本才出现的Dataset，
他有什么优点呢，提供了哪些优点呢？比如强类型，支持lambda表达式，还有还提供了sparksql执行引擎的一些优化，DataFrame里面大部分东西在Dataset里面都是能用的，
Dataset它能够通过哪些方式构建？一个是jvm对象，还有一些函数表达式比如map、flatMap、filter等等。这个Dataset可以使用在java和scala语言里面，
注意python暂时还不能支持Dataset的API，如果你使用python开发，那就老老实实使用DataFrame的API。
1）是Dataframe API的一个扩展，是Spark最新的数据抽象。
2）用户友好的API风格，既具有类型安全检查也具有Dataframe的查询优化特性。
3）Dataset支持编解码器，当需要访问非堆上的数据时可以避免反序列化整个对象，提高了效率。
4）样例类被用来在Dataset中定义数据的结构信息，样例类中每个属性的名称直接映射到DataSet中的字段名称。
5） Dataframe是Dataset的特列，DataFrame=Dataset[Row] ，所以可以通过as方法将Dataframe转换为Dataset。Row是一个类型，跟Car、Person这些的类型一样，所有的表结构信息我都用Row来表示。
6）DataSet是强类型的。比如可以有Dataset[Car]，Dataset[Person].
7）DataFrame只是知道字段，但是不知道字段的类型，所以在执行这些操作的时候是没办法在编译的时候检查是否类型失败的，
比如你可以对一个String进行减法操作，在执行的时候才报错，而DataSet不仅仅知道字段，而且知道字段类型，所以有更严格的错误检查。就跟JSON对象和类对象之间的类比。
//示例：DataFrame ==> DataSet (创建样例类，DataFrame.as[样例类名称])
package com.hadoop.spark import org.apache.spark.sql.SparkSession /** * Dataset的操作 */ object DatasetAPP { def main(args: Array[String]): Unit = { val spark = SparkSession.builder().appName("DatasetAPP").master("local[2]").getOrCreate() //csv文件路径 val path = "file:///Users/chandler/Documents/Projects/SparkProjects/sales.csv" //导入隐式转换 import spark.implicits._ //spark如何解析csv文件（外部数据源功能） val csv_dataframe = spark.read.option("header","true").option("inferSchema","true").csv(path) csv_dataframe.show //把csv_dataframe转换成Dataset val csv_dataset = csv_dataframe.as[Sales] csv_dataset.map(line => line.itemId).show spark.stop() } //把csv文件的列名拷贝进来 case class Sales(transactionId:Int,customerId:Int,itemId:Int,amountPaid:Double) }
4、拓展：RDD、DataFrame及DataSet的互相转换
2.3.3 DataSet转换为RDD
调用rdd方法即可。 1）创建一个DataSet scala> val DS = Seq(Person("Andy", 32)).toDS() DS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint] 2）将DataSet转换为RDD scala> DS.rdd res11: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[15] at rdd at <console>:28 2.4 DataFrame与DataSet的互操作 1. DataFrame转换为DataSet 1）创建一个DateFrame scala> val df = spark.read.json("examples/src/main/resources/people.json") df: org.apache.spark.sql.DataFrame = [age: bigint, name: string] 2）创建一个样例类 scala> case class Person(name: String, age: Long) defined class Person 3）将DateFrame转化为DataSet scala> df.as[Person] res14: org.apache.spark.sql.Dataset[Person] = [age: bigint, name: string] 2. DataSet转换为DataFrame 1）创建一个样例类 scala> case class Person(name: String, age: Long) defined class Person 2）创建DataSet scala> val ds = Seq(Person("Andy", 32)).toDS() ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint] 3）将DataSet转化为DataFrame scala> val df = ds.toDF df: org.apache.spark.sql.DataFrame = [name: string, age: bigint] 4）展示 scala> df.show +----+---+ |name|age| +----+---+ |Andy| 32| +----+---+
5、DEMO方法封装
package com.hadoop.spark
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}
/**
* DataFrame和RDD的互操作
*/
object DataFrameRDDApp {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("DataFrameRDDApp").master("local[2]").getOrCreate()
//通过反射的方式
//peopleReflection(spark)
//通过编程的方式
//programe(spark)
spark.stop()
}
//1、通过编程的方式
def programe(spark: SparkSession): Unit = {
//RDD ==> DataFrame
val rdd = spark.sparkContext.textFile("file:///Users/chandler/Documents/Projects/SparkProjects/people.txt")
//1、创建一个RDD，我们用RowS来创建
val peopleRDD = rdd.map(_.split(",")).map(line => Row(line(0).toInt, line(1), line(2).toInt))
//2、定义一个Schema，我们用StructType来定义
val structType = StructType(Array(StructField("id", IntegerType, true),
StructField("name", StringType, true),
StructField("age", IntegerType, true)))
//3、把这个Schema作用到RDD的RowS上面通过createDataFrame这个方法来实现，当然这个方法是通过SaprkSession来提供的
val peopledataframe = spark.createDataFrame(peopleRDD, structType)
peopledataframe.printSchema()
peopledataframe.show()
}
//2、通过反射的方式
private def peopleReflection(spark: SparkSession) = {
//RDD ==> DataFrame
val rdd = spark.sparkContext.textFile("file:///Users/chandler/Documents/Projects/SparkProjects/people.txt")
//导入隐式转换
import spark.implicits._
val peopledataframe = rdd.map(_.split(",")).map(line => Info(line(0).toInt, line(1), line(2).toInt)).toDF()
peopledataframe.show()
//取出年龄大于20的数据
peopledataframe.filter(peopledataframe.col("age") > 20).show
//用sparkSQL的方式来查询数据
peopledataframe.createOrReplaceTempView("people_dataframe")
spark.sql("show tables").show()
spark.sql("select * from people_dataframe where age > 20").show()
}
case class Info(id: Int, name: String, age: Int)
}