0
0

全国职业院校技能大赛-大数据 离线数据处理模块-指标计算

赛题来源2023年全国职业院校技能大赛赛题第1套任务B中指标计算模块

子任务三：指标计算

编写Scala代码，使用Spark计算相关指标。

注：在指标计算中，不考虑订单信息表中order_status字段的值，将所有订单视为有效订单。计算订单金额或订单总金额时只使用final_total_amount字段。需注意dwd所有的维表取最新的分区。

第一题

根据dwd层表统计每个省份、每个地区、每个月下单的数量和下单的总金额，存入MySQL数据库shtd_result的provinceeverymonth表中（表结构如下），然后在Linux的MySQL命令行中根据订单总数、订单总金额、省份表主键均为降序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下

字段

类型

中文含义

备注

provinceid

int

省份表主键

provincename

text

省份名称

regionid

int

地区表主键

regionname

text

地区名称

totalconsumption

double

订单总金额

当月订单总金额

totalorder

int

订单总数

当月订单总数

year

int

年

订单产生的年

month

int

月

订单产生的月

  1. import org.apache.spark.SparkConf
  2. import org.apache.spark.sql.{SaveMode, SparkSession}
  4. object Compute01 {
  5.   def main(args: Array[String]): Unit = {
  6.     System.setProperty("HADOOP_USER_NAME", "atguigu")
  8.     // TODO 创建spark连接
  9.     val conf = new SparkConf().setMaster("local[*]").setAppName("Compute01")
  10.     val spark: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
  12.     // 开启动态分区
  13.     spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
  14.     // 关闭打印日志
  15.     spark.sparkContext.setLogLevel("OFF")
  17.     // TODO 执行核心查询SQL
  18.     val result = spark.sql(
  19.       """
  20.         |select
  21.         |    province.id provinceid,
  22.         |    province.name provincename,
  23.         |    region.id regionid,
  24.         |    region.region_name regionname,
  25.         |    sum(final_total_amount)
  26.         |       over(partition by province.id,region.id,year(od.create_time),month(od.create_time)) totalconsumption,
  27.         |    count(od.id)
  28.         |       over(partition by province.id,region.id,year(od.create_time),month(od.create_time)) totalorder,
  29.         |    year(od.create_time) year,
  30.         |    month(od.create_time) month
  31.         |from (
  32.         |     select
  33.         |        id,
  34.         |        province_id,
  35.         |        final_total_amount,
  36.         |        trade_body,
  37.         |        create_time
  38.         |     from dwd.act_order_info
  39.         |) od
  40.         |left join (
  41.         |    select
  42.         |        id,
  43.         |        name,
  44.         |        region_id
  45.         |    from ods.dim_province
  46.         |    where etl_date = (
  47.         |        select max(etl_date)
  48.         |        from ods.dim_province
  49.         |    )
  50.         |) province on od.province_id = province.id
  51.         |left join (
  52.         |    select
  53.         |        id,
  54.         |        region_name
  55.         |    from dwd.dim_region
  56.         |    where etl_date = (
  57.         |       select max(etl_date)
  58.         |       from ods.dim_province
  59.         |    )
  60.         |) region on province.region_id = region.id
  61.         |""".stripMargin)
  63.     // 查看结果
  64.     result.show()
  66.     // TODO 结果保存到指定的表中
  67.     result.write
  68.       .format("jdbc") // 使用jdbc格式写入带mysql
  69.       .mode(SaveMode.Append) // 保存方式为追加
  70.       .option("Driver", "com.mysql.jdbc.Driver")
  71.       .option("url", "jdbc:mysql://shtd_result")
  72.       .option("user", "root")
  73.       .option("password", "000000")
  74.       .option("dbtable", "provinceeverymonth")
  75.       .save()
  76.     // shtd_result.provinceeverymonth
  77.     // TODO 关闭spark连接
  78.     spark.close()
  79.   }
  80. }

结果查询SQL

  1. -- 订单总数
  2. select
  3.     *
  4. from dwd.shtd_result.provinceeverymonth
  5. order by totalorder desc
  6. limit 5;
  7. -- 订单总金额
  8. select
  9.     *
  10. from dwd.shtd_result.provinceeverymonth
  11. order by totalconsumption desc
  12. limit 5;
  13. -- 省份表主键
  14. select
  15.     *
  16. from dwd.shtd_result.provinceeverymonth
  17. order by provinceid desc
  18. limit 5;

第二题

请根据dwd层表计算出2020年4月每个省份的平均订单金额和所有省份平均订单金额相比较结果（“高/低/相同”）,存入MySQL数据库shtd_result的provinceavgcmp表（表结构如下）中，然后在Linux的MySQL命令行中根据省份表主键、该省平均订单金额均为降序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下

字段

类型

中文含义

备注

provinceid

int

省份表主键

provincename

text

省份名称

provinceavgconsumption

double

该省平均订单金额

allprovinceavgconsumption

double

所有省平均订单金额

comparison

text

比较结果

该省平均订单金额和所有省平均订单金额比较结果，值为：高/低/相同

  1. import org.apache.spark.SparkConf
  2. import org.apache.spark.sql.{SaveMode, SparkSession}
  4. object Compute02 {
  5.   def main(args: Array[String]): Unit = {
  6.     System.setProperty("HADOOP_USER_NAME", "atguigu")
  8.     val conf = new SparkConf().setMaster("local[*]").setAppName("Compute02")
  9.     val spark: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
  11.     spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
  12.     spark.sparkContext.setLogLevel("OFF")
  14.     val result = spark.sql(
  15.       """
  16.         |select
  17.         |    provinceid,
  18.         |    provincename,
  19.         |    provinceavgconsumption,
  20.         |    allprovinceavgconsumption,
  21.         |    case
  22.         |        when provinceavgconsumption > allprovinceavgconsumption then '高'
  23.         |        when provinceavgconsumption < allprovinceavgconsumption then '低'
  24.         |        else '相同'
  25.         |    end comparison -- 比较结果
  26.         |from
  27.         |(
  28.         |    select
  29.         |        id provinceid,
  30.         |        name provincename
  31.         |    from dwd.dim_province
  32.         |    where etl_date = (
  33.         |        select max(etl_date) from ods.base_province
  34.         |    )
  35.         |) province
  36.         |left join (
  37.         |    select
  38.         |        province_id,
  39.         |        avg(final_total_amount) provinceavgconsumption -- 该省平均订单金额
  40.         |    from ods.order_info
  41.         |    where create_time between '2020-04-01' and '2020-04-30'
  42.         |    group by dwd.act_order_info
  43.         |) od on od.province_id = province.provinceid
  44.         |left join (
  45.         |    select
  46.         |        province_id,
  47.         |        avg(final_total_amount) allprovinceavgconsumption -- 所有省平均订单金额
  48.         |    from dwd.act_order_info
  49.         |    where create_time between '2020-06-01' and '2022-06-30'
  50.         |) avgorder on avgorder.province_id = province.provinceid
  51.         |""".stripMargin)
  53.     result
  54.       .write
  55.       .format("jdbc")
  56.       .mode(SaveMode.Append)
  57.       .option("Driver", "com.mysql.jdbc.Driver")
  58.       .option("url", "jdbc:mysql://shtd_result")
  59.       .option("user", "root")
  60.       .option("password", "000000")
  61.       .option("dbtable", "provinceavgcmp")
  62.       .save()
  64.     spark.close()
  65.   }
  66. }

结果查询SQL

  1. -- 省份表主键
  2. select *
  3. from shtd_result.provinceavgcmp
  4. order by provinceid desc
  5. limit 5;
  6. -- 该省平均订单金额
  7. select
  8.     *
  9. from shtd_result.provinceavgcmp
  10. order by provinceavgconsumption desc
  11. limit 5;

第三题

根据dwd层表统计在两天内连续下单并且下单金额保持增长的用户，存入MySQL数据库shtd_result的usercontinueorder表(表结构如下)中，然后在Linux的MySQL命令行中根据订单总数、订单总金额、客户主键均为降序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下

字段

类型

中文含义

备注

userid

int

客户主键

username

text

客户名称

day

text

日

记录下单日的时间，格式为

yyyyMMdd_yyyyMMdd

例如： 20220101_20220102

totalconsumption

double

订单总金额

连续两天的订单总金额

totalorder

int

订单总数

连续两天的订单总数

  1. import org.apache.spark.SparkConf
  2. import org.apache.spark.sql.{SaveMode, SparkSession}
  4. object Compute03 {
  5.   def main(args: Array[String]): Unit = {
  6.     System.setProperty("HADOOP_USER_NAME", "atguigu")
  8.     val conf = new SparkConf().setMaster("local[*]").setAppName("Compute03")
  9.     val spark: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
  11.     spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")
  12.     spark.sparkContext.setLogLevel("OFF")
  14.     val result = spark.sql(
  15.       """
  16.         | select
  17.         |     userid,
  18.         |     username,
  19.         |     buy_date_first,
  20.         |     buy_date_second,
  21.         |     concat(buy_date_first, '_', buy_date_second) day,
  22.         |     totalconsumption,
  23.         |     totalorder
  24.         | from (
  25.         |     select
  26.         |      od1.user_id userid,
  27.         |      od2.consignee username,
  28.         |      buy_date_first,
  29.         |      buy_date_second,
  30.         |      totalconsumption,
  31.         |      od1.totalorder,
  32.         |      datediff(buy_date_second, buy_date_first) part_date_num,
  33.         |      if (buy_amount_second - total_amount > 0, 1, 0) part_amount_increase
  34.         |     from (
  35.         |         select
  36.         |             user_id ,
  37.         |             create_time buy_date_first, -- 获取当前时间的下一天
  38.         |             count(id) totalorder,
  39.         |             lead(create_time, 1, "9999-12-31 00:00:00") over (partition by user_id order by create_time) buy_date_second,
  40.         |             lead(final_total_amount) over(partition by user_id order by create_time) buy_amount_second,
  41.         |             sum(total_amount) over (partition by user_id) totalconsumption
  42.         |         from dwd.act_order_info
  43.         |         group by user_id, date_format(create_time, 'yyyyMMdd')
  44.         |     ) od1
  45.         |     left join (
  46.         |     select
  47.         |         user_id,
  48.         |         consignee,
  49.         |         final_total_amount
  50.         |     from dwd.act_order_info
  51.         |     ) od2 on od1.user_id = od2.user_id
  52.         | )
  53.         | where part_date_num = 1        -- 连续两天的订单
  54.         | and part_amount_increase = 1   -- 订单金额保持增长
  55.         |""".stripMargin)
  57.     result
  58.       .write
  59.       .format("jdbc")
  60.       .mode(SaveMode.Append)
  61.       .option("Driver", "com.mysql.jdbc.Driver")
  62.       .option("url", "jdbc:mysql://shtd_result")
  63.       .option("user", "root")
  64.       .option("password", "000000")
  65.       .option("dbtable", "provinceavgcmp")
  66.       .save()
  68.     spark.close()
  69.   }
  70. }

结果查询sql

  1. -- 订单总数
  2. select
  3.     *
  4. from shtd_result.usercontinueorder
  5. order by totalorder desc
  6. limit 5;
  7. -- 订单总金额
  8. select
  9.     *
  10. from shtd_result.usercontinueorder
  11. order by totalconsumption
  12. limit 5;
  13. -- 客户主键
  14. select
  15.     *
  16. from shtd_result.usercontinueorder
  17. order by userid desc
  18. limit 5;

指标计算部分的难点就是多表查询的部分已经开窗函数的合理运用，因此熟练掌握HiveSQL中高级函数的部分是非常重要的，不然此部分将会很难完成

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

“全国职业院校技能大赛-大数据 离线数据处理模块-指标计算”的评论:

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

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

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

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

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

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

Kafka学习笔记

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

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