pyspark 中的 --files 选项不起作用

Posted 2023-04-15

【中文标题】pyspark 中的 --files 选项不起作用

【英文标题】：--files option in pyspark not working

【发布时间】：2017-11-08 18:54:42

【问题描述】：

我从命令行尝试了sc.addFile 选项（没有任何问题）和--files 选项（失败）。

运行 1：spark_distro.py

from pyspark import SparkContext, SparkConf
from pyspark import SparkFiles

def import_my_special_package(x):
    from external_package import external
    ext = external()
    return ext.fun(x)

conf = SparkConf().setAppName("Using External Library")
sc = SparkContext(conf=conf)
sc.addFile("/local-path/readme.txt")
with open(SparkFiles.get('readme.txt')) as test_file:
    lines = [line.strip() for line in test_file]
print(lines)
int_rdd = sc.parallelize([1, 2, 4, 3])
mod_rdd = sorted(int_rdd.filter(lambda z: z%2 == 1).map(lambda x:import_my_special_package(x)))

外部包：external_package.py

class external(object):
    def __init__(self):
        pass
    def fun(self,input):
        return input*2

readme.txt

MY TEXT HERE

spark-submit 命令

spark-submit \
  --master yarn-client \
  --py-files /path to local codelib/external_package.py  \
  /local-pgm-path/spark_distro.py  \
  1000

输出：按预期工作

['MY TEXT HERE']

但是，如果我尝试使用 --files（而不是 sc.addFile）选项从命令行传递文件（readme.txt），则会失败。 如下所示。

运行 2：spark_distro.py

from pyspark import SparkContext, SparkConf
from pyspark import SparkFiles

def import_my_special_package(x):
    from external_package import external
    ext = external()
    return ext.fun(x)

conf = SparkConf().setAppName("Using External Library")
sc = SparkContext(conf=conf)
with open(SparkFiles.get('readme.txt')) as test_file:
    lines = [line.strip() for line in test_file]
print(lines)
int_rdd = sc.parallelize([1, 2, 4, 3])
mod_rdd = sorted(int_rdd.filter(lambda z: z%2 == 1).map(lambda x: import_my_special_package(x)))

external_package.py同上

火花提交

spark-submit \
  --master yarn-client \
  --py-files /path to local codelib/external_package.py  \
  --files /local-path/readme.txt#readme.txt  \
  /local-pgm-path/spark_distro.py  \
  1000

输出：

Traceback (most recent call last):
  File "/local-pgm-path/spark_distro.py", line 31, in <module>
    with open(SparkFiles.get('readme.txt')) as test_file:
IOError: [Errno 2] No such file or directory: u'/tmp/spark-42dff0d7-c52f-46a8-8323-08bccb412cd6/userFiles-8bd16297-1291-4a37-b080-bbc3836cb512/readme.txt'

sc.addFile 和 --file 是否用于相同目的？有人可以分享您的想法吗？

【问题讨论】：

只是出于好奇，你为什么一直在spark-submit 中包含1000？没错，它在示例中使用，但只是因为它确实是SparkPi.scala 和pi.py 的预期参数，而您的spark_distro.py 不是这种情况...

@desertnaut 上面的错误消息说它正在驱动程序路径“u”/tmp/spark-42dff0d7-c52f-46a8-8323-08bccb412cd6/userFiles-8bd16297-1291 中寻找“readme.txt”文件-4a37-b080-bbc3836cb512/readme.txt'" 并且在日志中我的本地文件（'readme.txt'）被复制到驱动程序节点位置（使用'--files'选项）。上面的 spark-submit 命令正在使用“yarn-client”。我认为把它改成“yarn-cluster”有什么不同吗？

没有（已经检查过）

我开始怀疑这两种方法 (sc.addFile & --files) 不等效 - 无论如何，我没有设法使 --files 工作，也找不到任何参考从过去的某人在线...

@desertnaut 你遇到了什么问题--files???

【参考方案1】：

我终于弄清楚了这个问题，而且确实是一个非常微妙的问题。

正如怀疑的那样，这两个选项（sc.addFile 和 --files）不等效，并且（诚然非常巧妙地）暗示了文档（添加了重点）：

addFile（路径，递归=False） 在每个节点上添加一个要与此 Spark 作业一起下载的文件。

--files 文件 要放置在工作中的文件的逗号分隔列表 每个执行者的目录。

简而言之，虽然添加了sc.addFile的文件对执行者和驱动程序都可用，但添加了--files的文件仅对执行者可用；因此，当尝试从驱动程序访问它们时（就像 OP 中的情况一样），我们会收到 No such file or directory 错误。

让我们确认一下（去掉 OP 中所有不相关的 --py-files 和 1000 内容）：

test_fail.py：

from pyspark import SparkContext, SparkConf
from pyspark import SparkFiles

conf = SparkConf().setAppName("Use External File")
sc = SparkContext(conf=conf)
with open(SparkFiles.get('readme.txt')) as test_file:  
    lines = [line.strip() for line in test_file]
print(lines)

测试：

spark-submit --master yarn \
             --deploy-mode client \
             --files /home/ctsats/readme.txt \
             /home/ctsats/scripts/SO/test_fail.py

结果：

[...]
17/11/10 15:05:39 INFO yarn.Client: Uploading resource file:/home/ctsats/readme.txt -> hdfs://host-hd-01.corp.nodalpoint.com:8020/user/ctsats/.sparkStaging/application_1507295423401_0047/readme.txt
[...]
Traceback (most recent call last):
  File "/home/ctsats/scripts/SO/test_fail.py", line 6, in <module>
    with open(SparkFiles.get('readme.txt')) as test_file:
IOError: [Errno 2] No such file or directory: u'/tmp/spark-8715b4d9-a23b-4002-a1f0-63a1e9d3e00e/userFiles-60053a41-472e-4844-a587-6d10ed769e1a/readme.txt'

在上述脚本test_fail.py中，是驱动程序程序请求访问文件readme.txt；让我们更改脚本，以便为 executors (test_success.py) 请求访问权限：

from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName("Use External File")
sc = SparkContext(conf=conf)

lines = sc.textFile("readme.txt") # run in the executors
print(lines.collect())

测试：

spark-submit --master yarn \
             --deploy-mode client \
             --files /home/ctsats/readme.txt \
             /home/ctsats/scripts/SO/test_success.py

结果：

[...]
17/11/10 15:16:05 INFO yarn.Client: Uploading resource file:/home/ctsats/readme.txt -> hdfs://host-hd-01.corp.nodalpoint.com:8020/user/ctsats/.sparkStaging/application_1507295423401_0049/readme.txt
[...]
[u'MY TEXT HERE']

另请注意，这里我们不需要SparkFiles.get - 该文件很容易访问。

如上所述，sc.addFile 在这两种情况下都可以使用，即当驱动程序或执行程序请求访问时（经过测试但未在此处显示）。

关于命令行选项的顺序：正如我所说的elsewhere，所有与Spark相关的参数都必须在要执行的脚本之前；可以说，--files 和 --py-files 的相对顺序无关紧要（留作练习）。

使用 Spark 1.6.0 和 2.2.0 测试。

更新（在 cmets 之后）：似乎我的 fs.defaultFS 设置也指向 HDFS：

$ hdfs getconf -confKey fs.defaultFS
hdfs://host-hd-01.corp.nodalpoint.com:8020

但让我在这里关注森林（而不是树木），并解释为什么整个讨论仅具有学术兴趣：

使用--files 标志传递待处理文件是不好的做法；事后看来，我现在明白了为什么我在网上几乎找不到任何使用参考——可能没有人在实践中使用它，而且有充分的理由。

（请注意，我说的不是--py-files，它具有不同的合法角色。）

由于 Spark 是一个分布式处理框架，运行在一个集群和一个分布式文件系统 (HDFS) 上，最好的办法是让所有要处理的文件都已经进入 HDFS - 期间 . Spark 处理文件的“自然”位置是 HDFS，而不是本地 FS - 尽管有一些 toy 示例使用本地 FS 仅用于演示目的。更重要的是，如果您希望将来有时间将部署模式更改为cluster，您会发现默认情况下，集群对本地路径和文件一无所知，这是理所当然的......

【讨论】：

超级棒:)

--files /home/ctsats/readme.txt 正在将文件复制到暂存目录 hdfs://host-hd-01.corp.nodalpoint.com:8020/user/ctsats/.sparkStaging /application_1507295423401_0049/readme.txt 但 lines = sc.textFile("readme.txt") 期望文件位于 hdfs://host-hd-01.corp.nodalpoint.com:8020/user/ctsats/ 目录中。所以它不适合我

@goks 可能您的core-site.xml conf 文件中的fs.defaultFS 设置设置为HDFS，而我的必须是default value file:///。将在星期一检查它（现在无法访问） - 同时，请参阅mail-archives.us.apache.org/mod_mbox/spark-user/201402.mbox/…

是的。 “fs.defaultFS”指向 HDFS。我想如果我可以在 pyspark 脚本（spark_distro.py）中访问“fs.defaultFS”的值（需要找到一种方法），那么我可以访问该文件。 stg_path = str(fs.defaultFS) + "/user/" + str(os.environ['USER']) + "/.sparkStaging/" + str(sc.applicationId) + "/" lines = sc.textFile(os.path.join(stg_path,'readme.txt')) print(lines.collect())

@Fizi 如果您有新问题，请打开一个新问题（如果需要，您可以在此处链接）；自 2017 年以来，Spark 中的许多细节可能发生了变化，而 cmets 不适合此类后续问题。