PySpark自定义UDF报ModuleNotFoundError问题求助

针对你遇到的ModuleNotFoundError，结合Python 3.8、Spark 3.2的环境，可尝试以下几种针对性解决方案：

1. 确认压缩包目录结构正确性 #

压缩jobs文件夹时，必须保证压缩包根目录直接是jobs，而非包含jobs的父文件夹。执行压缩命令时要注意路径：

# 进入jobs所在的父目录，执行以下命令
zip -r jobs.zip jobs/

解压后检查结构，确保jobs下的子模块（lib_a、udf、scripts）都存在对应的__init__.py文件（空文件即可），Python需要这个文件识别子目录为可导入模块。

2. 全局配置Python路径（驱动+执行器） #

在Spark任务提交时，除了通过--py-files指定压缩包，还要显式配置驱动和执行器的Python路径，避免执行器端路径缺失：

spark-submit \
  --master k8s://https://<你的K8s API地址> \
  --deploy-mode cluster \
  --py-files jobs.zip \
  --conf spark.executorEnv.PYTHONPATH=./jobs.zip \
  --conf spark.driverEnv.PYTHONPATH=./jobs.zip \
  jobs/scripts/test_script.py

3. 在脚本内主动添加PyFile #

集群模式下，有时提交时指定--py-files的优先级不如脚本内主动添加可靠，可在初始化SparkSession后添加：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("UDFTest").getOrCreate()
# 主动将压缩包添加到Spark上下文
spark.sparkContext.addPyFile("jobs.zip")

# 此时再导入UDF模块
from jobs.udf import udf_function

4. 更换UDF序列化器 #

PySpark默认用pickle序列化UDF，部分场景下会导致模块导入异常，可切换为cloudpickle序列化（Spark 3.0+支持）：

方式一：全局配置 #

spark-submit \
  # 其他参数...
  --conf spark.sql.execution.pythonUDF.serializer=cloudpickle \
  jobs/scripts/test_script.py

方式二：定义UDF时指定 #

from pyspark.sql.functions import udf
import cloudpickle

from jobs.udf import udf_function
# 指定cloudpickle作为序列化器
my_udf = udf(udf_function, returnType="string", serializer=cloudpickle)

5. 验证执行器端路径加载 #

如果以上方法无效，可在脚本中添加调试代码，打印执行器端的Python路径，确认jobs.zip是否被正确加载：

import sys
from pyspark.sql.functions import udf

def debug_path():
    return str(sys.path)

debug_udf = udf(debug_path, returnType="string")
# 执行一个简单的DataFrame操作，查看执行器的路径
spark.range(1).select(debug_udf()).show(truncate=False)

通过输出结果，确认jobs.zip是否在sys.path列表中，若不在则需调整路径配置。

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