来源：https://sparkbyexamples.com/pyspark/pyspark-udf-user-defined-function/
PySpark UDF例子
PySpark UDF(又名用户定义函数)是Spark SQL & DataFrame中最有用的特性，用于扩展PySpark构建的功能。在本文中，我将解释什么是UDF?为什么我们需要它，以及如何创建和使用它在DataFrame select()， withColumn()和SQL使用PySpark (Spark with Python)示例。
注意:UDF是最昂贵的操作，因此只有在必要时才使用它们。在本文后面的部分中，我将详细解释为什么使用UDF是一个昂贵的操作。
1. PySpark UDF的介绍
1.1什么是UDF?
UDF的别名是用户定义函数，如果您来自SQL背景，那么UDF对您来说并不新鲜，因为大多数传统RDBMS数据库都支持用户定义函数，这些函数需要在数据库库中注册，并在SQL中将它们作为常规函数使用。
PySpark的UDF与传统数据库上的UDF类似。在PySpark中，您可以在Python语法中创建一个函数，并用PySpark SQL udf()包装它，或者将它注册为udf，并分别在DataFrame和SQL上使用它。
1.2为什么我们需要一个UDF?
这些函数用于扩展框架的函数，并在多个DataFrame上重用这些函数。例如，您希望将名称字符串中单词的每个首字母都转换为大写字母;PySpark的内置特性没有这个函数，因此您可以为它创建一个UDF，并根据需要在许多DataFrame上重用它。UDF一旦创建，就可以在多个数据框架和SQL表达式上重用。
在创建任何UDF之前，先研究一下Spark SQL函数中是否已经有您想要的类似函数。PySpark SQL提供了一些预定义的通用函数，并且在每个版本中都添加了更多的新函数。因此，在你重新发明轮子之前最好检查一下。
当你创建UDF时，你需要非常仔细地设计它们，否则你会遇到优化和性能问题。
2. 创建PySpark UDF
2.1创建数据帧
在我们开始创建UDF之前，首先让我们创建一个PySpark DataFrame。
spark=SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
columns=["Seqno","Name"]data=[("1","john jones"),("2","tracey smith"),("3","amy sanders")]
df=spark.createDataFrame(data=data,schema=columns)
df.show(truncate=False)
产生以下输出。
2.2创建一个Python函数
创建UDF的第一步是创建Python函数。下面的代码片段创建了一个函数convertCase()，它接受一个字符串参数，并将每个单词的第一个字母转换为大写字母。UDF接受您选择的参数并返回一个值。
2.3将Python函数转换为PySpark UDF
现在，通过将函数传递给PySpark SQL UDF()，将这个函数convertCase()转换为UDF，这个函数可以在org.apache.spark.sql.functions.UDF中找到。请确保在使用这个包之前import它。
PySpark SQL udf()函数返回org.apache.spark.sql.expressions.UserDefinedFunction类对象。
注意:udf()的默认类型是StringType，因此，您也可以在不返回类型的情况下编写上述语句。
3.在DataFrame中使用UDF
3.1利用PySpark DataFrame select()使用UDF
现在可以在DataFrame列上使用convertUDF()作为常规的内置函数。
结果如下。
3.2在PySpark DataFrame withColumn()中使用UDF
您也可以在DataFrame withColumn()函数上使用udf，为了解释这一点，我将创建另一个upperCase()函数，它将输入字符串转换为大写字母。
让我们将upperCase() python函数转换为UDF，然后将其与DataFrame withColumn()一起使用。下面的示例将“Name”列的值转换为大写，并创建一个新列“Curated Name”。
结果如下。
3.3注册PySpark UDF &在SQL上使用它
为了在PySpark SQL上使用convertCase()函数，您需要通过使用spark.udf.register()向PySpark注册该函数。
产生与3.1相似的结果。
4. 使用注释创建UDF
在前面的小节中，您已经了解了创建UDF需要两个步骤，首先，您需要创建一个Python函数，然后使用SQL UDF()函数将函数转换为UDF，但是，您可以避免这两个步骤，通过使用注释只需一个步骤就可以创建它。
其结果与3.2相似。
5. 特殊处理
5.1执行顺序
需要注意的一点是，在PySpark/Spark中并不保证子表达式求值的顺序，这意味着表达式不能保证从左到右或以任何其他固定的顺序求值。PySpark重新排序查询优化和计划的执行顺序，因此，AND, OR, WHERE and HAVING 将产生副作用。
因此，在设计和使用UDF时，必须非常小心，尤其是null处理，因为这会导致运行时异常。
5.2处理null
如果不仔细设计，UDF很容易出错。例如，当您有一个列在某些记录上包含值null时。
注意，在上面的代码片段中，带有“Seqno 4”的记录的“name”列的值为“None”。因为我们没有使用UDF函数处理null，所以在DataFrame上使用这个函数返回错误。注意，在Python中None被认为是空的。
要记住以下几点：
（1）最好的做法是在UDF函数内部检查是否为空，而不是在外部检查是否为空。
（2）在任何情况下，如果您不能在UDF中执行null检查，那么至少使用if或case WHEN来检查null并有条件地调用UDF。
当我们在注册UDF时检查null/none时，这将成功执行，不会出现错误。
5.3使用UDF的性能问题
UDF对PySpark来说是一个黑盒，因此它不能应用优化，你将失去所有PySpark在Dataframe/Dataset上做的优化。如果可能的话，应该使用Spark SQL内置函数，因为这些函数提供了优化。请考虑仅在现有的内置SQL函数没有UDF时创建UDF。
6 样例代码
https://github.com/spark-examples/pyspark-examples/blob/master/pyspark-udf.py
结论
在本文中，您已经了解了以下内容：
（1）PySpark UDF是一个用户定义的函数，用于在Spark中创建一个可重用的函数。
（2）一旦创建了UDF，就可以在多个DataFrames和SQL上重用它(注册后)。
（3）udf()的默认类型是StringType。
（4）您需要显式地处理空值，否则您将看到副作用。
网友评论