PySpark 2.2.0自定义UDF执行报错：找不到cast_to_float模块

这个问题的核心原因其实是PySpark的分布式执行机制：你在tests/test.py里给Driver进程的sys.path添加了父目录，但Worker节点（哪怕是本地模式下的Worker进程）的Python环境并没有同步这个路径。当自定义UDF要在Worker上执行时，它找不到你的cast_to_float模块，就会抛出Py4JJavaError。

下面给你几个针对性的解决方案，按推荐优先级排序：

方案1：用sparkContext.addPyFile()分发模块文件 #

这是最直接适配PySpark执行逻辑的方法，它会把指定的Python文件/包分发到所有Worker节点，并自动添加到它们的PYTHONPATH中。

修改你的tests/test.py代码，在创建SparkSession后加上这一步：

import sys
import os
from pyspark.sql import SparkSession

# 获取根目录绝对路径
root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(root_dir)

import cast_to_float

# 初始化SparkSession
spark = SparkSession.builder.appName("TestCastUDF").getOrCreate()

# 关键：把你的模块文件分发到所有Worker节点
spark.sparkContext.addPyFile(os.path.join(root_dir, "cast_to_float.py"))

# 后续创建DataFrame、调用函数的逻辑保持不变
data = [("1.23",), ("4.56",)]
df = spark.createDataFrame(data, ["value_str"])

df1 = cast_to_float.cast_with_cast_method(df, "value_str")
df1.show()
df1.printSchema()

df2 = cast_to_float.cast_with_udf(df, "value_str")
df2.show()
df2.printSchema()

这样不管你在根目录执行python tests/test.py，还是用其他方式运行测试，Worker节点都能获取到cast_to_float.py文件，UDF就能正常执行了。

方案2：打包成Python包并安装（适合长期维护的项目） #

如果你的代码会长期迭代，建议把根目录做成一个可安装的Python包：

1. 在根目录下创建setup.py或pyproject.toml文件，定义包信息（比如包名、版本、依赖等）
2. 在Driver和所有Worker节点上安装这个包：pip install -e .（本地开发用 editable 模式）

这样不管是Driver还是Worker，都能直接通过import cast_to_float导入模块，不需要额外处理路径。

方案3：打包成Zip包分发（多模块场景） #

如果你的项目有多个模块文件，可以把整个根目录打包成Zip包，再通过addPyFile分发：

1. 在根目录执行打包命令：zip -r cast_utils.zip cast_to_float.py（如果有其他模块也可以加进去）
2. 在test.py里修改分发逻辑：

spark.sparkContext.addPyFile(os.path.join(root_dir, "cast_utils.zip"))

Worker节点会自动把Zip包添加到PYTHONPATH，里面的模块就能正常导入。

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