spark sql结构化数据文件处理-dataframe_通过结构化数据结构创建dataframe

spark sql

这个类似于hive为mapreduce提供上层接口，使数据分析师可以不必为写mapreduce的Java代码而烦恼，同样并不是所有工程师都会Scala语言，所以spark sql就是做这个事情的。
spark sql是spark用来处理结构化数据的模块，它提供一个叫dataframe的编程抽象结构数据模型（带有schema信息的RDD），用户可以通过sql，dataframe API，dataset api三种方式
现在讲dataframe

dataframe的创建

spark.read.text("people.txt")
spark.read.csv("people.csv")
spark.read.json("people.json")
spark.read.parquet("people.parquet")
1
2
3
4

如果hdfs文件系统的/spark目录下有一个word.txt文件，那么就可以创建dataframe

val personDF=spark.read.text("/spark/word.txt")
1

查看personDF属性

personDF.printSchema()
1

打印personDF

personDF.show()
1

RDD转换dataframe
调用RDD的toDF()方法

val lines=sc.textFile("/spark/student.txt").map(_.split(" "))
case class Person(id:Int,name: String, age:Int)
val personRDD=lines.map(x=>Person(x(0).toInt,x(1),x(2).toInt))
val personDF=personRDD.toDF()
scala> personDF.show
+---+-----+---+
| id| name|age|
+---+-----+---+
|  1|zhang| 20|
|  2|   li| 10|
|  3| xiao| 21|
|  4| wang| 90|
|  5|jiang| 80|

1
2
3
4
5
6
7
8
9
10
11
12
13
14

查看数据属性

personDF.printSchema
1

对dataframe的操作和Python极其相似

dataframe的常用操作

show()
printSchema()
select(), select的语法和sql相似

personDF.select(personDF.col("name")).show()
personDF.select(personDF("name").as("username"),personDF("age").show())
1
2

filter(),过滤,其实就是条件查询

personDF.filter(personDF("age")>=25).show()
1

groupBy() 对记录进行分组

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

sort(), 对特定字段进行排序操作
对年龄排序

personDF.sort(personDF("age").desc).show()
1

sql风格操作

dataframe的强大之处就是可以将它看做一个关系型数据表，然后可以在程序中直接使用spark.sql()的方式执行sql 查询
而使用sql表操作的前提是需要将dataframe注册成为一个临时表

personDF.registerTempTable("t_person")
1

这样的好处，就是可以直接使用sql语句

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

不过注意，这个操作，就是sql的操作貌似是要依赖 hive的

dataset

提供了一种特定域对象中的强类型集合，也就是在RDD的每行数据中添加了类型的约束条件，只有满足约束条件的数据类型才能正常运行。
这就是相当于Java中的泛型
dataset的创建

val personDS=spark.createDataset(sc.textFile("/spark/person.txt"))
personDS.show()
1
2