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

的列，我们想要将这一列的名字转换为大写。

importorg.apache.spark.sql.SparkSession

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

// 创建一个简单的DataFrameval 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，我们想要提取出年份和月份。

// 创建一个包含日期的DataFrameval 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，我们想要对这些数字进行四舍五入和取平方根。

// 创建一个包含数字的DataFrameval 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，我们想要计算总销售额和平均销售额。

// 创建一个销售数据的DataFrameval 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的驱动程序中执行。