使用spark.sparkContext.addPyFile导入Pandas UDF报_jvm属性缺失错误

该报错是因为你在toy_example.py模块层直接使用@f.pandas_udf装饰器，当Spark通过addPyFile将文件分发到工作节点并导入模块时，装饰器会在模块加载阶段就尝试调用Spark的JVM实例，而此时工作节点上的SparkContext还未完成初始化，因此拿到None对象触发属性报错。

以下两种常用方案都可以解决该问题： #

方案1：外部文件仅保留核心逻辑，导入时手动注册UDF #

该方案逻辑最直观，适配绝大多数场景

• 首先修改helpful_pandas_udfs/toy_example.py内容：

import pandas as pd

# 仅保留原始业务逻辑，不提前加pandas_udf装饰器
def multiply_func(a: pd.Series, b: pd.Series) -> pd.Series:
    return a * b

def udf_multiply_logic(a: pd.Series, b: pd.Series) -> pd.Series:
    df = pd.DataFrame({'a': a, 'b': b})
    df['product'] = df.apply(lambda x : multiply_func(x['a'], x['b']), axis = 1)
    return df['product']

• 然后在Jupyter Notebook中导入并注册UDF：

import pandas as pd
from pyspark.sql import functions as f

spark.sparkContext.addPyFile("helpful_pandas_udfs/toy_example.py")
from toy_example import udf_multiply_logic

# 在当前有活跃SparkSession的上下文中手动注册为pandas UDF
udf_multiply = f.pandas_udf(udf_multiply_logic, returnType="float")

# 后续正常使用即可
df = spark.createDataFrame(pd.DataFrame(pd.Series([1,2,3]), columns=["x"]))
df.select(udf_multiply(f.col("x"), f.col("x"))).show()

方案2：外部文件用工厂函数封装UDF创建逻辑 #

如果不想在Notebook中写注册逻辑，可以把UDF生成步骤封装到函数中，只有调用时才会执行装饰器逻辑

• 修改toy_example.py内容：

import pandas as pd
from pyspark.sql import functions as f

def multiply_func(a: pd.Series, b: pd.Series) -> pd.Series:
    return a * b

def get_udf_multiply():
    @f.pandas_udf("float")
    def udf_multiply(a: pd.Series, b: pd.Series) -> pd.Series:
        df = pd.DataFrame({'a': a, 'b': b})
        df['product'] = df.apply(lambda x : multiply_func(x['a'], x['b']), axis = 1)
        return df['product']
    return udf_multiply

• 在Notebook中调用：

spark.sparkContext.addPyFile("helpful_pandas_udfs/toy_example.py")
from toy_example import get_udf_multiply

udf_multiply = get_udf_multiply()
# 后续正常使用即可

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