在之前的博客SparkSQL系列中,已经大致为大家介绍了DataFrame
,DataSet
的概念以及它们之间与RDD
之间的互转的操作描述。本篇博客,为大家带来的是关于如何在IDEA上创建SparkSQL程序,并实现数据查询与(DataFrame,DataSet,RDD)互相转换的功能!
IDEA中创建SparkSQL程序
IDEA中程序的打包和运行方式都和SparkCore类似。首先Maven依赖中需要添加新的依赖项:
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.1.1</version>
</dependency>
准备数据源people.json
{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}
<1>数据查询
object SparkSQL01_Demo {
def main(args: Array[String]): Unit = {
// 创建配置对象
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_Demo")
// 创建 SparkSQL的环境变量
// SparkConf()
val spark = SparkSession
.builder()
.config(sparkConf)
.getOrCreate()
// 导入隐式转换
import spark.implicits._
val df = spark.read.json("in/people.json")
// 查询所有数据
df.show()
// 过滤器查询
df.filter($"age">21).show()
// 创建临时表
df.createOrReplaceTempView("persons")
// Sparksql 查询
spark.sql("SELECT * FROM persons where age > 21").show()
// 释放资源
spark.stop()
}
}
<2>类型转化
object SparkSQL02_SQL {
def main(args: Array[String]): Unit = {
// 创建配置对象
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_Demo")
// 创建 SparkSQL的环境变量
// SparkSession()
val spark:SparkSession = SparkSession
.builder()
.config(sparkConf)
.getOrCreate()
// 导入隐式转换
// 这里的spark不是包名的含义,是SparkSession对象的名字
import spark.implicits._
// 创建RDD
val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1,"zhangsan",20),(2,"lisi",30),(3,"wangwu",40)))
// 转换为DF,需给出结构
val df: DataFrame = rdd.toDF("id","name","age")
// 转换为DS,需给出类型
val ds: Dataset[User] = df.as[User]
// 转换为DF
val df1: DataFrame = ds.toDF()
// 转换为RDD
val rdd1: RDD[Row] = df1.rdd
//访问
rdd1.foreach(row=>{
// 获取数据时,可以通过索引访问数据
println(row.getInt(0))
})
// 释放资源
spark.stop()
}
// 定义一个样例类
case class User(id:Int,name:String,age:Int)
}
本次的分享就到这里,受益或感兴趣的朋友记得点个赞支持一下~