Hadoop+Spark大数据技术 第七次作业

• 第七次作业

  • 1. 简述Spark SQL使用的数据抽象DataFrame与Dataset的区别。

    • DataFrame: 基于 Row 对象的二维表格结构，类似于关系型数据库中的表。 行和列都有明确的 Schema（模式），可以进行类型推断。 提供了丰富的操作接口，如 select、filter、group by、agg 等。 缺点： 需要在操作时进行类型转换，例如使用 col("age").cast("int") 将类型转换为 Int。

    • Dataset: 基于特定类型的数据结构，例如 Dataset[Student]，其中 Student 是一个 case class。 相比 DataFrame，Dataset 能更方便地进行类型推断，无需显式转换。 优点： 代码更简洁，类型安全，编译器可以进行类型检查。 总结： 当需要进行类型安全的操作时，Dataset 是更好的选择。当数据结构复杂，需要进行类型转换时，DataFrame 更灵活。

  • 2. 简述创建DataFrame对象的常用方法

    • 1. Parquet文件创建：使用`SparkSession`的`read.parquet()`方法从Parquet文件中创建DataFrame。

    • 2. json文件创建DataFrame对象：通过`SparkSession`的`read.json()`方法，可以从JSON文件中读取数据并创建DataFrame。

    • 3. RDD创建DataFrame对象：如果有一个RDD，使用`SparkSession`的`createDataFrame()`方法将其转换为DataFrame。

    • 4. SparkSession创建：`SparkSession`是Spark 2.0引入的入口点，可以直接用来创建DataFrame。如

    • 5. Seq创建DataFrame对象： `SparkSession`的`createDataFrame()`方法也可以接受一个包含元组的Scala `Seq`。

  • 3. 简述DataFrame对象的常用操作

    • 见实验8

  • 4. 阅读、分析下列各程序段中各语句的功能，并给出运行结果。

（1） 设grade.json文件的内容如下，给出分析
{"ID":"106","Name":"Ding","Class":"1","Scala":92,"Spark":91}
{"ID":"242","Name":"Yan","Class":"2","Scala":96,"Spark":90}
{"ID":"107","Name":"Feng","Class":"1","Scala":84,"Spark":91}
{"ID":"230","Name":"Wang","Class":"2","Scala":87,"Spark":91}
{"ID":"153","Name":"Zhang","Class":"1","Scala":62,"Spark":71}
{"ID":"242","Name":"Xu","Class":"2","Scala":88,"Spark":90}
{"ID":"235","Name":"Wu","Class":"1","Scala":83,"Spark":91}
{"ID":"224","Name":"Xia","Class":"2","Scala":89,"Spark":91}
 
val gradeDF=spark.read.json("grade.json")
gradeDF.count()
gradeDF.groupBy("Class").count().show()
 
（2） 设grade.txt文件的内容如下，给出分析
106,Ding,92,95,91
242,Yan,96,93,90
107,Feng,84,92,91
230,Wang,87,86,91
153,Zhang,85,90,92
224,Men,83,86,90
236,Wang1,87,85,89
210,Han,73,93,88
101,An,84,93,88
127,Pen,81,93,91
237,Du,83,81,85
 
    val lineRDD = sc.textFile("grade.txt").map{line => line.split(",")}
    val studentRDD = lineRDD.map(x => (x(0).toInt,x(1).toString,x(2).toInt,x(3).toInt,x(4).toInt))
    val studentDF = studentRDD.toDF("ID","Name","Scala","Spark","Python")
    studentDF.show()
 
（3） 设grade.txt文件的内容与（2）相同，给出分析
case class Student(ID:Int,Name:String,Scala:Int,Spark:Int,Python:Int)
val stuDS = spark.read.textFile("grade.txt")
val studentDataset = stuDS.map(line => {val x = line.split(",");
val ID = x(0).toInt;
val Name = x(1);
val Java = x(2).toInt;
val Scala = x(3).toInt;
val Python = x(4).toInt;
Student(ID,Name,Java,Scala,Python)})
studentDataset.sort(studentDataset("ID").desc).show()

（1） 使用 Spark 读取 JSON 文件并进行统计分析

// 读取 grade.json 文件并创建 DataFrame
val gradeDF=spark.read.json("grade.json")
// 统计 DataFrame 中的行数
gradeDF.count()
// 对 DataFrame 进行分组，统计每个班级的学生人数，并显示结果
gradeDF.groupBy("Class").count().show()

代码分析：

• val gradeDF=spark.read.json("grade.json"): 使用 SparkSession 对象的 read 方法读取名为 "grade.json" 的 JSON 文件，并将其转换为 DataFrame，赋值给变量 gradeDF。
• gradeDF.count(): 调用 DataFrame 的 count() 方法，统计 DataFrame 中的行数。
• gradeDF.groupBy("Class").count().show():
  • groupBy("Class"): 对 DataFrame 按 "Class" 列进行分组。
  • count(): 计算每个分组中的行数。
  • show(): 将统计结果显示出来。

（2） 使用 Spark 读取文本文件并创建 DataFrame

// 读取 grade.txt 文件并创建 RDD
val lineRDD = sc.textFile("grade.txt").map{line => line.split(",")}
// 将 RDD 中的每行数据转换为元组，并创建新的 RDD
val studentRDD = lineRDD.map(x => (x(0).toInt,x(1).toString,x(2).toInt,x(3).toInt,x(4).toInt))
// 将 RDD 转换为 DataFrame，并指定列名
val studentDF = studentRDD.toDF("ID","Name","Scala","Spark","Python")
// 显示 DataFrame
studentDF.show()

代码分析：

• val lineRDD = sc.textFile("grade.txt").map{line => line.split(",")} :
  • sc.textFile("grade.txt"): 使用 SparkContext 对象的 textFile() 方法读取名为 "grade.txt" 的文本文件，并将其转换为包含每行文本的 RDD。
  • map{line => line.split(",")} : 使用 map 操作将 RDD 中的每行文本按逗号分割成字符串数组，并创建新的 RDD。
• val studentRDD = lineRDD.map(x => (x(0).toInt,x(1).toString,x(2).toInt,x(3).toInt,x(4).toInt)):
  • 使用 map 操作将 lineRDD 中的每个字符串数组转换为元组，其中元素分别对应 ID、姓名、Scala成绩、Spark成绩、Python成绩。
• val studentDF = studentRDD.toDF("ID","Name","Scala","Spark","Python"):
  • 使用 toDF 方法将 studentRDD 转换为 DataFrame，并指定列名。
• studentDF.show(): 显示 DataFrame 的内容。

（3） 使用 Spark 读取文本文件并创建 Dataset，并进行排序

// 定义 Student 类
case class Student(ID:Int,Name:String,Scala:Int,Spark:Int,Python:Int)
// 读取 grade.txt 文件并创建 Dataset
val stuDS = spark.read.textFile("grade.txt")
// 使用 map 操作将 Dataset 中的每行文本转换为 Student 对象
val studentDataset = stuDS.map(line => {val x = line.split(",");
val ID = x(0).toInt;
val Name = x(1);
val Java = x(2).toInt;
val Scala = x(3).toInt;
val Python = x(4).toInt;
Student(ID,Name,Java,Scala,Python)})
// 对 Dataset 按 ID 列降序排序，并显示结果
studentDataset.sort(studentDataset("ID").desc).show()

代码分析：

• case class Student(ID:Int,Name:String,Scala:Int,Spark:Int,Python:Int): 定义一个名为 Student 的 case class，用于表示学生信息。
• val stuDS = spark.read.textFile("grade.txt"): 使用 SparkSession 对象的 read 方法读取名为 "grade.txt" 的文本文件，并将其转换为 Dataset，赋值给变量 stuDS。
• val studentDataset = stuDS.map(line => { ... }):
  • 使用 map 操作将 stuDS 中的每行文本转换为 Student 对象。
  • line.split(","): 将每行文本按逗号分割成字符串数组。
  • x(0).toInt: 将第一个元素转换为整型，作为 ID。
  • x(1): 将第二个元素作为姓名。
  • x(2).toInt, x(3).toInt, x(4).toInt: 将第三、四、五个元素转换为整型，分别作为 Java、Scala、Python 成绩。
  • Student(ID,Name,Java,Scala,Python): 使用 Student 类创建新的对象，并将提取出的信息赋值给对象的属性。
• studentDataset.sort(studentDataset("ID").desc).show():
  • sort(studentDataset("ID").desc): 对 studentDataset 按 "ID" 列进行降序排序。
  • show(): 显示排序后的 Dataset 的内容。

（1）这段代码分析及运行结果：
   - `spark.read.json("grade.json")`：从给定的grade.json文件中读取数据，创建一个DataFrame `gradeDF`。
   - `gradeDF.count()`：返回DataFrame的行数，这里有8行数据。
   - `gradeDF.groupBy("Class").count().show()`：按照"Class"字段对数据进行分组，并计算每组的行数，然后显示结果。
     ```
     +-----+-----+
     |Class|count|
     +-----+-----+
     |   1 |   4 |
     |   2 |   4 |
     +-----+-----+
     ```
   - 输出显示有两个班级，每个班级各有4个学生。

（2）这段代码分析及运行结果：
   - `sc.textFile("grade.txt").map{line => line.split(",")}`：读取grade.txt文件，将每一行分割成数组，并创建一个新的RDD `lineRDD`。
   - `lineRDD.map(x => (x(0).toInt,x(1).toString,x(2).toInt,x(3).toInt,x(4).toInt))`：将RDD `lineRDD`中的每个元素（数组）转换为元组，然后创建一个新的RDD `studentRDD`。
   - `studentRDD.toDF("ID","Name","Scala","Spark","Python")`：将`studentRDD`转换为DataFrame `studentDF`，并指定列名。
   - `studentDF.show()`：显示DataFrame `studentDF`的内容。
     ```
     +---+----+-----+-----+-----+
     | ID|Name|Scala|Spark|Python|
     +---+----+-----+-----+-----+
     |106| Ding|   92|   95|   91|
     |242| Yan |   96|   93|   90|
     |107| Feng|   84|   92|   91|
     |230| Wang|   87|   86|   91|
     |153|Zhang|   85|   90|   92|
     |224| Men |   83|   86|   90|
     |236|Wang1|   87|   85|   89|
     |210| Han |   73|   93|   88|
     |101|  An |   84|   93|   88|
     |127| Pen |   81|   93|   91|
     |237|  Du |   83|   81|   85|
     +---+----+-----+-----+-----+
     ```
   - 输出展示了10个学生的信息，包括ID、Name、Scala、Spark和Python的分数。

（3）这段代码分析及运行结果：
   - `case class Student(ID:Int,Name:String,Scala:Int,Spark:Int,Python:Int)`：定义了一个名为`Student`的case class，包含5个字段。
   - `spark.read.textFile("grade.txt")`：读取grade.txt文件，创建RDD `stuDS`。
   - `stuDS.map...`：将RDD `stuDS`中的每一行数据转换为`Student`对象，然后创建一个新的Dataset `studentDataset`。
   - `studentDataset.sort(studentDataset("ID").desc).show()`：按ID字段降序排序`studentDataset`，然后显示结果。
     ```
     +---+----+-----+-----+-----+
     | ID|Name|Java |Scala|Python|
     +---+----+-----+-----+-----+
     |237|  Du|   83|   81|   85|
     |236|Wang1|   87|   85|   89|
     |230| Wang|   87|   86|   91|
     |224| Men |   83|   86|   90|
     |210| Han |   73|   93|   88|
     |127| Pen |   81|   93|   91|
     |107| Feng|   84|   92|   91|
     |106| Ding|   92|   95|   91