Spark-Scala语言实战（18）SQL-DataFrame方法

之前的文章中学习了在集群环境下编写我们的spark已经在IDEA中打jar包。想了解的朋友可以查看这篇文章。同时，希望我的文章能帮助到你，如果觉得我的文章写的不错，请留下你宝贵的点赞，谢谢。

Spark-Scala语言实战（17）-CSDN博客文章浏览阅读2.3k次，点赞46次，收藏39次。这篇文章起我会带来两种方式将我们开发环境（IDEA）的代码文件打jar包到我们的集群环境下运行。今天的文章首先来讲我们的本地模式。今天的文章，我会带着大家一起来到Linux集群环境下，学习我们的spark。希望我的文章能帮助到大家，也欢迎大家来我的文章下交流讨论，共同进步。https://blog.csdn.net/qq_49513817/article/details/137929033?spm=1001.2014.3001.5501这篇文章我会带着大家学习Spark SQL中的DataFrame中show()，selectExpr()，select()，filter()/where，groupBy()，sort()6种方法。了解Spark SQL以及DataFrame。

目录

一、结构化数据

Spark SQL

什么是Spark SQL

Spark SQL架构 

DataFrame

什么是DataFrame

二、DataFrame方法 

1.show()

2.selectExpr()

3.select()

4.filter()/where

5.groupBy()

6.sort()

一、结构化数据

Spark SQL

什么是Spark SQL

• Spark SQL 是一个用于处理结构化数据的 Spark 模块。它允许你使用 SQL 语句来查询 DataFrame。
• Spark SQL 提供了 DataFrame 和 DataSet 的 API，这些 API 允许你以编程方式执行 SQL 查询。此外，它还提供了一个 SQL 引擎，允许你直接在 Spark 集群上执行 SQL 语句。
• Spark SQL 支持多种数据源，包括 Hive、JSON、Parquet、CSV 等。它还支持多种外部数据库和存储系统，如 JDBC、ODBC 和 Apache Arrow。
• Spark SQL 的主要目标是使数据处理更加简单、直观和高效。通过使用 SQL 和 DataFrame API，你可以轻松地执行复杂的数据转换和查询操作，而无需编写大量的底层代码。

  Spark SQL架构 

   spark sql架构图

  Spark SQL架构与Hive架构相比，把底层的MapReduce执行引擎更改为Spark，还修改了Catalyst优化器，Spark SQL快速的计算效率得益于Catalyst优化器。从HiveQL被解析成语法抽象树起，执行计划生成和优化的工作全部交给Spark SQL的Catalyst优化器进行负责和管理。 

  DataFrame

  什么是DataFrame

  • DataFrame 是一个分布式数据集，它有一个不可变的、不可空的 schema（即列和类型）。这意味着 DataFrame 知道它的每一列都包含什么类型的数据。
  • DataFrame 可以从各种来源创建，如结构化文件（如 Parquet、JSON）、Hive 表、外部数据库或现有的 RDD（通过执行转换操作）。
  • DataFrame API 提供了丰富的函数来操作数据，包括选择、过滤、分组、排序等。这些操作通常使用 DataFrame 的 DSL（领域特定语言）风格的方法来实现，这使得代码更加简洁和易于理解。
  • DataFrame 底层由 RDD 支持，但 RDD 是低级的、不可读的，并且没有 schema。相比之下，DataFrame 提供了更高级别的抽象，使得处理结构化数据更加容易。

    二、DataFrame方法 

    1.show()

     show()方法用来查看我们DataFrame数据，show方法默认只显示我们数据中前20条记录，且最多显示20个字符

    方法参数	说明
    show()	显示 DataFrame 的前 20 行数据（默认）。
    show(n)	显示 DataFrame 的前 n 行数据。其中 n 是一个整数，表示要显示的行数。
    show(truncate=False)	显示 DataFrame 的所有列，不对列的内容进行截断。默认情况下，如果列的内容过长，Spark 会截断它以适应控制台输出。将此参数设置为 False 可以避免截断。
    show(n, truncate=False)	同时设置显示行数和是否截断列的内容。这将显示前 n 行数据，并且不对列的内容进行截断。

    示例代码，显示内容: 

    import org.apache.spark.sql.SparkSessionobject DataFrameExample {  def main(args: Array[String]): Unit = {    // 创建一个SparkSession    val spark = SparkSession.builder()      .appName("Peng")      .master("local[*]") // 使用所有可用的本地线程      .getOrCreate()    import spark.implicits._    // 创建一个简单的Seq来作为数据源    val data = Seq(("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3),("Alice", 1), ("Bob", 2), ("Charlie", 3))    // 将Seq转换为DataFrame    case class Person(name: String, age: Int)    val p = data.toDF("name", "age") // 或者使用data.map(Person.tupled).toDF()    // 显示DataFrame的内容    p.show()  }}

    可以看到我们的数据集超过了20条，那么使用我们的show（）方法不添加参数肯定也只会输出20条。

    2.selectExpr()

    selectExpr() 方法在 Apache Spark 的 DataFrame API 中用于选择列，并且可以支持 SQL 表达式，这样我们就可以对字段进行特殊处理。 

    参数	说明
    exprs: String*	一个或多个 SQL 表达式字符串，用于选择或计算 DataFrame 中的列。

    示例代码，使用 selectExpr() 方法将年龄+1:

    import org.apache.spark.sql.SparkSessionobject p1 {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder()      .appName("PENG")      .master("local[*]")      .getOrCreate()    import spark.implicits._ // 导入隐式转换和DSL    // 创建一个简单的Seq来作为数据源      val data = Seq(("Alice", 28), ("Bob", 22), ("Charlie", 30), ("David", 25))    // 将Seq转换为DataFrame      val peopleDF = data.toDF("name", "age")    // 使用selectExpr()方法选择列并进行一些计算      val selectedDF = peopleDF.selectExpr(      "name", // 选择name列        "age",  // 选择age列        "(age + 1) as new_age" // 使用SQL表达式计算新的列new_age      )    // 显示选择后的DataFrame的内容      selectedDF.show()  }}

    3.select()

     select() 方法在 Apache Spark 的 DataFrame API 中用于选择 DataFrame 中的列，获取相对应的值。

    参数	说明
    cols: Column*	一个或多个 Column 表达式，用于指定要选择的列。这些可以是列名、Column 对象或 Column 表达式。

    示例代码，对列重命名： 

    import org.apache.spark.sql.SparkSessionobject DataFrameSelectExample {  def main(args: Array[String]): Unit = {    // 创建一个SparkSession    val spark = SparkSession.builder()      .appName("Peng")      .master("local[*]") // 使用所有可用的本地线程      .getOrCreate()    import spark.implicits._ // 导入隐式转换和DSL    // 创建一个简单的Seq来作为数据源    val data = Seq(("Alice", 28), ("Bob", 22), ("Charlie", 30), ("David", 25))    // 将Seq转换为DataFrame    val peopleDF = data.toDF("name", "age")    // 使用select()方法选择列    val selectedDF = peopleDF.select($"name", $"age")    // 显示选择后的DataFrame的内容    selectedDF.show()    // 对列进行重命名，直接使用$"columnName".as("newName")    val renamedDF = peopleDF.select($"name", $"age".as("yearsOld"))    // 显示重命名后的DataFrame的内容    renamedDF.show()  }}

    4.filter()/where

     使用filter或where方法可以查询数据中符合条件的所有字段信息。

    方法	参数	类型	说明	示例
    filter()	conditionExpr: Column	Column	使用Column表达式来过滤DataFrame的行。这可以是一个Column对象、SQL表达式字符串或Scala中的Column API表达式。	df.filter($"column1" === 1).show()
    where()	conditionExpr: String	String	使用SQL表达式字符串来过滤DataFrame的行。	df.where("column1 = 1").show()

    示例代码，实现过滤年龄大小：

    import org.apache.spark.sql.SparkSessionobject p1 {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder()      .appName("PENG")      .master("local[*]")      .getOrCreate()    import spark.implicits._ // 导入隐式转换和DSL    // 创建一个简单的Seq来作为数据源      val data = Seq(("Alice", 28), ("Bob", 22), ("Charlie", 30), ("David", 25))    // 将Seq转换为DataFrame    val peopleDF = data.toDF("name", "age")    // 使用filter()方法过滤年龄大于25的人    val filteredDF1 = peopleDF.filter($"age" > 25)    // 使用where()方法过滤年龄小于30的人    val filteredDF2 = peopleDF.where($"age"   
    5.groupBy()
     
    groupBy() 方法接受一个或多个列名、列表达式或列引用作为参数，并返回一个分组后的对象（在 Spark SQL 中是 RelationalGroupedDataset），该对象表示按指定列或表达式分组的数据。 
     

    属性/参数	类型	说明	示例
    分组列	列名、表达式、列表等	指定要按照哪些列或表达式对数据进行分组	df.groupBy("columnName") 或 df.groupBy($"columnName")
    聚合函数	聚合函数	在每个分组上应用的聚合函数，如count(), sum(), avg(), max(), min()等	df.groupBy("columnName").count().show()
    返回值	RelationalGroupedDataset	返回一个RelationalGroupedDataset对象，该对象可以进一步调用聚合函数	val grouped = df.groupBy("columnName")
    1. 返回值	RelationalGroupedDataset	groupBy()方法返回一个RelationalGroupedDataset对象，而不是一个普通的DataFrame。该对象允许你进一步调用聚合函数。	-
    2. 排序	-	在使用groupBy()方法之前，通常不需要对数据进行排序，因为Spark会根据分组列的值自动进行排序。但如果你需要按照特定的顺序输出结果，可以在调用groupBy()之后使用orderBy()方法。	grouped.orderBy("columnName").show()
    3. 分组键	-	分组后的结果中，分组键会成为结果DataFrame的索引（在Spark SQL中）。如果你不希望这样，可以使用drop()方法删除索引列。	grouped.drop("columnName").show()
    4. 多列分组	-	你可以通过传递一个包含多个列名的列表或元组来按多列进行分组。	df.groupBy("columnName1", "columnName2").count().show()
    5. 聚合多个指标	-	你可以在一个groupBy()调用中聚合多个指标。这可以通过使用agg()方法并传递一个包含多个聚合函数的列表来实现。	df.groupBy("columnName").agg(sum("columnValue"), avg("columnValue")).show(）

    下面是使用groupBy实现简单分组的示例代码： 

    import org.apache.spark.sql.SparkSessionobject p1 {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder()      .appName("PENG")      .master("local[*]")      .getOrCreate()    import spark.implicits._    // 创建一个简单的 DataFrame    val data = Seq(("A", 1), ("B", 2), ("A", 3), ("B", 1), ("C", 2))    val df = data.toDF("category", "value")    // 使用 groupBy 对 category 列进行分组，并计算每个分组中的行数    val p = df.groupBy("category").count()    p.show()  }}

     

    6.sort()

      sort() 方法通常用于对数据进行排序。

    属性/参数	类型	说明	示例
    col	str 或 Column 或 list of str/Column	要按其进行排序的列名、Column 对象或列名/Column 对象的列表。	df.sort("columnName") 或 df.sort(df("columnName").desc) 或 df.sort(col("columnName1"), col("columnName2").desc)
    ascending	bool 或 list of bool	是否按升序排序，默认为 True。如果提供了多个列，则可以传递一个布尔值列表来指定每个列的排序顺序。	df.sort("columnName", ascending=False)
    global	bool	在分布式环境中，是否进行全局排序。请注意，全局排序可能非常昂贵，并且可能不适用于大数据集。默认为 False（即分区内排序）。	通常不直接设置此参数，而是通过其他方式（如 repartition）来控制排序的范围。

    以下是对年龄排序的示例代码： 

    import org.apache.spark.sql.SparkSessionobject p1 {  def main(args: Array[String]): Unit = {    val spark = SparkSession.builder()      .appName("PENG")      .master("local[*]")      .getOrCreate()    import spark.implicits._ // 导入隐式转换和DSL    // 创建一个简单的Seq来作为数据源    val data = Seq(("Alice", 28), ("Bob", 22), ("Charlie", 30), ("David", 25))    // 将Seq转换为DataFrame    val peopleDF = data.toDF("name", "age")    // 对DataFrame按年龄进行排序    val sortedDF = peopleDF.sort("age") // 默认为升序    // 显示排序后的DataFrame的内容    sortedDF.show()    // 如果你想要降序排序，可以添加一个降序参数    val sortedDescDF = peopleDF.sort($"age".desc)    // 显示降序排序后的DataFrame的内容    sortedDescDF.show()  }}