实验目标
- 通过实践掌握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 SparkSQLimport 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""".stripMarginval 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")// 关闭sparkspark.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保存。
