apache-spark - 带有函数的pySpark withColumn

Question

我有一个包含 2 列的数据框:account_id 和 email_address，现在我想再添加一列 updated_email_address，我称之为email_address 上的函数以获取 updated_email_address。这是我的代码:

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))

但结果显示 updated_email_address 列为空:

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

在函数updated_email里面打印出来:

Column<b'(email_address + == email to be udpated: )'>

它还将df的列数据类型显示为:

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

为什么 updated_email_address 列类型是 double 的？

Answer 1

您正在调用具有 Column 类型的 Python 函数。您必须从 update_email 创建 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()

您可以找到here所有 Spark SQL 内置函数。