Spark SQL内置函数

Apache Spark SQL 提供了一组丰富的内置函数（或称为UDFs，用户定义函数），用于在查询时处理数据。这些函数涵盖了字符串操作、日期/时间处理、数学运算、聚合等多个方面。以下是一些 Spark SQL 中常见的内置函数类别及其示例：

1. 字符串函数

   • concat(str1, str2, ...): 连接两个或多个字符串。
   • length(str): 返回字符串的长度。
   • upper(str): 将字符串转换为大写。
   • lower(str): 将字符串转换为小写。
   • trim(str): 去除字符串两端的空格。
   • regexp_replace(str, pattern, replacement): 使用正则表达式替换字符串中的子串。

2. 日期/时间函数

   • current_date: 返回当前日期。
   • current_timestamp: 返回当前日期和时间。
   • date_format(date/timestamp, format): 将日期/时间格式化为指定格式的字符串。
   • year(date/timestamp): 提取日期/时间的年份部分。
   • month(date/timestamp): 提取日期/时间的月份部分。
   • dayofmonth(date/timestamp): 提取日期/时间的月份中的第几天。

3. 数学函数

   • abs(num): 返回数的绝对值。
   • round(num, d): 将数四舍五入到指定的小数位数。
   • floor(num): 返回小于或等于给定数字的最大整数。
   • ceil(num) 或 ceiling(num): 返回大于或等于给定数字的最小整数。
   • sqrt(num): 返回数的平方根。
   • pow(num, exponent): 返回数的指数幂。

4. 聚合函数

   • count(*): 计算行数。
   • sum(col): 计算列的总和。
   • avg(col): 计算列的平均值。
   • min(col): 返回列的最小值。
   • max(col): 返回列的最大值。
   • collect_list(col): 将列的值收集到一个列表中。

5. 其他常用函数

   • coalesce(value1, value2, ...): 返回参数列表中的第一个非空值。
   • ifnull(value1, value2): 如果 value1 为空，则返回 value2，否则返回 value1。
   • case when ... then ... else ... end: 条件表达式，类似于 SQL 中的 CASE 语句。

6. 窗口函数

   • row_number() over (order by ...): 为结果集中的每一行分配一个唯一的连续整数。
   • rank() over (order by ...): 为结果集中的每一行分配一个排名，考虑平级关系。
   • dense_rank() over (order by ...): 为结果集中的每一行分配一个排名，不考虑平级关系（即没有间隔）。
   • lead(col) over (order by ...): 返回当前行之后的第一行的值。
   • lag(col) over (order by ...): 返回当前行之前的第一行的值。

这只是 Spark SQL 提供的内置函数的一小部分示例。要获取完整的函数列表和详细文档，请参阅 Spark 官方文档。

以下是一些使用Spark SQL内置函数的代码示例。

示例1：字符串函数

假设我们有一个包含用户信息的DataFrame，其中有一个名为name的列，我们想要将这一列的名字转换为大写。

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Spark SQL Built-in Functions Example")
  .getOrCreate()

import spark.implicits._

// 创建一个简单的DataFrame
val data = Seq(("John Doe"), ("Jane Smith"), ("Alice Johnson"))
val df = data.toDF("name")

// 使用upper函数将名字转换为大写
val uppercasedNamesDF = df.withColumn("uppercase_name", upper($"name"))

uppercasedNamesDF.show()

示例2：日期函数

假设我们有一个包含日期的DataFrame，我们想要提取出年份和月份。

// 创建一个包含日期的DataFrame
val dateData = Seq(("2023-03-15"), ("2022-07-01"), ("2023-01-31"))
val datesDF = dateData.toDF("date_string")
  .withColumn("date", to_date($"date_string", "yyyy-MM-dd")) // 将字符串转换为日期类型

// 提取年份和月份
val yearMonthDF = datesDF.withColumn("year", year($"date"))
  .withColumn("month", month($"date"))

yearMonthDF.show()

示例3：数学函数

假设我们有一个包含数字的DataFrame，我们想要对这些数字进行四舍五入和取平方根。

// 创建一个包含数字的DataFrame
val numberData = Seq((10.5678), (15.2345), (9.8765))
val numbersDF = numberData.toDF("number")

// 对数字进行四舍五入到小数点后两位，并取平方根
val roundedAndSqrtDF = numbersDF.withColumn("rounded_number", round($"number", 2))
  .withColumn("sqrt_number", sqrt($"number"))

roundedAndSqrtDF.show()

示例4：聚合函数

假设我们有一个销售数据的DataFrame，我们想要计算总销售额和平均销售额。

// 创建一个销售数据的DataFrame
val salesData = Seq(
  ("Product1", 100.0),
  ("Product2", 150.0),
  ("Product1", 200.0),
  ("Product3", 120.0),
  ("Product2", 180.0)
)
val salesDF = salesData.toDF("product", "sale_amount")

// 计算总销售额和平均销售额
val totalSales = salesDF.agg(sum($"sale_amount").alias("total_sales")).first().getAs[Double]("total_sales")
val averageSales = salesDF.agg(avg($"sale_amount").alias("average_sales")).first().getAs[Double]("average_sales")

println(s"Total Sales: $totalSales")
println(s"Average Sales: $averageSales")

这些示例展示了如何在Spark SQL中使用不同类型的内置函数来处理数据。请注意，为了运行这些示例，你需要有一个运行中的Spark环境，并且这些代码应该在Spark的驱动程序中执行。