如何导入我通过 --py-files 添加到集群的 python 模块？答案

【问题标题】：How to import a python module that I added to a cluster via --py-files?如何导入我通过 --py-files 添加到集群的 python 模块？
【发布时间】：2020-02-18 01:46:47
【问题描述】：

我有一些想要在应用程序中使用的自定义 jdbc 驱动程序。当我向 Kubernetes spark 集群提交 spark 时，我将它们包含为 --py-files：

spark-submit --py-files s3a://bucket/pyfiles/pyspark_jdbc.zip my_application.py

这给了我：

java.io.FileNotFoundException: File file:/opt/spark/work-dir/pyspark_jdbc.zip does not exist

正如other answers 告诉我的那样，我实际上需要将该 zip 文件添加到 PYTHONPATH。现在，我发现至少 Spark 2.3+ 不再适用，但让我们这样做：

spark.sparkContext.addPyFile("pyspark_jdbc.zip")

查看集群日志，我看到了：

19/10/21 22:40:56 INFO Utils: Fetching s3a://bucket/pyfiles/pyspark_jdbc.zip to 
/var/data/spark-52e390f5-85f4-41c4-9957-ff79f1433f64/spark-402e0a00-6806-40a7-a17d-5adf39a5c2d4/userFiles-680c1bce-ad5f-4a0b-9160-2c3037eefc29/fetchFileTemp5609787392859819321.tmp

所以，pyfiles 确实被导入了，但是导入到/var/data/... 而不是我的工作目录。因此，当我将 .zip 文件的位置添加到我的 python 路径时，我不知道它在哪里。在尝试添加 python 文件之前对集群进行一些诊断：

> print(sys.path)
[..., 
 '/var/data/spark-52e390f5-85f4-41c4-9957-ff79f1433f64/spark-402e0a00-6806-40a7-a17d-5adf39a5c2d4/userFiles-680c1bce-ad5f-4a0b-9160-2c3037eefc29', 
 '/opt/spark/work-dir/s3a', 
 '//bucket/pyfiles/pyspark_jdbc.zip'
...]
> print(os.getcwd())
/opt/spark/work-dir
> subprocess.run(["ls", "-l"])
total 0

所以我们看到 pyspark 确实尝试将我通过 --py-files 添加的 s3a:// 文件添加到 PYTHONPATH，但它错误地解释了 : 并且没有正确添加路径。 /var/data/... 目录位于 PYTHONPATH 中，但特定的 .zip 文件不在，因此我无法从中导入。

我该如何解决这个问题？ .zip 文件未正确添加到路径中，但在我的程序中，我也不知道

一个。 pyspark 尝试添加到 s3a:// 的路径蟒蛇路径

b. .zip 文件的 `var/data/.../ 本地位置的路径。我知道它在某处的路径中，我想我可以将其解析出来，但这会很混乱。

什么是优雅的解决方案？

【问题讨论】：

标签： apache-spark import pyspark python-import

【解决方案1】：

一个（更好的）解决方案是使用 pyspark 中的 SparkFiles 对象来定位您的导入。

from pyspark import SparkFiles

spark.sparkContext.addPyFile(SparkFiles.get("pyspark_jdbc.zp"))

【讨论】：

很好的答案！.. 我只是在寻找这个。

【解决方案2】：

一个（不好的）解决方案是简单地解析出看起来可能包含 .zip 文件的路径，并将它们添加到 sys.path。

for pth in [p for p in sys.path if p.startswith("/var/data/spark-")]:
    try:
        sys.path.append("{}/pyspark_jdbc.zip".format(pth))
    except:
        passed

此解决方案有效，允许我们测试实际的 spark 应用程序，但我不认为这是一个生产就绪的解决方案。

【讨论】：