pyspark系列--自定义函数

Myth丶恋晨 2022-05-29 12:41 240阅读 0赞

**自定义函数**

*  1. 概览
 *  2. 自定义函数的一般流程
 *  3. 简单的自定义函数
 *  4. 自定义函数进阶

# 1. 概览 #

自定义函数的重点在于定义返回值类型的数据格式，其数据类型基本都是从`from pyspark.sql.types import *` 导入，常用的包括：  
\- StructType()：结构体  
\- StructField()：结构体中的元素  
\- LongType()：长整型  
\- StringType()：字符串  
\- IntegerType()：一般整型  
\- FloatType()：浮点型

还记得我们在前面的创建`spark.dataframe`提到的例子吗，dataframe的数据结构定义如下：

from pyspark.sql.types import StructType, StructField, LongType, StringType
    schema = StructType([
        StructField("id", LongType(), True),
        StructField("name", StringType(), True),
        StructField("age", LongType(), True),
        StructField("eyeColor", StringType(), True)
    ])

# 2. 自定义函数的一般流程 #

# 1.创建普通的python函数
    def toDate(s):
        return str(s)+'-'
    
    # 2.注册自定义函数
    from pyspark.sql.functions import udf
    from pyspark.sql.types import StringType
    
    # 根据python的返回值类型定义好spark对应的数据类型
    # python函数中返回的是string，对应的pyspark是StringType
    toDateUDF=udf(toDate, StringType())  
    
    # 使用自定义函数
    df1.withColumn('color',toDateUDF('color')).show()

# 3. 简单的自定义函数 #

最简单的就是通过lambda函数，不需要定义返回值类型，可以直接使用

# 创建udf自定义函数
    from pyspark.sql import functions
    concat_func = functions.udf(lambda name,age:name+'_'+str(age))  # 简单的连接两个字符串
    
    # 应用自定义函数
    concat_df = spark_df.withColumn("name_age",concat_func(final_data.name, final_data.age))
    concat_df.show()

# 4. 自定义函数进阶 #

对于简单的自定义函数，用上面的方法是足够了，但是很多时候，我们面对的问题比较复杂，需要对返回值的格式要求比较特别，就需要定义复杂的返回值类型了。