头歌sparksql自定义函数

### 创建和使用自定义函数（UDF）

在 Spark SQL 中，可以通过 `UserDefinedFunction` 接口来创建 UDF 函数。这些函数可以用于执行特定的数据转换操作，尤其是在内置函数无法满足需求的情况下非常有用。

#### 导入必要的包

为了能够注册并使用 UDF 函数，首先需要导入相应的库：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf

#### 注册简单的 Scala/Python 函数作为 UDF

对于较为简单的情况可以直接将现有的 Scala 或 Python 方法转化为 UDF 并向 DataFrame API 或者 SQL 查询中注册该方法以便后续调用[^1]。

##### 使用Scala:

val spark = SparkSession.builder.appName("example").getOrCreate()

// 定义一个普通的Scala函数
def myFunc(x: Int): String = {
    if (x > 0) "positive"
    else "non-positive"
}

// 将其注册为UDF
val myUdf = udf(myFunc _)

spark.udf.register("myCustomFunc", myFunc _) // 可选：通过名称注册到SQL上下文中

##### 使用Python:

from pyspark.sql import SparkSession
from pyspark.sql.types import StringType

spark = SparkSession.builder.appName('example').getOrCreate()

# 定义一个普通的Python函数
def python_my_func(x):
    return 'positive' if x > 0 else 'non-positive'

# 转换为UDF
udf_python_my_func = udf(python_my_func, StringType())

spark.udf.register("pyMyCustomFunc", python_my_func, StringType()) # 同样可以选择性地将其命名为可查询的形式

#### 应用已注册的 UDF 到DataFrame或SQL语句

一旦完成了上述步骤，则可以在 DataFrames 上应用此新创建好的 UDF ，也可以直接写成 SQL 表达式形式来进行更复杂的逻辑运算。

SELECT col_name, myCustomFunc(col_name) FROM table_name;

或者是在 DataFrame 的列上直接调用已经注册过的 UDF :

val dfWithNewColumn = originalDf.withColumn("new_col", myUdf($"existing_column"))