Spark SQL 快速入门系列(5) | 一文教你如何使用 IDEA 创建 SparkSQL 程序(小白入门！)
【摘要】
大家好，我是不温卜火，是一名计算机学院大数据专业大二的学生，昵称来源于成语—不温不火，本意是希望自己性情温和。作为一名互联网行业的小白，博主写博客一方面是为了记录自己的学习过程，另一方面是总结自己所犯的错误希望能够帮助到很多和自己一样处于起步阶段的萌新。但由于水平有限，博客中难免会有一些错误出现，有纰漏之处恳请各位大佬不吝赐教！暂时只有csdn这一个平台，博客...
大家好，我是不温卜火，是一名计算机学院大数据专业大二的学生，昵称来源于成语—
不温不火，本意是希望自己性情温和。作为一名互联网行业的小白，博主写博客一方面是为了记录自己的学习过程，另一方面是总结自己所犯的错误希望能够帮助到很多和自己一样处于起步阶段的萌新。但由于水平有限，博客中难免会有一些错误出现，有纰漏之处恳请各位大佬不吝赐教！暂时只有csdn这一个平台，博客主页：https://buwenbuhuo.blog.csdn.net/
本片博文为大家带来的是一文教你如何使用 IDEA 创建 SparkSQL 程序。
一. 添加 SparkSQL 依赖
<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>2.1.1</version>
</dependency>
- 1
- 2
- 3
- 4
- 5
- 6
二. 具体代码
2.1 DataFrameDemo
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-03 15:30
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object DataFrameDemo {
def main(args: Array[String]): Unit = { // 创建一个新的 SparkSession 对象 val spark: SparkSession = SparkSession.builder() .master("local[*]") .appName("DataFrameDemo") .getOrCreate() // 导入用到隐式转换. 如果想要使用: $"age" 则必须导入 val df: DataFrame = spark.read.json("d:/users.json") // 打印信息 df.show // 创建临时表 df.createTempView("user") spark.sql("select * from user where age > 19").show //关闭连接 spark.stop() }
}
- 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
- 2. 运行结果
2.2 创建DF
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import org.apache.spark.sql.{DataFrame, SparkSession}
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-02 12:36
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object CreateDF {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("CreateDF").master("local[2]") .getOrCreate() // 2. 通过SparkSession创建DF val df: DataFrame = spark.read.json("d:/users.json") // 3. 对DF做操作(sql) //3.1 创建临时表 df.createOrReplaceTempView("user") // 3.2 查询临时表
// spark.sql("select * from user").show spark.sql( """ |select | name, | age | from user |""".stripMargin).show // 4. 关闭SparkSession spark.stop() }
}
/*
创建df
*/
- 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
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 2. 运行结果
2.3 创建DS
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import com.buwenbuhuo.spark.sql.day01.DF2RDD.User
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-03 11:22
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object CreateDS {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("CreateDS") .master("local[2]") .getOrCreate() import spark.implicits._
/* val list1: List[Int] = List(30,50,70,60,10,20) // 把集合转成ds val ds: Dataset[Int] = list1.toDS() // df能用的ds一定可以用 ds.show*/ val list: List[User] = List(User("zs",10),User("lisi",20),User("ww",15)) val ds: Dataset[User] = list.toDS()
/* val ds: Dataset[String] = list.toDS().map(user => user.name) ds.show()*/ // 在ds做sql查询 ds.createOrReplaceTempView("user") spark.sql("select * from user where age > 15").show spark.stop() spark.close()
}
}
/*
创建DS
*/
- 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
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 2. 运行结果
2.4 DF to RDD
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession} /**
**
*
* @author 不温卜火
* *
* @create 2020-08-02 13:05
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object DF2RDD {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("DF2RDD") .master("local[2]") .getOrCreate() import spark.implicits._ // 直接从一个scala的集合到df，做练习或测试用
/* val df: DataFrame = (1 to 10).toDF("number") // df.show // 转rdd rdd中存储的一定是Row val rdd: RDD[Row] = df.rdd // rdd.collect.foreach(println) val rdd1: RDD[Int] = rdd.map(row => row.getInt(0)) rdd1.collect.foreach(println)
*/ val df: DataFrame = spark.read.json("d:/users.json") df.printSchema() val rdd1: RDD[User] = df.rdd.map(row => { User(row.getString(1), row.getLong(0).toInt) }) rdd1.collect.foreach(println) // 4. 关闭SparkSession spark.stop() } case class User(name: String, age: Int)
}
- 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
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 2. 运行结果
2.5 RDD to DF(1)
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import com.buwenbuhuo.spark.sql.day01.DF2RDD.User
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-03 10:44
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object RDD2DF {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("RDD2DF") .master("local[2]") .getOrCreate() import spark.implicits._ val sc: SparkContext = spark.sparkContext val rdd: RDD[User] = sc.parallelize(Array(User("lisi",10),User("zs",20),User("wanger",15))) rdd.toDF("buwen","buhuo").show spark.stop() }
}
- 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
- 2. 运行结果
2.6 RDD to DF(2)
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import org.apache.spark.SparkContext
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-03 10:44
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object RDD2DF2 {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("RDD2DF2") .master("local[2]") .getOrCreate() import spark.implicits._ val sc: SparkContext = spark.sparkContext val rdd = sc.parallelize(("lisi",10)::("zs",20)::Nil).map{ case (name, age) => Row(name,age) } val schema = StructType(Array(StructField("name",StringType),StructField("age",IntegerType))) // 使用提供了一些api val df: DataFrame = spark.createDataFrame(rdd,schema) df.show spark.stop() }
}
/*
通过API的方式把rdd转成df
*/
- 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
- 39
- 40
- 41
- 42
- 43
- 44
- 2. 运行结果
2.7 DS to RDD
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import org.apache.spark.SparkContext
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-03 10:44
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object RDD2DF2 {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("RDD2DF2") .master("local[2]") .getOrCreate() import spark.implicits._ val sc: SparkContext = spark.sparkContext val rdd = sc.parallelize(("lisi",10)::("zs",20)::Nil).map{ case (name, age) => Row(name,age) } val schema = StructType(Array(StructField("name",StringType),StructField("age",IntegerType))) // 使用提供了一些api val df: DataFrame = spark.createDataFrame(rdd,schema) df.show spark.stop() }
}
/*
通过API的方式把rdd转成df
*/
- 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
- 39
- 40
- 41
- 42
- 43
- 44
- 2. 运行结果
2.8 RDD to DS
- 1. 源码
package com.buwenbuhuo.spark.sql.day01
import com.buwenbuhuo.spark.sql.day01.DF2RDD.User
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
/**
**
*
* @author 不温卜火
* *
* @create 2020-08-03 11:52
**
* MyCSDN ： https://buwenbuhuo.blog.csdn.net/
*
*/
object RDDtoDS {
def main(args: Array[String]): Unit = { // 1. 先创建SparkSession val spark: SparkSession = SparkSession.builder() .appName("RDDtoDS") .master("local[2]") .getOrCreate() import spark.implicits._ val rdd: RDD[User] = spark.sparkContext.parallelize(Array(User("lisi",10),User("zs",20),User("wanger",15))) val ds: Dataset[User] = rdd.toDS() ds.show() spark.close()
}
}
- 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
- 2. 运行结果
本次的分享就到这里了,
好书不厌读百回，熟读课思子自知。而我想要成为全场最靓的仔，就必须坚持通过学习来获取更多知识，用知识改变命运，用博客见证成长，用行动证明我在努力。
如果我的博客对你有帮助、如果你喜欢我的博客内容，请“点赞” “评论”“收藏”一键三连哦！听说点赞的人运气不会太差，每一天都会元气满满呦！如果实在要白嫖的话，那祝你开心每一天，欢迎常来我博客看看。
码字不易，大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦！
文章来源: buwenbuhuo.blog.csdn.net，作者：不温卜火，版权归原作者所有，如需转载，请联系作者。
原文链接：buwenbuhuo.blog.csdn.net/article/details/107741238
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）