当前位置： 首页 > article >正文

SparkSQL数据模型综合实践

article 2025/1/21 14:22:04

文章目录

  • 1. 实战概述
  • 2. 实战步骤
    • 2.1 创建数据集
    • 2.2 创建数据模型对象
      • 2.2.1 创建常量
      • 2.2.2 创建加载数据方法
      • 2.2.3 创建过滤年龄方法
      • 2.2.4 创建平均薪水方法
      • 2.2.5 创建主方法
      • 2.2.6 查看完整代码
    • 2.3 运行程序，查看结果
  • 3. 实战小结

1. 实战概述

  • 在本次实战中，创建一个名为DataModel的Spark SQL数据模型对象，用于演示如何加载数据集、过滤数据以及计算统计信息。首先，在项目根目录下创建data目录，并在其中创建了包含员工信息的employees.json文件。然后，创建DataModel对象，并定义spark常量以及三个方法：loadData()、filterAge()和avgSalary()，分别用于加载数据、过滤年龄大于20岁的员工和计算不同性别的平均工资。最后，在main()方法中调用这些方法来执行数据处理任务。

2. 实战步骤

2.1 创建数据集

  • 在项目根目录创建data目录
    在这里插入图片描述
  • 在data里创建employees.json
    在这里插入图片描述
{"name": "赵天宇", "gender": "男", "age": "19", "salary": "10000"}
{"name": "钱文博", "gender": "男", "age": "29", "salary": "8000"}
{"name": "孙志强", "gender": "男", "age": "39", "salary": "9000"}
{"name": "李明宇", "gender": "男", "age": "22", "salary": "11000"}
{"name": "周雨涵", "gender": "女", "age": "19", "salary": "14000"}
{"name": "吴美琪", "gender": "女", "age": "35", "salary": "10000"}

2.2 创建数据模型对象

  • 创建net.huawei.practice包
    在这里插入图片描述

  • 在practice子包里创建DataModel对象
    在这里插入图片描述

2.2.1 创建常量

  • 在DataModel对象里创建spark常量
    在这里插入图片描述
// 获取或创建Spark会话对象                                  
val spark = SparkSession.builder() // 创建Builder对象  
  .appName("DataModel") // 设置应用程序名称                
  .master("local[*]") // 运行模式：本地运行                 
  .getOrCreate() // 获取或创建Spark会话对象

2.2.2 创建加载数据方法

  • loadData()方法的需求说明：其主要功能是加载指定路径的文件并生成 DataFrame。方法接受一个字符串类型的参数 filePath，表示文件的路径。执行流程包括使用 spark.read.json(filePath) 方法读取 JSON 文件并创建 DataFrame，然后将生成的 DataFrame 返回给调用者。此方法简化了从文件加载数据到 DataFrame 的过程，提高了数据处理的效率和便捷性。

  • 创建loadData()方法
    在这里插入图片描述

// 加载数据方法                                                
def loadData(): DataFrame = {                            
  // 加载数据得到数据帧对象                                         
  val fileDF = spark.read.json("data/employees.json")    
  // 返回数据帧对象                                             
  fileDF                                                 
}

2.2.3 创建过滤年龄方法

  • filterAge() 方法的需求说明：该方法用于过滤 DataFrame 中年龄大于20岁的数据，并将结果打印到控制台。方法接受一个 DataFrame 类型的参数 employeeDF，表示原始 JSON 文件对应的 DataFrame 实例对象。方法不返回任何值（返回类型为 Unit）。核心思路包括使用 DataFrame 的 filter 方法根据条件过滤数据，然后使用 show 方法打印过滤后的结果。此方法简化了数据过滤和展示的过程，有助于快速分析和查看特定条件下的数据。
  • 创建filterAge()方法
    在这里插入图片描述
// 过滤年龄方法                                           
def filterAge(employeeDF: DataFrame): Unit = {      
  // 过滤年龄大于20岁的员工                                   
  val filterAgeDF = employeeDF.filter("age > 20")   
  // 显示过滤后的数据                                       
  filterAgeDF.show()                                
}

2.2.4 创建平均薪水方法

  • avgSalary() 方法，其需求是计算并打印 DataFrame 中不同性别的平均收入。方法接受一个 DataFrame 类型的参数 employeeDF，表示原始 JSON 文件对应的 DataFrame 实例对象。方法不返回任何值（返回类型为 Unit）。核心思路包括使用 createOrReplaceTempView 方法将 DataFrame 注册为临时表，然后通过 SQL 查询计算不同性别的平均收入，最后使用 show 方法将结果打印到控制台。此方法简化了数据处理流程，有助于快速分析和展示特定统计信息。
  • 创建avgSalary()方法
    在这里插入图片描述
// 计算平均工资方法                                              
def avgSalary(employeeDF: DataFrame): Unit = {           
  // 根据输入参数注册临时表                                         
  employeeDF.createOrReplaceTempView("employee")         
  // 计算平均工资                                              
  val avgSalaryDF = spark.sql(                           
    s"""                                                 
       |SELECT                                           
       |  gender, avg(salary) AS avg_salary                 
       |FROM                                             
       |  employee                                       
       |GROUP BY                                         
       |  gender                                         
       |""".stripMargin                                  
  )                                                      
  // 显示平均工资                                              
  avgSalaryDF.show()                                     
}

2.2.5 创建主方法

  • main() 方法，该方法是程序的入口点，用于调用 filterAge 和 avgSalary 方法。main 方法接受一个 Array[String] 类型的参数 args，这些参数可以在程序执行时从外部传递，使得程序能够动态使用这些参数而无需修改代码。main 方法不返回任何值（返回类型为 Unit）。业务代码的核心思路是首先调用 loadData() 方法加载数据，然后将返回的 DataFrame 传递给 filterAge 和 avgSalary 方法进行处理。这种方法结构清晰，便于管理和扩展程序功能。
  • 创建main()方法
    在这里插入图片描述
// 主方法                                     
def main(args: Array[String]): Unit = {    
  // 调用过滤年龄方法                              
  filterAge(loadData())                    
  // 调用计算平均工资方法                            
  avgSalary(loadData())                    
}

2.2.6 查看完整代码

package net.huawei.practice

import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 功能：数据模型演示
 * 作者：华卫
 * 日期：2025年01月16日
 */
object DataModel {
  // 获取或创建Spark会话对象
  val spark = SparkSession.builder() // 创建Builder对象
    .appName("DataModel") // 设置应用程序名称
    .master("local[*]") // 运行模式：本地运行
    .getOrCreate() // 获取或创建Spark会话对象

  // 加载数据方法
  def loadData(): DataFrame = {
    // 加载数据得到数据帧对象
    val fileDF = spark.read.json("data/employees.json")
    // 返回数据帧对象
    fileDF
  }

  // 过滤年龄方法
  def filterAge(employeeDF: DataFrame): Unit = {
    // 过滤年龄大于20岁的员工
    val filterAgeDF = employeeDF.filter("age > 20")
    // 显示过滤后的数据
    filterAgeDF.show()
  }

  // 计算平均工资方法
  def avgSalary(employeeDF: DataFrame): Unit = {
    // 根据输入参数注册临时表
    employeeDF.createOrReplaceTempView("employee")
    // 计算平均工资
    val avgSalaryDF = spark.sql(
      s"""
         |SELECT
         |  gender, avg(salary) AS avg_salary
         |FROM
         |  employee
         |GROUP BY
         |  gender
         |""".stripMargin
    )
    // 显示平均工资
    avgSalaryDF.show()
  }

  // 主方法
  def main(args: Array[String]): Unit = {
    // 调用过滤年龄方法
    filterAge(loadData())
    // 调用计算平均工资方法
    avgSalary(loadData())
  }
}

2.3 运行程序，查看结果

  • 运行DataModel对象
    在这里插入图片描述

3. 实战小结

  • 在本次拓展练习中，我们通过创建一个 SparkSQL 数据模型综合实践项目，深入理解了 Spark 中的数据模型和数据处理流程。首先，我们在项目根目录下创建了 data 目录，并在其中创建了 employees.json 文件，用于存储员工数据。接着，我们创建了 DataModel 对象，并在其中定义了 spark 常量和三个方法：loadData()、filterAge() 和 avgSalary()，分别用于加载数据、过滤年龄大于20岁的员工和计算不同性别的平均薪水。在 main() 方法中，我们调用了这些方法来执行数据处理任务。通过这个练习，我们不仅学会了如何在 Spark 中操作 DataFrame，还学会了如何将数据处理逻辑封装成方法，提高了代码的可读性和可维护性。此外，我们还学会了如何使用 SQL 查询来分析数据，这在处理结构化数据时非常有用。总的来说，这个练习帮助我们更好地理解了 SparkSQL 的数据模型和数据处理流程，为今后的数据处理工作打下了坚实的基础。

http://www.kler.cn/a/512184.html

相关文章：

  • 一次理清楚Java中的日期和时间
  • MySQL 主从复制原理及其工作过程的配置
  • html全局遮罩，通过websocket来实现实时发布公告
  • llava 导入process报错
  • Android SystemUI——CarSystemBar视图解析（十一）
  • 代码随想录算法训练营第三十五天-动态规划-01背包（二维）
  • 电路研究9——GPRS用的AT命令手册
  • Javascript IndexedDB 数据库
  • Golang学习笔记_28——工厂方法模式（实例）
  • 【开源免费】基于SpringBoot+Vue.JS夕阳红公寓管理系统（JAVA毕业设计）
  • 告别手动编辑:如何用Python快速创建Ansible hosts文件?
  • MyBatis与Hibernate的全面对比
  • 软件测试 —— Postman（断言）
  • Windows FileZila Server共享电脑文件夹 映射21端口外网连接
  • centos设置开机自启的几种方案（frp为例）
  • leetcode——无重复字符的最长字串（java）
  • 网站HTTP改成HTTPS
  • 一种简单又强势的Js-Forward脚本编写方式
  • 计算机网络 （50）两类密码体制
  • 客户端/服务端 负载均衡
  • C/C++、网络协议、网络安全类文章汇总
  • 自动驾驶之DriveMM： All-in-One Large Multimodal Model for Autonomous Driving
  • 『 C++ 』深入理解类中的 this 指针在适配器的作用
  • 数据库高可用方案-03-主备等高可用架构
  • Node.js npm 安装过程中 EBUSY 错误的分析与解决方案
  • 如何在 Pytest 中使用命令行界面和标记运行测试