pyspark 編寫 UDF函數

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