实验目标
- 通过实践掌握Spark SQL中复杂查询（包括子查询、窗口函数、联接等）的实现方式。
- 了解如何通过合理的数据分区和缓存策略进行性能优化。
- 实现一个基于Spark SQL的ETL数据处理流程，应用相关优化技巧。
实验背景
在本实验中，学员将使用Spark SQL处理一个典型的企业级大数据处理场景：从日志文件和交易数据中提取信息、清洗数据、进行复杂查询，并优化查询性能。
实验内容
-
环境准备
- 配置并启动一个Spark集群，确保每个学员有一个可用的Spark环境。
- 准备实验数据：模拟一份包含交易记录、用户信息和产品数据的日志文件，以及对应的CSV格式数据文件。
-
实验步骤
- 使用Spark SQL进行数据加载（加载CSV、JSON等数据格式）。
- 对加载的数据进行基本清洗与转换。
- 编写并优化SQL查询，使用窗口函数、JOIN操作和子查询。
- 对查询过程进行性能分析，并采用缓存、分区等优化策略。
实验数据如下：
users.csv
user_id,name,age,gender
1,李静,62,M
2,梁静,64,M
3,梁静,46,M
4,赵伟,59,M
5,徐丽娟,32,F
6,赵伟,23,M
7,王伟,46,F
8,徐涛,63,M
9,梁强,23,M
10,吴晓,18,M
11,周波,53,Mtransactions.csv
transaction_id,user_id,amount,transaction_date
1,486,429.85170924871284,2024-10-09
2,102,736.1594138169264,2024-08-19
3,758,958.0420403336467,2024-05-02
4,156,137.85335989595777,2024-10-17
5,436,962.1964461356514,2023-12-28
6,10,472.1597363615911,2024-07-25
7,349,247.35900107583026,2023-11-26
8,901,349.2802498314715,2024-05-26实验代码
package SparkSQL
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, row_number, to_date}
import org.apache.spark.sql.{SparkSession, functions => F}
/**
* @projectName SparkLearning2023
* @package SparkSQL
* @className SparkSQL.SparkSQLAdvancedExp
* @description SparkSQL进阶案例实验分析用户消费习惯
* @author pblh123
* @date 2024/11/14 22:24
* @version 1.0
*/
object SparkSQLAdvancedExp {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("Spark SQL Advanced Operations")
.master("local[2]")
.config("spark.sql.warehouse.dir", "tmp/spark-warehouse")
.enableHiveSupport()
.getOrCreate()
// 设置日志级别为ERROR，避免冗余日志信息
spark.sparkContext.setLogLevel("ERROR")
// 加载CSV文件
val dfUsers = spark.read.option("header", "true").csv("datas/sparksqldemo/users/users.csv")
val dfTransactions = spark.read.option("header", "true").csv("datas/sparksqldemo/transactions/transactions.csv")
// 执行操作，查看缓存是否有效
dfUsers.show(3,0)
dfTransactions.show(3,false)
dfUsers.printSchema()
dfTransactions.printSchema()
// 去除重复数据
val dfUsersClean = dfUsers.dropDuplicates()
val dfTransactionsClean = dfTransactions.dropDuplicates()
// 填充空值
val dfUsersFilled = dfUsersClean.na.fill(Map("age" -> "0", "gender" -> "unknown"))
// 使用filter去掉amount为null或空字符串的行
val dfTransactionsNoNull = dfTransactionsClean
.filter(dfTransactionsClean("amount").isNotNull && dfTransactionsClean("amount") =!= "")
// 列转换：将年龄转换为整数类型
val dfUsersWithAge = dfUsersFilled.withColumn("age", dfUsersFilled("age").cast("int"))
val dfTransactioncc = dfTransactionsNoNull
.withColumn("amount", dfTransactionsNoNull("amount").cast("double"))
.withColumn("transaction_date", to_date(col("transaction_date"), "yyyy-MM-dd"))
dfUsersWithAge.printSchema()
dfTransactioncc.printSchema()
// 使用缓存缓存数据
val dfUsersCached = dfUsersWithAge.cache()
val dfTransactionsCached = dfTransactioncc.persist()
// 执行操作，查看缓存是否有效
dfUsersCached.show(3,0)
dfTransactionsCached.show(3,0)
// 将dataframe注册成临时视图，供sparksql使用
dfUsersWithAge.createOrReplaceTempView("user")
dfTransactioncc.createOrReplaceTempView("trans")
// 获取每个用户的总交易金额（子查询）
val totalSpentQuery =
"""
|SELECT user_id,
|(SELECT SUM(amount) FROM trans WHERE trans.user_id = user.user_id) AS total_spent
|FROM user
|order by total_spent desc""".stripMargin
val dfTotalSpent = spark.sql(totalSpentQuery)
dfTotalSpent.show(3,0)
// 定义窗口函数：按金额排序
// col("amount").desc 降序排序，col("amount") 升序排序
val windowSpec = Window.partitionBy("user_id").orderBy(col("amount").desc)
// 为每个用户根据交易金额排序
val dfWithRank = dfTransactioncc.withColumn("rank", F.row_number().over(windowSpec))
.select("user_id", "transaction_id", "amount", "rank")
dfWithRank.show(3,0)
// 使用窗口函数为每个用户的交易记录分配行号
val dfWithRank2 = dfTransactioncc.withColumn("rank", row_number().over(windowSpec))
// 筛选出每个用户前5条记录
val top5Df = dfWithRank2.filter(col("rank") <= 5)
.select("user_id", "transaction_id", "amount", "rank")
// 显示结果
top5Df.show(10,0)
// 内联接（JOIN）操作：将用户与Top5交易数据联接
val dfUserTransactions = dfUsersWithAge.join(top5Df, "user_id")
dfUserTransactions.show(3,0)
// 查看查询的执行计划
dfUserTransactions.explain(true)
dfUserTransactions.coalesce(1).write.parquet("datas/sparksqldemo/usersTrans")
// 关闭spark
spark.stop()
}
}
代码执行过程图
// 加载CSV文件
val dfUsers = spark.read.option("header", "true").csv("datas/sparksqldemo/users/users.csv")
val dfTransactions = spark.read.option("header", "true").csv("datas/sparksqldemo/transactions/transactions.csv")
// 执行操作，查看缓存是否有效
dfUsers.show(3,0)
dfTransactions.show(3,false)
dfUsers.printSchema()
dfTransactions.printSchema()
// 列转换：将年龄转换为整数类型
val dfUsersWithAge = dfUsersFilled.withColumn("age", dfUsersFilled("age").cast("int"))
val dfTransactioncc = dfTransactionsNoNull
.withColumn("amount", dfTransactionsNoNull("amount").cast("double"))
.withColumn("transaction_date", to_date(col("transaction_date"), "yyyy-MM-dd"))
dfUsersWithAge.printSchema()
dfTransactioncc.printSchema()
val dfTotalSpent = spark.sql(totalSpentQuery)
dfTotalSpent.show(3,0)
// 定义窗口函数：按金额排序
// col("amount").desc 降序排序，col("amount") 升序排序
val windowSpec = Window.partitionBy("user_id").orderBy(col("amount").desc)
// 为每个用户根据交易金额排序
val dfWithRank = dfTransactioncc.withColumn("rank", F.row_number().over(windowSpec))
.select("user_id", "transaction_id", "amount", "rank")
dfWithRank.show(3,0)
// 使用窗口函数为每个用户的交易记录分配行号
val dfWithRank2 = dfTransactioncc.withColumn("rank", row_number().over(windowSpec))
// 筛选出每个用户前5条记录
val top5Df = dfWithRank2.filter(col("rank") <= 5)
.select("user_id", "transaction_id", "amount", "rank")
// 显示结果
top5Df.show(10,0)
// 内联接（JOIN）操作：将用户与Top5交易数据联接
val dfUserTransactions = dfUsersWithAge.join(top5Df, "user_id")
dfUserTransactions.show(3,0)
// 查看查询的执行计划
dfUserTransactions.explain(true)
任务清单：
1. 获取近三个月消费额大于所有用户同期消费平均值的用户的前三笔销售额清单
2. 将每个月每个用户的的总消费额数据存储到MySQL中表userTmonth保存。
