入口-SparkSession

• 在spark2.0版本之前
  SQLContext是创建DataFrame和执行SQL的入口
  HiveContext通过hive sql语句操作hive表数据，兼容hive操作，hiveContext继承自SQLContext。
• 在spark2.0之后
  SparkSession 封装了SqlContext及HiveContext所有功能。通过SparkSession还可以获取到SparkConetxt。
  SparkSession可以执行SparkSQL也可以执行HiveSQL.
  

创建DataFrame

1. 读取文本文件

1. 在本地创建一个文件，有id、name、age三列，用空格分隔，然后上传到hdfs上

vim /root/person.txt 1 zhangsan 20 2 lisi 29 3 wangwu 25 4 zhaoliu 30 5 tianqi 35 6 kobe 40 

上传数据文件到HDFS上：hadoop fs -put /root/person.txt /
2. 在spark shell执行下面命令，读取数据，将每一行的数据使用列分隔符分割
打开spark-shell

/export/servers/spark/bin/spark-shell  

创建RDD:

val lineRDD= sc.textFile("hdfs://node01:8020/person.txt").map(_.split(" ")) //RDD[Array[String]] 

3. 定义case class(相当于表的schema)

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

4. 将RDD和case class关联

val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt)) //RDD[Person] 

5. 将RDD转换成DataFrame

val personDF = personRDD.toDF //DataFrame 

6. 查看数据和schema

personDF.show +---+--------+---+ | id|    name|age| +---+--------+---+ | 1|zhangsan| 20| | 2|    lisi| 29| | 3|  wangwu| 25| | 4| zhaoliu| 30| | 5|  tianqi| 35| | 6|    kobe| 40| +---+--------+---+ personDF.printSchema 

7. 注册表

personDF.createOrReplaceTempView("t_person") 

8. 执行SQL

spark.sql("select id,name from t_person where id > 3").show 

9. 也可以通过SparkSession构建DataFrame

val dataFrame=spark.read.text("hdfs://node01:8020/person.txt")  dataFrame.show //注意：直接读取的文本文件没有完整schema信息 dataFrame.printSchema  

2. 读取json文件

1. 数据文件

//使用spark安装包下的json文件 PATH:more/export/servers/spark/examples/src/main/resources/people.json {"name":"Michael"} {"name":"Andy", "age":30} {"name":"Justin", "age":19} 

2. 在spark shell执行下面命令，读取数据

val jsonDF= spark.read.json("file:///export/servers/spark/examples/src/main/resources/people.json") 

3.接下来就可以使用DataFrame的函数操作

//注意:直接读取json文件有schema信息，因为json文件本身含有Schema信息，SparkSQL可以自动解析 jsonDF.show  

3. 读取parquet文件

1. 数据文件
   使用spark安装包下的parquet文件

more /export/servers/spark/examples/src/main/resources/users.parquet 

2. 在spark shell执行下面命令，读取数据

val parquetDF=spark.read.parquet("file:///export/servers/spark/examples/src/main/resources/users.parquet") 

3. 接下来就可以使用DataFrame的函数操作

//注意：直接读取parquet文件有schema信息，因为parquet文件中保存了列的信息 parquetDF.show   

创建DataSet

1. 通过spark.createDataset创建Dataset

val fileRdd = sc.textFile("hdfs://node01:8020/person.txt") //RDD[String] val ds1 = spark.createDataset(fileRdd) //DataSet[String]  ds1.show   

2. 通RDD.toDS方法生成DataSet

case class Person(name:String, age:Int) val data = List(Person("zhangsan",20),Person("lisi",30)) //List[Person] val dataRDD = sc.makeRDD(data) val ds2 = dataRDD.toDS  //Dataset[Person] ds2.show   

3. 通过DataFrame.as[泛型]转化生成DataSet

case class Person(name:String, age:Long) val jsonDF= spark.read.json("file:///export/servers/spark/examples/src/main/resources/people.json") val jsonDS = jsonDF.as[Person] //DataSet[Person] jsonDS.show 

4. DataSet也可以注册成表进行查询

jsonDS.createOrReplaceTempView("t_person") spark.sql("select * from t_person").show 

两种查询风格

准备工作

//读取文件并转换为DataFrame或DataSet val lineRDD= sc.textFile("hdfs://node01:8020/person.txt").map(_.split(" ")) case class Person(id:Int, name:String, age:Int) val personRDD = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt)) val personDF = personRDD.toDF personDF.show //val personDS = personRDD.toDS //personDS.show  

DSL风格

SparkSQL提供了一个领域特定语言(DSL)以方便操作结构化数据

1. 查看name字段的数据

personDF.select(personDF.col("name")).show personDF.select(personDF("name")).show personDF.select(col("name")).show personDF.select("name").show 

2. 查看 name 和age字段数据

personDF.select("name", "age").show 

3. 查询所有的name和age，并将age+1

personDF.select(personDF.col("name"), personDF.col("age") + 1).show personDF.select(personDF("name"), personDF("age") + 1).show personDF.select(col("name"), col("age") + 1).show personDF.select("name","age").show //personDF.select("name", "age"+1).show  personDF.select($"name",$"age",$"age"+1).show 

4. 过滤age大于等于25的，使用filter方法过滤

personDF.filter(col("age") >= 25).show personDF.filter($"age" >25).show 

5. 统计年龄大于30的人数

personDF.filter(col("age")>30).count() personDF.filter($"age" >30).count() 

6. 按年龄进行分组并统计相同年龄的人数

personDF.groupBy("age").count().show 

SQL风格

DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表，然后可以通过在程序中使用spark.sql() 来执行SQL查询，结果将作为一个DataFrame返回
如果想使用SQL风格的语法，需要将DataFrame注册成表,采用如下的方式：

personDF.createOrReplaceTempView("t_person") spark.sql("select * from t_person").show 

1. 显示表的描述信息

spark.sql("desc t_person").show 

2. 查询年龄最大的前两名

spark.sql("select * from t_person order by age desc limit 2").show   

3. 查询年龄大于30的人的信息

spark.sql("select * from t_person where age > 30 ").show 

4. 使用SQL风格完成DSL中的需求

spark.sql("select name, age + 1 from t_person").show spark.sql("select name, age from t_person where age > 25").show spark.sql("select count(age) from t_person where age > 30").show spark.sql("select age, count(age) from t_person group by age").show 

总结

1. DataFrame和DataSet都可以通过RDD来进行创建
2. 也可以通过读取普通文本创建–注意:直接读取没有完整的约束,需要通过RDD+Schema
3. 通过josn/parquet会有完整的约束
4. 不管是DataFrame还是DataSet都可以注册成表，之后就可以使用SQL进行查询了! 也可以使用DSL!