如何使用IDEA开发Spark SQL程序(一文搞懂)
更新时间：2021年08月10日 11:12:11 作者：小哪吒的BD
Spark SQL 是一个用来处理结构化数据的spark组件。它提供了一个叫做DataFrames的可编程抽象数据模型，并且可被视为一个分布式的SQL查询引擎。这篇文章主要介绍了如何使用IDEA开发Spark SQL程序(一文搞懂),需要的朋友可以参考下
前言
大家好，我是DJ丶小哪吒，我又来跟你们分享知识了。对软件开发有着浓厚的兴趣。喜欢与人分享知识。做博客的目的就是为了能与 他 人知识共享。由于水平有限。博客中难免会有一些错误。如有 纰漏 之处，欢迎大家在留言区指正。小编也会及时改正。
DJ丶小哪吒又来与各位分享知识了。今天我们不飙车，今天就静静的坐下来，我们来聊一聊关于sparkSQL。准备好茶水，听老朽与你娓娓道来。
Spark SQL是什么
Spark SQL 是一个用来处理结构化数据的spark组件。它提供了一个叫做DataFrames的可编程抽象数据模型，并且可被视为一个分布式的SQL查询引擎。
1、使用IDEA开发Spark SQL
Spark会根据文件信息尝试着去推断DataFrame/DataSet的Schema，当然我们也可以手动指定，手动指定的方式有以下几种：
- 第1种：指定列名添加Schema
- 第2种：通过StructType指定Schema
- 第3种：编写样例类，利用反射机制推断Schema
1.1、指定列名添加Schema
package cn.itcast.sql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
object CreateDFDS {
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 fileRDD: RDD[String] = sc.textFile("D:\\data\\person.txt")
val linesRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
val rowRDD: RDD[(Int, String, Int)] = linesRDD.map(line =>(line(0).toInt,line(1),line(2).toInt))
//3.将RDD转成DF
//注意:RDD中原本没有toDF方法,新版本中要给它增加一个方法,可以使用隐式转换
import spark.implicits._
val personDF: DataFrame = rowRDD.toDF("id","name","age")
personDF.show(10)
personDF.printSchema()
sc.stop()
spark.stop()
}
}
1.2、通过StructType指定Schema
package cn.itcast.sql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
object CreateDFDS2 {
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 fileRDD: RDD[String] = sc.textFile("D:\\data\\person.txt")
val linesRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
val rowRDD: RDD[Row] = linesRDD.map(line =>Row(line(0).toInt,line(1),line(2).toInt))
//3.将RDD转成DF
//注意:RDD中原本没有toDF方法,新版本中要给它增加一个方法,可以使用隐式转换
//import spark.implicits._
val schema: StructType = StructType(Seq(
StructField("id", IntegerType, true),//允许为空
StructField("name", StringType, true),
StructField("age", IntegerType, true))
)
val personDF: DataFrame = spark.createDataFrame(rowRDD,schema)
personDF.show(10)
personDF.printSchema()
sc.stop()
spark.stop()
}
}
1.3、反射推断Schema–掌握
package cn.itcast.sql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
object CreateDFDS3 {
case class Person(id:Int,name:String,age:Int)
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 fileRDD: RDD[String] = sc.textFile("D:\\data\\person.txt")
val linesRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
val rowRDD: RDD[Person] = linesRDD.map(line =>Person(line(0).toInt,line(1),line(2).toInt))
//3.将RDD转成DF
//注意:RDD中原本没有toDF方法,新版本中要给它增加一个方法,可以使用隐式转换
import spark.implicits._
//注意:上面的rowRDD的泛型是Person,里面包含了Schema信息
//所以SparkSQL可以通过反射自动获取到并添加给DF
val personDF: DataFrame = rowRDD.toDF
personDF.show(10)
personDF.printSchema()
sc.stop()
spark.stop()
}
}
1.4、花式查询
package cn.itcast.sql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
object QueryDemo {
case class Person(id:Int,name:String,age:Int)
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 fileRDD: RDD[String] = sc.textFile("D:\\data\\person.txt")
val linesRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
val rowRDD: RDD[Person] = linesRDD.map(line =>Person(line(0).toInt,line(1),line(2).toInt))
//3.将RDD转成DF
//注意:RDD中原本没有toDF方法,新版本中要给它增加一个方法,可以使用隐式转换
import spark.implicits._
//注意:上面的rowRDD的泛型是Person,里面包含了Schema信息
//所以SparkSQL可以通过反射自动获取到并添加给DF
val personDF: DataFrame = rowRDD.toDF
personDF.show(10)
personDF.printSchema()
//=======================SQL方式查询=======================
//0.注册表
personDF.createOrReplaceTempView("t_person")
//1.查询所有数据
spark.sql("select * from t_person").show()
//2.查询age+1
spark.sql("select age,age+1 from t_person").show()
//3.查询age最大的两人
spark.sql("select name,age from t_person order by age desc limit 2").show()
//4.查询各个年龄的人数
spark.sql("select age,count(*) from t_person group by age").show()
//5.查询年龄大于30的
spark.sql("select * from t_person where age > 30").show()
//=======================DSL方式查询=======================
//1.查询所有数据
personDF.select("name","age")
//2.查询age+1
personDF.select($"name",$"age" + 1)
//3.查询age最大的两人
personDF.sort($"age".desc).show(2)
//4.查询各个年龄的人数
personDF.groupBy("age").count().show()
//5.查询年龄大于30的
personDF.filter($"age" > 30).show()
sc.stop()
spark.stop()
}
}
1.5、 相互转化
RDD、DF、DS之间的相互转换有很多(6种)，但是我们实际操作就只有2类:
1)使用RDD算子操作
2)使用DSL/SQL对表操作
package cn.itcast.sql
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
object TransformDemo {
case class Person(id:Int,name:String,age:Int)
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 fileRDD: RDD[String] = sc.textFile("D:\\data\\person.txt")
val linesRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
val personRDD: RDD[Person] = linesRDD.map(line =>Person(line(0).toInt,line(1),line(2).toInt))
//3.将RDD转成DF
//注意:RDD中原本没有toDF方法,新版本中要给它增加一个方法,可以使用隐式转换
import spark.implicits._
//注意:上面的rowRDD的泛型是Person,里面包含了Schema信息
//所以SparkSQL可以通过反射自动获取到并添加给DF
//=========================相互转换======================
//1.RDD-->DF
val personDF: DataFrame = personRDD.toDF
//2.DF-->RDD
val rdd: RDD[Row] = personDF.rdd
//3.RDD-->DS
val DS: Dataset[Person] = personRDD.toDS()
//4.DS-->RDD
val rdd2: RDD[Person] = DS.rdd
//5.DF-->DS
val DS2: Dataset[Person] = personDF.as[Person]
//6.DS-->DF
val DF: DataFrame = DS2.toDF()
sc.stop()
spark.stop()
}
}
1.6、Spark SQL完成WordCount（案例）
1.6.1、SQL风格
package cn.itcast.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()
}
}
1.6.2、DQL风格
package cn.itcast.sql
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()
}
}
好了，以上内容就到这里了。你学到了吗。
到此这篇关于如何使用IDEA开发Spark SQL程序(一文搞懂)的文章就介绍到这了,更多相关IDEA开发Spark SQL内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
Java程序是跨平台的，可以运行在windows也可以运行在linux。但是平台不同，平台中的文件权限也是不同的。windows大家经常使用，并且是可视化的权限管理，这里就不多讲了。本文主要讲讲linux下面的文件的权限和安全性问题，并且探讨一下如何在java程序中考虑文件的安全性2021-06-06
这篇文章主要介绍了SpringMVC中常用参数校验类注解使用示例教程，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2022-03-03
DynamicDataSource怎样解决多数据源的事务问题
这篇文章主要介绍了DynamicDataSource怎样解决多数据源的事务问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2023-07-07
依赖冲突是指项目依赖的某一个 jar 包,有多个不同的版本,因而造成类包版本冲突依赖冲突很经常是类包之间的间接依赖引起的,本文将给大家介绍Maven依赖冲突原因以及解决方法,需要的朋友可以参考下2023-12-12
Netty分布式抽象编码器MessageToByteEncoder逻辑分析
这篇文章主要介绍了Netty分布式抽象编码器MessageToByteEncoder的抽象逻辑分析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2022-03-03
这篇文章主要介绍了Java 根据网址查询DNS/IP地址的方法，具体实现代码，大家参考下本文2017-12-12
这篇文章主要以图文讲解IDEA中根据数据库自动生成实体类,本文主要以Mysql数据库为例,应该会对大家有所帮助,如果有错误的地方,还望指正2023-03-03
这篇文章主要介绍了java生成二维码并且给二维码添加logo的实例代码，代码简单易懂，非常不错，具有一定的参考借鉴价值,需要的朋友可以参考下2019-12-12
Java使用poi做加自定义注解实现对象与Excel相互转换
这篇文章主要介绍了Java使用poi做加自定义注解实现对象与Excel相互转换，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2021-05-05
这篇文章主要介绍了使用Java语言将XML转为PDF的方法,本文将介绍通过Java代码来实现该格式转换的方法，需要的朋友可以参考下2022-03-03
最新评论