Spark数据分析技术学习笔记（二）—

1 概述

1.1 Spark SQL

Spark SQL是Spark用来操作结构化和半结构化数据的接口，Spark SQL提供了三大功能：
1）Spark SQL可以从各种结构化数据源(JSON ,Hive,CSV等)中读取数据
2）支持使用SQL和HQL语句查询数据，查询结果以DataSet或DataFrame形式返回
3）Spark SQL支持SQL和常规的Python/Java/Scala代码高度整合
Spark SQL的实现：若处理的数据集是典型结构化数据源，可在Spark程序中引入Spark SQL模块，首先读取待处理数据并将其转化为Spark SQL的核心数据抽象---DataFrame,进而调用DataFrame API来对数据进有分析处理，也可以将DataFrame注册成表，直接使用SQL语句在数据表上进行交互式查询。

1.2 DataFrame

为了支持结构化数据的处理，Spark SQL提供了新的数据结构DataFrame。 DataFrame是一个由具体列组成的数据集。它在概念上等同于关系数据库中的表或R/Python语言中的data frame。 DataFrame和RDDs最主要的区别在于一个面向的是结构化数据，一个面向的是非结构化数据。 DataFrame内部的有明确Scheme结构，即列名、列字段类型都是已知的，这带来的好处是可以减少数据读取以及更好地优化执行计划，从而保证查询效率。

1.3 DataSet

Dataset也是分布式的数据集合，在Spark 1.6版本被引入，它集成了RDD和DataFrame的优点，具备强类型的特点，同时支持Lambda函数，但只能在Scala和Java语言中使用。在Spark 2.0后，为了方便开发者，Spark将DataFrame和Dataset的API融合到一起，提供了结构化的API(Structured API)，即用户可以通过一套标准的API就能完成对两者的操作。

DataFrame、DataSet和Spark SQL的实际执行流程都是相同的：

进行DataFrame/Dataset/SQL编程；
如果是有效的代码，即代码没有编译错误，Spark会将其转换为一个逻辑计划；
Spark将此逻辑计划转换为物理计划，同时进行代码优化；
Spark然后在集群上执行这个物理计划(基于RDD操作) 。

DataFrame和RDD应该如何选择？

如果想使用函数式编程而不是DataFrame API，则使用RDDs；
如果数据是非结构化的(比如流媒体或者字符流)，则使用RDDs，
如果数据是结构化的(如RDBMS中的数据)或者半结构化的(如日志)，出于性能上的考虑，应优先使用DataFrame。

RDD DataFrame和DataSet总结：

RDD适合非结构化数据的处理，而DataFrame & DataSet更适合结构化数据和半结构化的处理；
DataFrame & DataSet可以通过统一的Structured API进行访问，而RDDs则更适合函数式编程的场景；
相比于DataFrame而言，DataSet是强类型的(Typed)，有着更为严格的静态类型检查；
DataSets、DataFrames、SQL的底层都依赖了RDDs API，并对外提供结构化的访问接口。

2 spark dataframe操作

2.1 DataFrame函数

DataFrame 的常用函数——Action 操作
collect() ,返回值是一个数组，返回dataframe集合所有的行
count() 返回一个number类型的，返回dataframe集合的行数
describe(cols: String*) 返回一个通过数学计算的类表值(count, mean, stddev, min, and max)，这个可以传多个参数，中间用逗号分隔，如果有字段为空，那么不参与运算，只这对数值类型的字段。例如df.describe("age", "height").show()
first() 返回第一行，类型是row类型
head() 返回第一行，类型是row类型
head(n:Int)返回n行，类型是row 类型
show()返回dataframe集合的值默认是20行，返回类型是unit
show(n:Int)返回n行，，返回值类型是unit
table(n:Int) 返回n行，类型是row 类型

2.2 DataFrame的基本操作

cache()同步数据的内存
columns 返回一个string类型的数组，返回值是所有列的名字
dtypes返回一个string类型的二维数组，返回值是所有列的名字以及类型
persist(newlevel:StorageLevel) 返回一个dataframe.this.type 输入存储模型类型
schema 返回structType 类型，将字段名称和类型按照结构体类型返回
toDF()返回一个新的dataframe类型的
toDF(colnames：String*)将参数中的几个字段返回一个新的dataframe类型的，
unpersist() 返回dataframe.this.type 类型，去除模式中的数据
unpersist(blocking:Boolean)返回dataframe.this.type类型 true 和unpersist是一样的作用false 是去除RDD

3 创建DataFrame

spark sql 中所有功能的入口点是SparkSession 类。它可以用于创建DataFrame、注册DataFrame为table、在table 上执行SQL、缓存table、读写文件等等。在一个SparkSession 中，应用程序可以从一个已经存在的RDD、HIVE表、或者spark数据源中创建一个DataFrame。

scala> val df = spark.createDataFrame(Seq(("ming",20),("Michael",30),("Andy",18))).toDF("name","age")
df: org.apache.spark.sql.DataFrame = [name: string, age: int]scala> df.show()
+-------+---+
|   name|age|
+-------+---+
|   ming| 20|
|Michael| 30|
|   Andy| 18|
+-------+---+scala> val df = spark.read.json("file:///opt/spark/testfile/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]scala> df.show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

参考：Spark快速大数据分析

转载于:https://www.cnblogs.com/eugene0/p/11483637.html