PySpark向量化UDF空值判断问题：如何正确检测email列是否为空？

你的问题出在Pandas UDF的参数类型判断错误：你传入的email是Pandas Series对象，不是单个标量值，用if email is not None只会检查整个Series是否为None，完全不会检测Series内部的空值（比如Spark空值对应的Pandas NaN/pd.NA），所以即使email列有大量空值，这个条件永远为True，else分支根本不会触发。

正确实现方式（推荐矢量化操作） #

Pandas UDF的核心是矢量化处理，直接用Pandas的内置方法处理整列，效率远高于逐元素判断：

from pyspark.sql.functions import pandas_udf, StringType
import pandas as pd
import numpy as np

@pandas_udf(returnType=StringType())
def test(email, headers):
    # 从headers列的字典中提取"default"值，生成新的Series
    default_values = headers.str.get("default")
    # 用numpy.where做矢量化判断：email非空则取email，否则取default值
    return pd.Series(np.where(email.notna(), email, default_values))

或者用Pandas的fillna方法更简洁：

@pandas_udf(returnType=StringType())
def test(email, headers):
    default_values = headers.str.get("default")
    # 填充email的空值为对应的default值
    return email.fillna(default_values)

逐元素处理方式（不推荐，效率低） #

如果一定要逐行判断，需要遍历Series的每个元素，用pd.notna()检测空值：

@pandas_udf(returnType=StringType())
def test(email, headers):
    default_values = headers.str.get("default")
    # 逐行配对email和default值，判断空值后返回对应结果
    return pd.Series([e if pd.notna(e) else d for e, d in zip(email, default_values)])

额外注意 #

你调用UDF的代码少了一个右括号，修正后：

res = df.withColumn("out", test(col("email"), col("headers")))

内容的提问来源于stack exchange，提问作者Andrea Campolonghi