SparkSQL数据模型综合实践

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.practiceimport 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 的数据模型和数据处理流程，为今后的数据处理工作打下了坚实的基础。