1、RDD的局限性
- RDD仅表示数据集，RDD没有元数据，也就是说没有字段语义定义。
- RDD需要用户自己优化程序，对程序员要求较高。
- 从不同数据源读取数据相对困难。
- 合并多个数据源中的数据也较困难。
2 DataFrame和Dataset
（1）DataFrame
由于RDD的局限性，Spark产生了DataFrame。
DataFrame=RDD+Schema
其中Schema是就是元数据，是语义描述信息。
在Spark1.3之前，DataFrame被称为SchemaRDD。以行为单位构成的分布式数据集合，按照列赋予不同的名称。对select、filter、aggregation和sort等操作符的抽象。
- 内部数据无类型，统一为Row
- DataFrame是一种特殊类型的Dataset
- DataFrame自带优化器Catalyst，可以自动优化程序。
- DataFrame提供了一整套的Data Source API。
（2）Dataset
由于DataFrame的数据类型统一是Row，所以DataFrame也是有缺点的。
- Row运行时类型检查
比如salary是字符串类型，下面语句也只有运行时才进行类型检查。
dataframe.filter("salary>1000").show()
- Row不能直接操作domain对象
- 函数风格编程，没有面向对象风格的API
所以，Spark SQL引入了Dataset，扩展了DataFrame API，提供了编译时类型检查，面向对象风格的API。
Dataset可以和DataFrame、RDD相互转换。
DataFrame = Dataset[Row]
可见DataFrame是一种特殊的Dataset。
3 为什么需要DataFrame和Dataset？
我们知道Spark SQL提供了两种方式操作数据：
- SQL查询
- DataFrame和Dataset API
既然Spark SQL提供了SQL访问方式，那为什么还需要DataFrame和Dataset的API呢？
这是因为SQL语句虽然简单，但是SQL的表达能力却是有限的（所以Oracle数据库提供了PL/SQL）。DataFrame和Dataset可以采用更加通用的语言（Scala或Python）来表达用户的查询请求。此外，Dataset可以更快扑捉错误，因为SQL是运行时捕获异常，而Dataset是编译时检查错误。
4 基本步骤
- 创建SparkSession对象
SparkSession封装了Spark SQL执行环境信息，是所有Spark SQL程序唯一的入口。 - 创建DataFrame或Dataset
Spark SQL支持多种数据源 - 在DataFrame或Dataset之上进行转换和Action
Spark SQL提供了多钟转换和Action函数 - 返回结果
保存结果到HDFS中，或直接打印出来 。
步骤1：创建SparkSession对象
val spark=SparkSessin.builder
.master("local")
.appName("spark session example")
.getOrCreate()
注意：SparkSession中封装了spark.sparkContext和spark.sqlContext
后面所有程序或程序片段中出现的spark变量均是SparkSession对象
将RDD隐式转换为DataFrame
import spark.implicits._
步骤2：创建DataFrame或Dataset
提供了读写各种格式数据的API，包括常见的JSON，JDBC，Parquet，HDFS
步骤3：在DataFrame或Dataset之上进行各种操作
5 实例演示
（1）进入spark-shell
[root@node1 ~]# spark-shell
// :: WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark context Web UI available at http://192.168.80.131:4040
Spark context available as 'sc' (master = local[*], app id = local-).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.2.
/_/ Using Scala version 2.11. (Java HotSpot(TM) -Bit Server VM, Java 1.8.0_112)
Type in expressions to have them evaluated.
Type :help for more information. scala>
这里的Spark session对象是对Spark context对象的进一步封装。也就是说Spark session对象（spark）中的SparkContext就是Spark context对象（sc）,从下面输出信息可以验证。
scala> spark.sparkContext
res0: org.apache.spark.SparkContext = org.apache.spark.SparkContext@7bd7c4cf scala> println(sc)
org.apache.spark.SparkContext@7bd7c4cf scala>
（2）导入org.apache.spark.sql.Row
scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row
（3）定义case class
scala> case class User(userID:Long,gender:String,age:Int,occupation:String,zipcode:String)
defined class User scala> val usersRDD=sc.textFile("file:///root/data/ml-1m/users.dat")
usersRDD: org.apache.spark.rdd.RDD[String] = file:///root/data/ml-1m/users.dat MapPartitionsRDD[3] at textFile at <console>:25 scala> usersRDD.count
（4）case class作为RDD的schema
scala> val userRDD =usersRDD.map(_.split("::")).map(p=>User(p().toLong,p().trim,p().toInt,p(),p()))
userRDD: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[] at map at <console>:
（5）通过RDD.toDF将RDD转换为DataFrame
scala> val userDF=userRDD.toDF
userDF: org.apache.spark.sql.DataFrame = [userID: bigint, gender: string ... more fields]
（6）查看DataFrame所以方法
输入userDF.，然后tab键，可以看到DataFrame所以方法
scala> userDF.
agg cube hint randomSplitAsList take
alias describe inputFiles rdd takeAsList
apply distinct intersect reduce toDF
as drop isLocal registerTempTable toJSON
cache dropDuplicates isStreaming repartition toJavaRDD
checkpoint dtypes javaRDD rollup toLocalIterator
coalesce except join sample toString
col explain joinWith schema transform
collect explode limit select union
collectAsList filter map selectExpr unionAll
columns first mapPartitions show unpersist
count flatMap na sort where
createGlobalTempView foreach orderBy sortWithinPartitions withColumn
createOrReplaceGlobalTempView foreachPartition persist sparkSession withColumnRenamed
createOrReplaceTempView groupBy printSchema sqlContext withWatermark
createTempView groupByKey queryExecution stat write
crossJoin head randomSplit storageLevel writeStream scala>
（7）输出DataFrame的Schema
scala> userDF.printSchema
root
|-- userID: long (nullable = false)
|-- gender: string (nullable = true)
|-- age: integer (nullable = false)
|-- occupation: string (nullable = true)
|-- zipcode: string (nullable = true)
（8）DataFrame的其他方法
scala> userDF.first
res5: org.apache.spark.sql.Row = [,F,,,] scala> userDF.take()
res6: Array[org.apache.spark.sql.Row] = Array([,F,,,], [,M,,,], [,M,,,], [,M,,,], [,M,,,], [,F,,,], [,M,,,], [,M,,,], [,M,,,], [,F,,,]) scala>
（9）查看DataFrame可以转化的数据格式
输入userDF.write.，然后tab键，可以看到DataFrame可以转化的数据格式
scala> userDF.write.
bucketBy format jdbc mode options parquet save sortBy
csv insertInto json option orc partitionBy saveAsTable text scala>
（10）将DataFrame数据以JSON格式写入HDFS
scala> userDF.write.json("/tmp/json")
scala>
（11）查看HDFS
[root@node1 ~]# hdfs dfs -ls /tmp/json
Found items
-rw-r--r-- root supergroup -- : /tmp/json/_SUCCESS
-rw-r--r-- root supergroup -- : /tmp/json/part--6f19a241-2f72-4a06-a6bc-81706c89bf5b-c000.json
[root@node1 ~]#
（12）也可以写入本地
scala> userDF.write.json("file:///tmp/json")
[root@node1 ~]# ls /tmp/json
part--66aa0658---a809-468e4fde23a5-c000.json _SUCCESS
[root@node1 ~]# tail - /tmp/json/part--66aa0658---a809-468e4fde23a5-c000.json
{"userID":,"gender":"F","age":,"occupation":"","zipcode":""}
{"userID":,"gender":"F","age":,"occupation":"","zipcode":""}
{"userID":,"gender":"F","age":,"occupation":"","zipcode":""}
{"userID":,"gender":"F","age":,"occupation":"","zipcode":""}
{"userID":,"gender":"M","age":,"occupation":"","zipcode":""}
[root@node1 ~]#
（13）查看Spark SQL可以读的数据格式
scala> val df=spark.read.
csv format jdbc json load option options orc parquet schema table text textFile scala>
（14）将JSON文件转化为DataFrame
scala> val df=spark.read.json("/tmp/json")
df: org.apache.spark.sql.DataFrame = [age: bigint, gender: string ... more fields]
scala> df.take()
res9: Array[org.apache.spark.sql.Row] = Array([,F,,,], [,M,,,])
scala>
（15）再将DataFrame转化为ORC格式数据（该格式文件是二进制文件）
scala> df.write.orc("file:///tmp/orc")
[root@node1 ~]# ls /tmp/orc
part--09cf3025-cc71-4a76-a35d-a7cef4885be8-c000.snappy.orc _SUCCESS
[root@node1 ~]#
（16）读取目录/tmp/orc下的所有orc文件
scala> val orcDF=spark.read.orc("file:///tmp/orc")
orcDF: org.apache.spark.sql.DataFrame = [age: bigint, gender: string ... more fields]
scala> orcDF.first
res11: org.apache.spark.sql.Row = [,F,,,]
scala>
6 select和filter
（1）select
scala> userDF.select("UserID","age").show
+------+---+
|UserID|age|
+------+---+
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
| | |
+------+---+
only showing top rows
scala> userDF.select("UserID","age").show()
+------+---+
|UserID|age|
+------+---+
| | |
| | |
+------+---+
only showing top rows
scala> userDF.selectExpr("UserID","ceil(age/10) as newAge").show()
+------+------+
|UserID|newAge|
+------+------+
| | |
| | |
+------+------+
only showing top rows
scala> userDF.select(max('age),min('age),avg('age)).show(2)
+--------+--------+------------------+
|max(age)|min(age)| avg(age)|
+--------+--------+------------------+
| | |30.639238410596025|
+--------+--------+------------------+
**（）filter**
scala> userDF.filter(userDF("age")>).show()
+------+------+---+----------+-------+
|userID|gender|age|occupation|zipcode|
+------+------+---+----------+-------+
| | M| | | |
| | M| | | |
+------+------+---+----------+-------+
only showing top rows
scala> userDF.filter("age>30 and occupation=10").show()
+------+------+---+----------+-------+
|userID|gender|age|occupation|zipcode|
+------+------+---+----------+-------+
| | M| | | |
| | M| | | |
+------+------+---+----------+-------+
scala>
（3）select和filter组合
scala> userDF.select("userID","age").filter("age>30").show()
+------+---+
|userID|age|
+------+---+
| | |
| | |
+------+---+
only showing top rows
scala> userDF.filter("age>30").select("userID","age").show()
+------+---+
|userID|age|
+------+---+
| | |
| | |
+------+---+
only showing top rows
7 groupBy
scala> userDF.groupBy("age").count.show
+---+-----+
|age|count|
+---+-----+
| | |
| | |
| | |
| | |
| | |
| | |
| | |
+---+-----+
scala> userDF.groupBy("age").agg(count('gender),countDistinct('occupation)).show
+---+-------------+--------------------------+
|age|count(gender)|count(DISTINCT occupation)|
+---+-------------+--------------------------+
| | | |
| | | |
| | | |
| | | |
| | | |
| | | |
| | | |
+---+-------------+--------------------------+
scala> userDF.groupBy("age").agg("gender"->"count","occupation"->"count").show
+---+-------------+-----------------+
|age|count(gender)|count(occupation)|
+---+-------------+-----------------+
| | | |
| | | |
| | | |
| | | |
| | | |
| | | |
| | | |
+---+-------------+-----------------+
scala>
8 join
问题：求解看过movieID=2116电影的观众的性别与年龄的分布。
（1）Users DataFrame
scala> userDF.printSchema
root
|-- userID: long (nullable = false)
|-- gender: string (nullable = true)
|-- age: integer (nullable = false)
|-- occupation: string (nullable = true)
|-- zipcode: string (nullable = true) scala>
（2）Ratings DataFrame
scala> case class Rating(userID:Long,movieID:Long,Rating:Int,Timestamp:String)
defined class Rating scala> val ratingsRDD=sc.textFile("file:///root/data/ml-1m/ratings.dat")
ratingsRDD: org.apache.spark.rdd.RDD[String] = file:///root/data/ml-1m/ratings.dat MapPartitionsRDD[65] at textFile at <console>:25 scala> val ratingRDD =ratingsRDD.map(_.split("::")).map(p=>Rating(p().toLong,p().toLong,p().toInt,p()))
ratingRDD: org.apache.spark.rdd.RDD[Rating] = MapPartitionsRDD[] at map at <console>: scala> val ratingDF=ratingRDD.toDF
ratingDF: org.apache.spark.sql.DataFrame = [userID: bigint, movieID: bigint ... more fields] scala> scala> ratingDF.printSchema
root
|-- userID: long (nullable = false)
|-- movieID: long (nullable = false)
|-- Rating: integer (nullable = false)
|-- Timestamp: string (nullable = true) scala>
（2）join
scala> val mergeredDF=ratingDF.filter("movieID=2116").join(userDF,"userID").select("gender","age").groupBy("gender","age").count
mergeredDF: org.apache.spark.sql.DataFrame = [gender: string, age: int ... more field]
scala> mergeredDF.show
+------+---+-----+
|gender|age|count|
+------+---+-----+
| M| | |
| F| | |
| M| | |
| M| | |
| F| | |
| M| | |
| F| | |
| M| | |
| F| | |
| F| | |
| M| | |
| F| | |
| F| | |
| M| | |
+------+---+-----+
scala>
9 临时表
scala> userDF.createOrReplaceTempView("users")
scala> val groupedUsers=spark.sql("select gender,age,count(*) as num from users group by gender, age")
groupedUsers: org.apache.spark.sql.DataFrame = [gender: string, age: int ... more field]
scala> groupedUsers.show
+------+---+----+
|gender|age| num|
+------+---+----+
| M| | |
| F| | |
| M| | |
| M| | |
| F| | |
| M| ||
| F| | |
| M| | |
| F| | |
| F| | |
| M| | |
| F| | |
| F| | |
| M| | |
+------+---+----+
scala>
注意：在Spark程序运行中，临时表才存在。当Spark程序运行结束，临时表也被销毁。
10 Spark SQL的表
（1）Session范围内的临时表
- df.createOrReplaceTempView(“tableName”)
- 只在Session范围内有效，Session结束临时表自动销毁
（2）全局范围内的临时表
- df.createGlobalTempView(“tableName”)
- 所有Session共享
scala> userDF.createGlobalTempView("users")
scala> spark.sql("select * from global_temp.users").show
+------+------+---+----------+-------+
|userID|gender|age|occupation|zipcode|
+------+------+---+----------+-------+
| | F| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | F| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | F| | | |
| | F| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | F| | | |
| | M| | | |
| | F| | | |
| | M| | | |
| | M| | | |
+------+------+---+----------+-------+
only showing top rows
scala> spark.newSession().sql("select * from global_temp.users").show
+------+------+---+----------+-------+
|userID|gender|age|occupation|zipcode|
+------+------+---+----------+-------+
| | F| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | F| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | F| | | |
| | F| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | M| | | |
| | F| | | |
| | M| | | |
| | F| | | |
| | M| | | |
| | M| | | |
+------+------+---+----------+-------+
only showing top rows
scala>
（3）将DataFrame或Dataset持久化到Hive中
df.write.mode(“overwrite”).saveAsTable(“database.tableName”)