0
0

2023_Spark_实验十五：SparkSQL进阶操作

实验目标
  1. 通过实践掌握Spark SQL中复杂查询（包括子查询、窗口函数、联接等）的实现方式。
  2. 了解如何通过合理的数据分区和缓存策略进行性能优化。
  3. 实现一个基于Spark SQL的ETL数据处理流程，应用相关优化技巧。
实验背景

在本实验中，学员将使用Spark SQL处理一个典型的企业级大数据处理场景：从日志文件和交易数据中提取信息、清洗数据、进行复杂查询，并优化查询性能。

实验内容
  1. 环境准备- 配置并启动一个Spark集群，确保每个学员有一个可用的Spark环境。- 准备实验数据：模拟一份包含交易记录、用户信息和产品数据的日志文件，以及对应的CSV格式数据文件。
  2. 实验步骤- 使用Spark SQL进行数据加载（加载CSV、JSON等数据格式）。- 对加载的数据进行基本清洗与转换。- 编写并优化SQL查询，使用窗口函数、JOIN操作和子查询。- 对查询过程进行性能分析，并采用缓存、分区等优化策略。
实验数据如下：

users.csv

  1. user_id,name,age,gender
  2. 1,李静,62,M
  3. 2,梁静,64,M
  4. 3,梁静,46,M
  5. 4,赵伟,59,M
  6. 5,徐丽娟,32,F
  7. 6,赵伟,23,M
  8. 7,王伟,46,F
  9. 8,徐涛,63,M
  10. 9,梁强,23,M
  11. 10,吴晓,18,M
  12. 11,周波,53,M

transactions.csv

  1. transaction_id,user_id,amount,transaction_date
  2. 1,486,429.85170924871284,2024-10-09
  3. 2,102,736.1594138169264,2024-08-19
  4. 3,758,958.0420403336467,2024-05-02
  5. 4,156,137.85335989595777,2024-10-17
  6. 5,436,962.1964461356514,2023-12-28
  7. 6,10,472.1597363615911,2024-07-25
  8. 7,349,247.35900107583026,2023-11-26
  9. 8,901,349.2802498314715,2024-05-26
实验代码
  1. package SparkSQL
  3. import org.apache.spark.sql.expressions.Window
  4. import org.apache.spark.sql.functions.{col, row_number, to_date}
  5. import org.apache.spark.sql.{SparkSession, functions => F}
  7. /**
  8.  * @projectName SparkLearning2023  
  9.  * @package SparkSQL  
  10.  * @className SparkSQL.SparkSQLAdvancedExp  
  11.  * @description SparkSQL进阶案例实验分析用户消费习惯 
  12.  * @author pblh123
  13.   
  14. * @date 2024/11/14 22:24
  15.   
  16. * @version 1.0
  17.   
  18. */
  19.     
  20. object SparkSQLAdvancedExp {
  22.   def main(args: Array[String]): Unit = {
  24.     val spark = SparkSession.builder().appName("Spark SQL Advanced Operations")
  25.       .master("local[2]")
  26.       .config("spark.sql.warehouse.dir", "tmp/spark-warehouse")
  27.       .enableHiveSupport()
  28.       .getOrCreate()
  30.     // 设置日志级别为ERROR，避免冗余日志信息
  31.     spark.sparkContext.setLogLevel("ERROR")
  33.     // 加载CSV文件
  34.     val dfUsers = spark.read.option("header", "true").csv("datas/sparksqldemo/users/users.csv")
  35.     val dfTransactions = spark.read.option("header", "true").csv("datas/sparksqldemo/transactions/transactions.csv")
  37.     // 执行操作，查看缓存是否有效
  38.     dfUsers.show(3,0)
  39.     dfTransactions.show(3,false)
  40.     dfUsers.printSchema()
  41.     dfTransactions.printSchema()
  43.     // 去除重复数据
  44.     val dfUsersClean = dfUsers.dropDuplicates()
  45.     val dfTransactionsClean = dfTransactions.dropDuplicates()
  46.     // 填充空值
  47.     val dfUsersFilled = dfUsersClean.na.fill(Map("age" -> "0", "gender" -> "unknown"))
  48.     // 使用filter去掉amount为null或空字符串的行
  49.     val dfTransactionsNoNull = dfTransactionsClean
  50.       .filter(dfTransactionsClean("amount").isNotNull && dfTransactionsClean("amount") =!= "")
  52.     // 列转换：将年龄转换为整数类型
  53.     val dfUsersWithAge = dfUsersFilled.withColumn("age", dfUsersFilled("age").cast("int"))
  54.     val dfTransactioncc = dfTransactionsNoNull
  55.       .withColumn("amount", dfTransactionsNoNull("amount").cast("double"))
  56.       .withColumn("transaction_date", to_date(col("transaction_date"), "yyyy-MM-dd"))
  57.     dfUsersWithAge.printSchema()
  58.     dfTransactioncc.printSchema()
  60.     // 使用缓存缓存数据
  61.     val dfUsersCached = dfUsersWithAge.cache()
  62.     val dfTransactionsCached = dfTransactioncc.persist()
  63.     // 执行操作，查看缓存是否有效
  64.     dfUsersCached.show(3,0)
  65.     dfTransactionsCached.show(3,0)
  67. //    将dataframe注册成临时视图，供sparksql使用
  68.     dfUsersWithAge.createOrReplaceTempView("user")
  69.     dfTransactioncc.createOrReplaceTempView("trans")
  71.     // 获取每个用户的总交易金额（子查询）
  72.     val totalSpentQuery =
  73.       """
  74.         |SELECT user_id,
  75.         |(SELECT SUM(amount) FROM trans WHERE trans.user_id = user.user_id) AS total_spent
  76.         |FROM user
  77.         |order by total_spent desc""".stripMargin
  78.     val dfTotalSpent = spark.sql(totalSpentQuery)
  79.     dfTotalSpent.show(3,0)
  81.     // 定义窗口函数：按金额排序
  82.     // col("amount").desc 降序排序，col("amount") 升序排序
  83.     val windowSpec = Window.partitionBy("user_id").orderBy(col("amount").desc)
  84.     // 为每个用户根据交易金额排序
  85.     val dfWithRank = dfTransactioncc.withColumn("rank", F.row_number().over(windowSpec))
  86.       .select("user_id", "transaction_id", "amount", "rank")
  87.     dfWithRank.show(3,0)
  89.     // 使用窗口函数为每个用户的交易记录分配行号
  90.     val dfWithRank2 = dfTransactioncc.withColumn("rank", row_number().over(windowSpec))
  91.     // 筛选出每个用户前5条记录
  92.     val top5Df = dfWithRank2.filter(col("rank") <= 5)
  93.       .select("user_id", "transaction_id", "amount", "rank")
  94.     // 显示结果
  95.     top5Df.show(10,0)
  97.     // 内联接（JOIN）操作：将用户与Top5交易数据联接
  98.     val dfUserTransactions = dfUsersWithAge.join(top5Df, "user_id")
  99.     dfUserTransactions.show(3,0)
  101.     // 查看查询的执行计划
  102.     dfUserTransactions.explain(true)
  103.     dfUserTransactions.coalesce(1).write.parquet("datas/sparksqldemo/usersTrans")
  105. //    关闭spark
  106.     spark.stop()
  107.   }
  109. }
代码执行过程图
  1.     // 加载CSV文件
  2.     val dfUsers = spark.read.option("header", "true").csv("datas/sparksqldemo/users/users.csv")
  3.     val dfTransactions = spark.read.option("header", "true").csv("datas/sparksqldemo/transactions/transactions.csv")
  5.     // 执行操作，查看缓存是否有效
  6.     dfUsers.show(3,0)
  7.     dfTransactions.show(3,false)
  8.     dfUsers.printSchema()
  9.     dfTransactions.printSchema()

  1.     // 列转换：将年龄转换为整数类型
  2.     val dfUsersWithAge = dfUsersFilled.withColumn("age", dfUsersFilled("age").cast("int"))
  3.     val dfTransactioncc = dfTransactionsNoNull
  4.       .withColumn("amount", dfTransactionsNoNull("amount").cast("double"))
  5.       .withColumn("transaction_date", to_date(col("transaction_date"), "yyyy-MM-dd"))
  6.     dfUsersWithAge.printSchema()
  7.     dfTransactioncc.printSchema()

  1.     val dfTotalSpent = spark.sql(totalSpentQuery)
  2.     dfTotalSpent.show(3,0)
  4.     // 定义窗口函数：按金额排序
  5.     // col("amount").desc 降序排序，col("amount") 升序排序
  6.     val windowSpec = Window.partitionBy("user_id").orderBy(col("amount").desc)
  7.     // 为每个用户根据交易金额排序
  8.     val dfWithRank = dfTransactioncc.withColumn("rank", F.row_number().over(windowSpec))
  9.       .select("user_id", "transaction_id", "amount", "rank")
  10.     dfWithRank.show(3,0)

  1.     // 使用窗口函数为每个用户的交易记录分配行号
  2.     val dfWithRank2 = dfTransactioncc.withColumn("rank", row_number().over(windowSpec))
  3.     // 筛选出每个用户前5条记录
  4.     val top5Df = dfWithRank2.filter(col("rank") <= 5)
  5.       .select("user_id", "transaction_id", "amount", "rank")
  6.     // 显示结果
  7.     top5Df.show(10,0)
  9.     // 内联接（JOIN）操作：将用户与Top5交易数据联接
  10.     val dfUserTransactions = dfUsersWithAge.join(top5Df, "user_id")
  11.     dfUserTransactions.show(3,0)

  1.     // 查看查询的执行计划
  2.     dfUserTransactions.explain(true)

任务清单：

  1. 获取近三个月消费额大于所有用户同期消费平均值的用户的前三笔销售额清单

  2. 将每个月每个用户的的总消费额数据存储到MySQL中表userTmonth保存。

标签： spark 大数据 分布式
本文转载自: https://blog.csdn.net/pblh123/article/details/143780739
版权归原作者 pblh123 所有， 如有侵权，请联系我们删除。
登录后发布评论

“2023_Spark_实验十五：SparkSQL进阶操作”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。