Pandas用户定义函数（udf）是apachesparktm中用于数据科学的最重要的增强之一。它们带来了许多好处，比如允许用户使用Pandas api和提高性能。然而，Pandas udf随着时间的推移已经有了有机的发展，这导致了一些不一致的地方，并在用户中造成了混乱。apachespark3.0的完整版本很快就会发布，它将为Pandas UDF引入一个新的接口，它利用Python类型提示来解决Pandas UDF类型的激增，并帮助它们变得更加Python和自我描述。这篇博客文章介绍了带有Python类型提示的新Pandas udf，以及新的Pandas函数api，包括分组map、map和co-grouped-map。熊猫UDFPandas UDF是在Spark 2.3中引入的，另请参见为PySpark引入Pandas UDF。Pandas是数据科学家所熟知的，它与许多Python库和包（如NumPy、statsmodel和scikit learn）进行了无缝集成，Pandas udf不仅允许数据科学家扩展其工作负载，还可以利用Apache Spark中的Pandas api。用户定义的函数由以下方式执行：apachearrow，在JVM和Python驱动程序/执行器之间直接交换数据，串行化成本几乎为零。函数内的Pandas，使用Pandas实例和api。Pandas udf与函数内的Pandas api和apachearrow一起工作以交换数据。它允许向量化操作，与每次一行的Python udf相比，可以将性能提高100倍。下面的示例显示了一个Pandas自定义项，它只需在每个值上添加一个，在该自定义项中，它是用名为Pandas_plus_one的函数定义的，由Pandas_UDF修饰，Pandas UDF类型指定为PandasudType.SCALAR.从pyspark.sql.functions导入pandas_udf，PandasUDFType@熊猫‘，PandasudType.SCALAR)定义熊猫加一（v）：#‘v’是熊猫系列return v.add（1）#输出熊猫系列火花射程（10） .select（pandas_plus_one（"id"））.show（）Python函数获取并输出Pandas系列。通过使用这个函数中丰富的Pandas api集，您可以执行向每个值添加一个值的向量化操作。（De）序列化也通过在幕后利用apachearrow自动进行矢量化。Python类型提示Python类型提示是在python3.5的pep484中正式引入的。类型暗示是Python中静态指示值类型的一种官方方法。请参阅下面的示例。def greeting（名称：str）->str:return'Hello'+namename:stri表示name参数为str类型，->语法表示greeting（）函数返回字符串。Python类型提示为PySpark和Pandas UDF上下文带来了两个显著的好处。它给出了函数应该做什么的明确定义，使用户更容易理解代码。例如，除非有文档记录，否则用户无法知道如果没有类型提示，问候语是否可以接受。它可以避免用一堆测试用例和/或让用户自己测试和解决这些微妙的案例。它可以使静态分析更容易进行。像PyCharm和visualstudio代码这样的ide可以利用类型注释来提供代码完成、显示错误并支持更好的转到定义功能。大熊猫UDF类型激增自从apachespark2.3发布以来，已经实现了许多新的Pandas udf，这使得用户很难了解新规范以及如何使用它们。例如，以下三个Pandas udf输出的结果几乎相同：从pyspark.sql.functions导入pandas_udf，PandasUDFType@熊猫"长"，PandasudType.SCALAR)定义熊猫加一（v）：#‘v’是熊猫系列return v+1#输出熊猫系列火花射程（10） .select（pandas_plus_one（"id"））.show（）从pyspark.sql.functions导入pandas_udf，PandasUDFType#Spark 3.0中的新型熊猫UDF。@熊猫"长"，PandasudType.SCALAR编辑器)def熊猫加一号（itr）：#iterator是pandas系列的迭代器。return map（lambda v:v+1，itr）#输出pandas系列的迭代器。火花射程（10） .select（pandas_plus_one（"id"））.show（）从pyspark.sql.functions导入pandas_udf，PandasUDFType@pandas_udf（"id long"，PandasudType.GROUPED_映射)def pandas_+u-one（pdf）：#"pdf"是熊猫数据框return pdf+1#输出pandas数据帧#`pandas_plus_one`只能与`groupby（…）.apply（…）一起使用`火花射程（10） .groupby（'id'）.apply（熊猫加上1）.show（）尽管这些UDF类型中的每一种都有不同的用途，但也有几种可以应用。在这个简单的例子中，您可以使用这三种方法中的任何一种。然而，每个Pandas udf期望不同的输入和输出类型，并且以不同的方式工作，具有不同的语义和不同的性能。它混淆了用户使用和学习哪一个，以及每一个是如何工作的。此外，在第一和第二种情况下，pandas_plus_one可用于使用常规PySpark列的情况。考虑withColumn的参数或函数与其他表达式的组合，例如pandas_plus_one（"id"）+1。但是，最后一个pandas_plus_one只能与groupby（…）.apply（pandas_plus_one）一起使用。这种复杂程度引发了与Spark开发人员的多次讨论，并推动了通过官方提案引入新的Pandas api和Python类型提示的努力。其目标是使用户能够自然地使用Python类型提示来表达pandas udf，而不会像上面有问题的情况那样混淆。例如，上述情况可写如下：def熊猫加一（v：pd系列) -> pd系列:返回v+1def pandas_plus_one（itr：迭代器[pd系列])->迭代器[pd系列]:返回图（lambda v:v+1，itr）def pandas_plus_一（pdf:pd数据帧) -> pd数据帧:返回pdf+1带有Python类型提示的新Pandas api为了解决旧Pandas udf中的复杂性，从apachespark3.0到python3.6及更高版本，Python类型提示如下熊猫系列, 熊猫.DataFrame、Tuple和Iterator可用于表示新的Pandas UDF类型。此外，旧的Pandas udf被分成两个API类别：Pandas udf和Pandas函数API。虽然它们在内部的工作方式相似，但存在明显的差异。可以使用与使用其他PySpark列实例相同的方式处理Pandas udf。但是，不能对这些列实例使用Pandas函数api。以下是两个例子：#熊猫UDF将熊猫作为pd导入从pyspark.sql.functions导入pandas_udf，log2，col@熊猫（long）def熊猫pd系列) -> pd系列:返回s+1#pandas_plus_one（"id"）在内部被视为SQL表达式。#也就是说，可以与其他列、函数和表达式组合。火花射程（10） 。选择(熊猫加一（col（"id"）-1）+log2（"id"）+1.show（）#熊猫函数API从输入import Iterator将熊猫作为pd导入def pandas_plus_one（迭代器：迭代器[pd数据帧])->迭代器[pd数据帧]:返回映射（lambda v:v+1，迭代器）#pandas是一个普通的Python函数，mapInPandas是#逻辑上被视为单独的SQL查询计划而不是SQL表达式。#因此，与其他表达式的直接交互是不可能的。火花射程（10） .mapInPandas（pandas_plus_one，schema="id long"）.show（）另外，请注意Pandas udf需要Python类型提示，而Pandas函数api中的类型提示目前是可选的。类型提示是为Pandas函数api设计的，将来可能需要使用。新熊猫UDF新的Pandas UDF不是手动定义和指定每个Pandas UDF类型，而是根据Python函数中给定的Python类型提示推断Pandas UDF类型。目前Pandas udf中支持的Python类型提示有四种情况：系列到系列级数迭代器到级数迭代器多序列迭代器到级数迭代器序列到标量（单个值）在深入研究每一个案例之前，让我们先看看与新熊猫UDF合作的三个关键点。尽管Python类型提示在Python世界中通常是可选的，但是为了使用新的Pandas udf，必须为输入和输出指定Python类型提示。用户仍然可以通过手动指定Pandas UDF类型来使用旧方法。但是，鼓励使用Python类型提示。类型提示应该使用熊猫系列在所有情况下。但是，有一种变体熊猫.DataFrame应改为用于其输入或输出类型提示：当输入或输出列为StructType.Take看看这个例子下图：进口熊猫作为pd从pyspark.sql.functions进口熊猫df=spark.createDataFrame([[1，"字符串"，（"嵌套字符串"，）]]，"long_col long，string_col string，struct_col struct"）@pandas_udf（"col1字符串，col2 long"）定义熊猫(第一节：pd系列，s2：pd系列，pdf：pd数据帧) -> pd数据帧:#常规列是series，struct列是DataFrame。pdf['col2']=s1+s2。结构长度()return pdf#struct列需要返回一个DataFrame数据框选择（pandas_plus_len（"long_col"，"string_col"，"struct_col"））.show（）系列到系列Series-to-Series映射到apachespark 2.3中引入的scalar Pandas UDF。类型提示可以表示为熊猫系列, ... -> 熊猫系列. 它期望给定函数使用一个或多个熊猫系列输出一个熊猫系列. 输出长度应与输入长度相同。将熊猫作为pd导入从pyspark.sql.functions进口熊猫@熊猫（long）def熊猫pd系列)