大数据之Spark:Spark SQL
Posted 浊酒南街
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了大数据之Spark:Spark SQL相关的知识,希望对你有一定的参考价值。
目录
6. Spark SQL 应用
在 spark2.0 版本之前
SQLContext 是创建 DataFrame 和执行 SQL 的入口。
HiveContext 通过 hive sql 语句操作 hive 表数据,兼容 hive 操作,hiveContext 继承自 SQLContext。
在 spark2.0 之后
这些都统一于 SparkSession,SparkSession 封装了 SqlContext 及 HiveContext;
实现了 SQLContext 及 HiveContext 所有功能;
通过 SparkSession 还可以获取到 SparkConetxt。
1) 创建 DataFrame/DataSet
读取文本文件
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
2、打开 spark-shell
spark/bin/spark-shell
创建 RDD
val lineRDD= sc.textFile("hdfs://node1: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://node1:8020/person.txt")
dataFrame.show //注意:直接读取的文本文件没有完整schema信息
dataFrame.printSchema
备注:
01、读取 json 文件:
val jsonDF= spark.read.json("file:///resources/people.json")
接下来就可以使用 DataFrame 的函数操作
jsonDF.show
注意:直接读取 json 文件有 schema 信息,因为 json 文件本身含有 Schema 信息,SparkSQL 可以自动解析。
02、读取 parquet 文件:
val parquetDF=spark.read.parquet("file:///resources/users.parquet")
接下来就可以使用 DataFrame 的函数操作
parquetDF.show
注意:直接读取 parquet 文件有 schema 信息,因为 parquet 文件中保存了列的信息。
2) 两种查询风格:DSL 和 SQL
1、准备工作:
先读取文件并转换为 DataFrame 或 DataSet:
val lineRDD= sc.textFile("hdfs://node1: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
2、DSL 风格:
SparkSQL 提供了一个领域特定语言(DSL)以方便操作结构化数据
01、查看 name 字段的数据
personDF.select(personDF.col("name")).show
personDF.select(personDF("name")).show
personDF.select(col("name")).show
personDF.select("name").show
02、查看 name 和 age 字段数据
personDF.select("name", "age").show
03、查询所有的 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
04、过滤 age 大于等于 25 的,使用 filter 方法过滤
personDF.filter(col("age") >= 25).show
personDF.filter($"age" >25).show
05、统计年龄大于 30 的人数
personDF.filter(col("age")>30).count()
personDF.filter($"age" >30).count()
06、按年龄进行分组并统计相同年龄的人数
personDF.groupBy("age").count().show
3、SQL 风格:
DataFrame 的一个强大之处就是我们可以将它看作是一个关系型数据表,然后可以通过在程序中使用 spark.sql() 来执行 SQL 查询,结果将作为一个 DataFrame 返回。
如果想使用 SQL 风格的语法,需要将 DataFrame 注册成表,采用如下的方式:
personDF.createOrReplaceTempView("t_person")
spark.sql("select * from t_person").show
01、显示表的描述信息
spark.sql("desc t_person").show
02、查询年龄最大的前两名
spark.sql("select * from t_person order by age desc limit 2").show
03、查询年龄大于 30 的人的信息
spark.sql("select * from t_person where age > 30 ").show
04、使用 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
4、总结:
DataFrame 和 DataSet 都可以通过 RDD 来进行创建;
也可以通过读取普通文本创建–注意:直接读取没有完整的约束,需要通过 RDD+Schema;
通过 josn/parquet 会有完整的约束;
不管是 DataFrame 还是 DataSet 都可以注册成表,之后就可以使用 SQL 进行查询了! 也可以使用 DSL!
3) Spark SQL 完成 WordCount
1、SQL 风格:
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame, Dataset, SparkSession
object WordCount
def main(args: Array[String]): Unit =
//1.创建SparkSession
val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL").getOrCreate()
val sc: SparkContext = spark.sparkContext
sc.setLogLevel("WARN")
//2.读取文件
val fileDF: DataFrame = spark.read.text("D:\\\\data\\\\words.txt")
val fileDS: Dataset[String] = spark.read.textFile("D:\\\\data\\\\words.txt")
//fileDF.show()
//fileDS.show()
//3.对每一行按照空格进行切分并压平
//fileDF.flatMap(_.split(" ")) //注意:错误,因为DF没有泛型,不知道_是String
import spark.implicits._
val wordDS: Dataset[String] = fileDS.flatMap(_.split(" "))//注意:正确,因为DS有泛型,知道_是String
//wordDS.show()
/*
+-----+
|value|
+-----+
|hello|
| me|
|hello|
| you|
...
*/
//4.对上面的数据进行WordCount
wordDS.createOrReplaceTempView("t_word")
val sql =
"""
|select value ,count(value) as count
|from t_word
|group by value
|order by count desc
""".stripMargin
spark.sql(sql).show()
sc.stop()
spark.stop()
2、DSL 风格:
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame, Dataset, SparkSession
object WordCount2
def main(args: Array[String]): Unit =
//1.创建SparkSession
val spark: SparkSession = SparkSession.builder().master("local[*]").appName("SparkSQL").getOrCreate()
val sc: SparkContext = spark.sparkContext
sc.setLogLevel("WARN")
//2.读取文件
val fileDF: DataFrame = spark.read.text("D:\\\\data\\\\words.txt")
val fileDS: Dataset[String] = spark.read.textFile("D:\\\\data\\\\words.txt")
//fileDF.show()
//fileDS.show()
//3.对每一行按照空格进行切分并压平
//fileDF.flatMap(_.split(" ")) //注意:错误,因为DF没有泛型,不知道_是String
import spark.implicits._
val wordDS: Dataset[String] = fileDS.flatMap(_.split(" "))//注意:正确,因为DS有泛型,知道_是String
//wordDS.show()
/*
+-----+
|value|
+-----+
|hello|
| me|
|hello|
| you|
...
*/
//4.对上面的数据进行WordCount
wordDS.groupBy("value").count().orderBy($"count".desc).show()
sc.stop()
spark.stop()
4) Spark SQL 多数据源交互
1、读数据:
读取 json 文件:
spark.read.json(“D:\\data\\output\\json”).show()
读取 csv 文件:
spark.read.csv(“D:\\data\\output\\csv”).toDF(“id”,“name”,“age”).show()
读取 parquet 文件:
spark.read.parquet(“D:\\data\\output\\parquet”).show()
读取 mysql 表:
val prop = new Properties()
prop.setProperty("user","root")
prop.setProperty("password","root")
spark.read.jdbc(
"jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8","person",prop).show()
2、写数据:
写入 json 文件:
personDF.write.json(“D:\\data\\output\\json”)
写入 csv 文件:
personDF.write.csv(“D:\\data\\output\\csv”)
写入 parquet 文件:
personDF.write.parquet(“D:\\data\\output\\parquet”)
写入 mysql 表:
val prop = new Properties()
prop.setProperty("user","root")
prop.setProperty("password","root")
personDF.write.mode(SaveMode.Overwrite).jdbc(
"jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8","person",prop)
以上是关于大数据之Spark:Spark SQL的主要内容,如果未能解决你的问题,请参考以下文章