实验课教案

课程名称：

大数据综合项目实训-Spark SQL 编程实战

实验名称：

基于 Spark SQL 的 Hive 多表关联与销售数据分析

一、教学目标

通过本实验，学生能够：

1. 掌握如何在 Spark 中开启 Hive 支持并读取 Hive 表；

2. 理解并掌握如何注册临时视图以支持 SQL 查询；

3. 能够使用 Spark SQL 编写多表关联分析查询语句；

4. 能够用 DataFrame API 实现分组聚合、添加列等常见操作；

5. 培养学生的数据建模与分析思维，提升综合使用能力。

二、实验环境

• 操作系统：Windows/Linux

• 开发工具：IntelliJ IDEA（推荐）或 VS Code、Jupyter

• 大数据环境：

• Apache Spark（推荐 3.x）

• Apache Hive（已部署 Metastore）

• MySQL（Hive 元数据库）

• 已创建 Hive 表（sales_info.sales_info_raw、sales_info.customers、sales_info.products），并加载示例数据

三、实验重点与难点

教学重点：

• Spark 读取 Hive 多表数据的基本流程

• 注册临时视图并进行多表关联

• SQL 与 DataFrame API 混合使用的数据分析流程

• 聚合函数和排序语句的正确使用

教学难点：

• Spark 与 Hive 的连接配置理解（如 Hive Metastore URI）

• 临时视图与 SQL 查询的匹配关系（别名引用）

• 多表 JOIN 的逻辑关系梳理与字段选择

• DataFrame API 的链式调用与聚合表达式语法掌握

四、实验内容与步骤

第 1 步：创建 SparkSession，启用 Hive 支持

• 设置 Hive Metastore 地址

• 启用 .enableHiveSupport() 方法

• 启动 SparkSession

第 2 步：从 Hive 中读取三张表

• 使用 spark.read.table() 分别加载 销售明细表、客户表 和 商品表

第 3 步：注册为临时视图

• 使用 createOrReplaceTempView 注册为 SQL 可查询的临时表

第 4 步：使用 SQL 查询各类产品在不同地区的总销售额

• 编写 SQL：三表 JOIN，按分类和地区分组

• 使用 SUM(quantity * unit_price) 计算销售额

• 通过 COUNT(DISTINCT order_id) 统计订单数

第 5 步：使用 DataFrame API 查询每个产品的总销售额（Top10）

• 添加一列 total_amount

• 按 product_id 分组聚合

• 使用 sum() 和 orderBy() 排序并提取前10

第 6 步：关闭 Spark，释放资源

五、实验数据说明

六、核心代码（学生参考）

package org.example

// 引入 SparkSession（Spark 应用程序的入口）
import org.apache.spark.sql.SparkSession
// 引入用于列操作的函数工具包，如 col()、sum() 等
import org.apache.spark.sql.functions._

object Spark_Hive_task01 {
  def main(args: Array[String]): Unit = {

    // =============================
    // 1、创建 SparkSession，并开启 Hive 支持
    // =============================
    val spark = SparkSession.builder()
      .appName("Hive MultiTable Analysis")         // 设置程序名称
      .master("local[*]")                         // 使用本地模式，[*] 表示使用全部 CPU 核心
      .enableHiveSupport()                                // 启用 Hive 支持，可以访问 Hive 表
      .config("hive.metastore.uris", "thrift://192.168.36.100:9083") // 配置 Hive 元数据服务地址
      .getOrCreate()                                      // 创建或获取 SparkSession 实例

    println("✔ SparkSession 启动，Hive 支持已启用")

    // =============================
    // 2、从 Hive 中读取三张表
    // =============================
    // 表名带上数据库前缀 sales_info，读取成 DataFrame（即虚拟的二维表）
    val salesDF = spark.read.table("sales_info.sales_info_raw")   // 销售明细表
      .filter("order_id != 'order_id'")
    val customersDF = spark.read.table("sales_info.customers")   // 客户信息表
      .filter("customer_id != 'customer_id'")
    val productsDF = spark.read.table("sales_info.products")     // 商品信息表
      .filter("product_id != 'product_id'")

    // =============================
    // 3、注册为临时视图（供 SQL 查询用）
    // =============================
    // 将 DataFrame 注册为 SQL 临时表，便于用 SQL 查询分析
    salesDF.createOrReplaceTempView("sales")
    customersDF.createOrReplaceTempView("customers")
    productsDF.createOrReplaceTempView("products")

    // 打印各临时视图的前 3 行数据，便于查看结构与数据
    println(" sales 表（销售明细）示例数据：")
    spark.sql("SELECT * FROM sales").show(3, false)

    println(" customers 表（客户信息）示例数据：")
    spark.sql("SELECT * FROM customers").show(3, false)

    println(" products 表（商品信息）示例数据：")
    spark.sql("SELECT * FROM products").show(3, false)

    // =============================
    // 4、使用 Spark SQL 查询：统计各类商品在不同地区的总销售额
    // =============================
    println("✔ 查询各类产品的销售汇总：")
    val resultDF = spark.sql(
      """
        |SELECT
        |  p.category AS product_category,                     -- 产品分类
        |  s.region AS customer_region,                        -- 客户所在地区（来自销售表）
        |  SUM(s.quantity * s.unit_price) AS total_sales,      -- 总销售额 = 数量 × 单价
        |  COUNT(DISTINCT s.order_id) AS order_count           -- 订单数量（去重统计）
        |FROM sales s
        |JOIN customers c ON s.customer_id = c.customer_id     -- 销售表 JOIN 客户表
        |JOIN products p ON s.product_id = p.product_id        -- 再 JOIN 产品表
        |GROUP BY p.category, s.region                         -- 分组：产品分类 + 客户地区
        |ORDER BY total_sales DESC                             -- 按总销售额降序排列
        |""".stripMargin)

    // 展示结果表
    resultDF.show(truncate = false)

    // =============================
    // 5、使用 DataFrame API 查询：找出销售额最高的前10个商品
    // =============================
    println("✔ 每个产品的总销售额（单价 × 数量），前10名：")

    salesDF
      .withColumn("total_amount", col("quantity") * col("unit_price")) // 新增列：total_amount = 数量 × 单价
      .groupBy("product_id")                                      // 按产品 ID 分组
      .agg(
        sum("total_amount").as("total_sales"),      // 统计每个产品的总销售额
        sum("quantity").as("total_quantity")        // 可选：统计每个产品总销量
      )
      .orderBy(col("total_sales").desc)                      // 按总销售额降序排列
      .limit(10)                                                       // 只取前 10 名
      .show(truncate = false)

    // =============================
    // 6、关闭 Spark，释放资源
    // =============================
    spark.stop()
    println("✔ 程序执行完毕，资源释放。")
  }
}

七、实验结果展示

八、扩展方向

• 用 ECharts 展示 Top10 商品销售额柱状图

• 将结果写入 Hive 表（如 sales_summary）