目录
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
object Demo1Sess {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.master("local")
.appName("Demo1Sess")
// 设置spark sql产生shuffle后默认的分区数 => 并行度
// 默认是 200
.config("spark.sql.shuffle.partitions",3)
.getOrCreate()
// 从SparkSession获取SparkContext
// val sc: SparkContext = spark.sparkContext
// json中每条数据都自带结构 可以直接转换成DF
val stuDF: DataFrame = spark
.read
.format("json")
.load("spark/data/students.json")
stuDF.show() //默认显示20条
// 文本类的数据 默认是没有列名的 直接读进来是 _c0 _c1 _c2 ......
// 可以通过schema手动指定列名,空格隔开字段和字段类型
val stucsDF: DataFrame = spark
.read
.format("csv")
.schema("id String,name String,age Int,gender String,clazz String")
.load("scala/data/students.txt")
stucsDF.show()
// 直接将DataFrame注册成临时视图view
stucsDF.createOrReplaceTempView("stu")
// sql的方式
val ageDF: DataFrame = spark.sql("select * from stu where age=22")
ageDF.show()
// 同rdd一样,操作算子可以触发job
// DSL 类SQL的方式 介于SQL和代码中间的API
val dslDF: DataFrame = stucsDF.where("age=23")
.select("name", "age", "clazz")
dslDF.show()
// 统计班级人数
stucsDF.groupBy("clazz")
.count()
.write
.mode(SaveMode.Overwrite)
.save("spark/data/clazz_cnt")
// 保存的时候可以指定SaveMode
// Overwrite 覆盖
// Append 追加
// 默认以parquet形式保存
}
}
import Practice.Student
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
object Demo2CreateDF {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.master("local")
.appName("Demo2CreateDF")
.config("spark.sql.shuffle.partitions", 3)
.getOrCreate()
/**
* 1、读json数据
*/
val jsonDF: DataFrame = spark.read
.format("json")
.load("spark/data/students.json")
// jsonDF.show() // 默认显示20条
// jsonDF.show(100) // 显示100条
// jsonDF.show(false) // 完全显示
// jsonDF.show()
/**
* 2、读文本文件
*/
val csvDF: DataFrame = spark.read
.format("csv")
//csv 格式读取默认是以逗号分隔
.option("sep", ",")
.schema("id String,name String,age Int,gender String,clazz String")
.load("scala/data/students.txt")
// csvDF.show()
/**
* 3、JDBC 读取MySQL的一张表转换成 Spark SQL中的DF
*/
val jdbcDF: DataFrame = spark.read
.format("jdbc")
.option("url", "jdbc:mysql://master:3306/student")
.option("dbtable", "student")
.option("user", "root")
.option("password", "123456")
.load()
// jdbcDF.show()
// 将数据以parquet格式保存
// jdbcDF
// .write
// .mode(SaveMode.Overwrite)
// .parquet("spark/data/stu_parquet")
/**
* 4、读取parquet文件
* 无法直接查看,默认会进行压缩,而且自带表结构,读取时不需要指定schema
* 默认使用snappy压缩方式进行压缩
*/
spark.read
.format("parquet")
.load("spark/data/stu_parquet")
// .show()
// 将数据以orc格式保存
// jdbcDF.write.orc("spark/data/stu_orc")
/**
* 5、读取ORC格式的文件
* 也会默认进行压缩,空间占用率最小,默认带有表结构,可以直接读取
*/
// spark
// .read
// .format("orc")
// .load("spark/data/stu_orc")
/**
* 6、从RDD构建DF
*/
val stuRDD: RDD[String] = spark.sparkContext.textFile("scala/data/students.txt")
val stuRDD2: RDD[Student] = stuRDD.map(line => {
val splits: Array[String] = line.split(",")
val id: String = splits(0)
val name: String = splits(1)
val age: String = splits(2)
val gender: String = splits(3)
val clazz: String = splits(4)
Student(id, name, age, gender, clazz)
})
// 导入隐式转换
import spark.implicits._
val sDF: DataFrame = stuRDD2.toDF()
sDF.show()
// DataFrame to RDD
val rdd: RDD[Row] = sDF.rdd
rdd.foreach(row=>{
val id: String = row.getAs[String]("id")
val name: String = row.getAs[String]("name")
println(s"$id,$name")
})
}
case class Student(id:String,name:String,age: String, gender: String, clazz: String)
}
import org.apache.spark.sql.{DataFrame, SparkSession}
object DFapi {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.master("local")
.appName("DFapi")
.config("spark.sql.shuffle.partitions", 2)
.getOrCreate()
import spark.implicits._
val stuDF: DataFrame = spark.read
.format("csv")
.option("sep", ",")
.schema("id String,name String,age String,gender String,clazz String")
.load("scala/data/students.txt")
// 对多次使用的DF也可进行cache
stuDF.cache()
// 过滤 where
// 过滤出 年龄 大于 23的学生
// DSL
// 字符串表达式
stuDF.where("age>23")
// 列表达式 (推荐),需要先导入隐式转换
stuDF.where($"age" > 23)
// 使用filter加函数的方式进行过滤
stuDF.filter(row => {
val age: String = row.getAs[String]("age")
if (age.toInt > 23) {
true
}
else {
false
}
})
// select
stuDF.select($"id", $"name", $"age" + 100 as "newage")
// 分组 groupBy
// 聚合
// 统计班级人数
stuDF.groupBy($"clazz")
.count().show()
// 导入所有的sql函数
import org.apache.spark.sql.functions._
// 统计每个班的性别人数
stuDF.groupBy($"clazz", $"gender")
.agg(count($"gender"))
.show()
// 统计班级人数(数据可能有重复)
stuDF.groupBy($"clazz")
.agg(countDistinct($"id") as "去重人数")
.show()
// SQL 的方式
stuDF.createOrReplaceTempView("stu")
spark.sql(
"""
|select clazz,count(distinct id)
|from stu
|group by clazz
""".stripMargin
).show()
// join
val scoreDF: DataFrame = spark.read
.format("csv")
.schema("sid String,sub_id String,score Int")
.load("scala/data/score.txt")
// 当两张表的关联字段名字一样时
// 在这里直接指定 "id" 默认是inner join
// .join(scoreDF, "id")
// 可以将 "id" 放入 List 传入 再指定关联类型
// .join(scoreDF, List("id"), "left")
// 如果 关联字段不一样
stuDF.join(scoreDF,$"id"===$"sid","left").show()
stuDF.unpersist()
}
}
import org.apache.spark.sql.{DataFrame, SparkSession}
object DianXin {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.master("local")
.appName("DianXin")
.config("spark.sql.shuffle.partitions", 2)
.getOrCreate()
val dxDF: DataFrame = spark.read
.format("csv")
.option("sep", ",")
.schema("mdn String,grid_id String,city_id String,county_id String,t String,start_time String,end_time String,date String")
.load("spark/data/dianxin_data")
// 导入隐式转换
import spark.implicits._
// 导入Spark SQL中所有的函数
import org.apache.spark.sql.functions._
// 按城市统计每个区县的游客人数top3
dxDF.createOrReplaceTempView("dx")
spark.sql(
"""
|select tt1.city_id,tt1.county_id,tt1.sum,tt1.rk
|from
|(select t1. city_id,t1.county_id,t1.sum,row_number() over (partition by county_id order by t1.sum desc) as rk
|from
|(select city_id,county_id,count(distinct mdn) as sum
|from
|dx
|group by city_id,county_id) t1) tt1
|where tt1.rk<3
|
""".stripMargin
).show()
}
}
手机扫一扫
移动阅读更方便
你可能感兴趣的文章