python - 为什么我的 udf 调用看不到使用 SparkContext.addPyFile 添加的依赖项？

Question

我正在从 Jupyter 控制台执行一些简单的 PySpark，当我尝试调用一些外部代码时遇到问题。我的最小示例代码有一个依赖项，testpackage.zip。当我调用使用该包中的代码的 UserDefinedFunction 时，我收到一个 AttributeError ，显示 Spark 无法在 testpackage.zip 中找到该函数。

testpackage 是一个包含空的 __init__.py 和 testmod.py 的目录，其中包含

def testfunc(x):
    return float(x)+1.33

最小示例代码，我在 Jupyter Console 中一次运行一个 block :

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import UserDefinedFunction as udf
from pyspark.sql.types import DoubleType

sess = SparkSession.builder.appName("testing").getOrCreate()
sc = sess.sparkContext

DEP_PATH = < path on driver >.testpackage.zip
sc.addPyFile(DEP_PATH)
import testpackage

df = sess.range(0, 12*1000*1000, numPartitions=12)

test_udf = udf(lambda x: testpackage.testmod.testfunc(x), DoubleType())

df = df.withColumn("udf_res", test_udf(df.id))

df.head(5) # error

请注意，<驱动程序上的路径>(依赖项所在的目录)位于我的 PYTHONPATH 中。

错误消息部分读取:

"AttributeError: module 'testpackage' has no attribute 'testmod'".

我不确定问题是否与我在 udf 中调用 testfunc 有关，或者是否只是我没有正确添加依赖项。以编程方式添加 Python 依赖项(不使用 Spark-submit)的最佳方法是什么？

Answer 1

您需要通过以下方式导入模块。您可以尝试一下吗 -

从 testpackage.testmod 导入 *

test_udf = udf(lambda x: testfunc(x), DoubleType())