带有函数的pySpark withColumn

Question

我有一个dataframe，它有两列: account

_

id和电子邮件

_

address，现在我想再添加一列'updated

_

电子邮件

_

地址‘，我在电子邮件中调用了一些函数

_

获取更新的地址

_

电子邮件

_

地址。下面是我的代码：

def update_email(email):
  print("== email to be updated: " + email)
  today = datetime.date.today()
  updated = substring(email, -8, 8) + str(today.strftime('%m')) + str(today.strftime('%d')) + "_updated"
  return updated

df.withColumn('updated_email_address', update_email(df.email_address))

但结果显示“已更新”

_

电子邮件

_

address的列为null：

+---------------+--------------+---------------------+
|account_id     |email_address |updated_email_address|
+---------------+--------------+---------------------+
|123456gd7tuhha |abc@test.com  |null           |
|djasevneuagsj1 |cde@test.com  |null           |
+---------------+--------------+---------------+

在函数内部更新

_

电子邮件‘它打印出来：

Column

它还显示了df的列数据类型，如下所示：

dfData:pyspark.sql.dataframe.DataFrame
account_id:string
email_address:string
updated_email_address:double

为什么会更新

_

电子邮件

_

double?的地址列类型？

Answer 1

您正在使用以下命令调用Python函数

类型。您必须从以下位置创建udf

然后使用它：

update_email_udf = udf(update_email)

然而，我建议你不要使用UDF进行这样的转换，你可以只使用Spark内置的函数(UDF以糟糕的性能而闻名)：

df.withColumn('updated_email_address',
              concat(substring(col("email_address"), -8, 8), date_format(current_date(), "ddMM"), lit("_updated"))
             ).show()

你可以找到

这里

所有Spark SQL内置函数。

Answer 2

多亏了你，我重新学习了一些我在spark课上忘记的东西

您不能使用WithColumn直接调用您的自定义函数，您需要使用UserDefinedFunctions (UDF)

下面是一个快速示例，展示了我如何获得一个自定义函数来处理您的数据帧(StringType是函数的返回类型)

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def update_email(email):
  return email+"aaaa"
#df.dtypes

my_udf = udf(lambda x: update_email(x), StringType())

df.withColumn('updated_email_address', my_udf(df.email_address) ).show()