spark笔记
RDD、dataframe、dataset 区别和联系
1
2
3
4
5
6
7
8
9
10
11
| + spark 数据抽象是rdd
+ spark sql 数据抽象是dataframe、dataset
底层都是rdd 弹性分布式数据集，只是在rdd的基础上做了封装改造
+ dataframe类似于二维表，有schema元信息(每一列都有名称和类型)
+ rdd：spark并不知道对象的详细信息
+ df：有schema，所以spark知道对象的内部结构
+ dataset提供了强类型支持，在rdd的每行都加了类型约束
+ DataFrame表示为DataSet[Row]，即DataSet的子集。
|
rdd、dataframe、dataset互相转换
rdd转dataframe、dataset
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| def main(args: Array[String]): Unit = {
val spark = GetSpark.getSpark("local[2]", "SparkSqlSchema")
import spark.implicits._
val sc = GetSpark.getSC("local[2]", "SparkSqlSchema")
val arr = Array(Person(1, "zs"), Person(2, "xw"))
val rdd = sc.makeRDD(arr, 2)
val df = rdd.toDF
val ds = rdd.toDS()
}
case class Person(id:Int, name:String)
|
df转rdd、ds
- val rdd = df.rdddf.rdd
- val ds = df.as[Person]
1
2
3
4
5
|
val dfToRdd = df.rdd
val dfTods = df.as[Person]
|
ds转rdd、df
- val rdd = ds.rdd
- val df = ds.toDF
1
2
3
4
5
|
val rdd1 = ds.rdd
val df1 = ds.toDF
|
dataframe的创建
读取文本文件创建DataFrame
本地文件内容 person.txt
1
2
3
4
5
6
| 1 zhangsan 20
2 lisi 29
3 wangwu 25
4 zhaoliu 30
5 tianqi 35
6 kobe 40
|
- 直接用spark读生成的是DataFrame，但是只有一个字段
1
2
3
|
val df = spark.read.text("h:/person.txt")
df.show
|
1
2
3
4
5
6
7
8
9
10
11
12
| case class Person(id:Int, name:String, age:Int)
val lineRDD = sc.textFile("h:/person.txt").map(_.split(" "))
val rdd = lineRDD.map(x => Person(x(0).toInt, x(1), x(2).toInt))
val df = rdd.toDF()
df.show()
df.printSchema()
|
读取json文件创建DataFrame
1
2
3
| {"id":1,"name":"zs","age":18}
{"id":2,"name":"ls","age":18}
{"id":3,"name":"ww","age":18}
|
1
2
3
| val df = spark.read.json("H:\\tmp\\person.json")
df.show()
df.printSchema()
|
读取mysql数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
def main(args: Array[String]): Unit = {
val spark = GetSpark.getSpark("local[2]", "sqlSpark")
import spark.implicits._
val sc = GetSpark.getSC("local[2]", "sqlSpark")
val properties = new Properties()
properties.setProperty("user", "root")
properties.setProperty("password", "root")
val df = spark.read.jdbc("jdbc:mysql://localhost:3306/test?serverTimezone=UTC", "cinema", properties)
df.show()
spark.stop()
}
|
写入到mysql
- rsdDF.write.jdbc
- rsdDF.write.mode
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
| package com.demo
import java.util.Properties
import com.utils.GetSpark
import org.apache.spark.sql.SaveMode
object ConMysql {
def main(args: Array[String]): Unit = {
val spark = GetSpark.getSpark("local[2]", "sqlSpark")
import spark.implicits._
val sc = GetSpark.getSC("local[2]", "sqlSpark")
val properties = new Properties()
properties.setProperty("user", "root")
properties.setProperty("password", "My12345!")
val df = spark.read.json("H:\\tmp\\person.json")
df.createOrReplaceTempView("person")
val rsdDF = spark.sql("select * from person")
rsdDF.show()
rsdDF.write.jdbc("jdbc:mysql://192.168.72.130:3306/hive", "person", properties)
spark.stop()
}
case class Person(id:Int, name:String, age:Int)
}
|
DSL风格和SQL风格
DSL风格语法
DataFrame提供了一个领域特定语言(DSL)来操作结构化数据。
- 查看DataFrame中的内容，通过调用show方法
1
2
3
| 查看name字段的数据
personDF.select(personDF.col("name")).show
personDF.select("name").show
|
1
| personDF.select(col("name"), col("age")).show
|
1
2
| personDF.select(col("id"), col("name"), col("age") + 1).show
personDF.select(personDF("id"), personDF("name"), personDF("age") + 1).show
|
- 过滤age大于等于25的，使用filter方法过滤
1
| personDF.filter(col("age") >= 25).show
|
1
| personDF.groupBy("age").count().show
|
SQL风格语法
DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表，然后可以通过在程序中使用spark.sql() 来执行SQL语句查询，结果返回一个DataFrame。
如果想使用SQL风格的语法，需要将DataFrame注册成表,采用如下的方式：
- personDF.registerTempTable(“t_person”)
- personDF.createOrReplaceTempView(“t_person”)
1
| spark.sql("select * from t_person order by age desc limit 2").show
|
1
| spark.sql("desc t_person").show
|
1
| spark.sql("select * from t_person where age > 30 ").show
|
参考链接