pyspark 编写 UDF函数

前言

以前用的是Scala，最近有个东西要用Python，就查了一下如何编写pyspark的UDF。

pyspark udf

1. 也是先定义一个函数，例如：

def get_time(ts):

    try:

        res = datetime.fromtimestamp(int(ts)).strftime('%Y-%m-%d')

        return res

    except:

        return ''

    else:

        return ''

1. udf的注册，这里需要定义其返回值类型，可从pyspark.sql.types中引入，如下

from pyspark.sql.functions import udf, col

from pyspark.sql.types import StringType

time_udfs = udf(get_time, StringType())

1. udf的使用

abc.select("id",  time_udfs(abc["updated"])).distinct()

参考

• https://docs.databricks.com/spark/latest/spark-sql/udf-in-python.html