来自 pyspark worker 的 HDFS / Hadoop api 访问答案

【问题标题】：HDFS / Hadoop api access from pyspark worker来自 pyspark worker 的 HDFS / Hadoop api 访问
【发布时间】：2015-05-11 01:52:02
【问题描述】：

我需要从 pyspark worker 中读取/扫描/写入文件到 hdfs。

请注意，以下 api 不适用，因为它们会脱离 驱动程序：

sc.textFile()
sc.saveAsParquetFile()

等

最好不要涉及额外的第三方库（例如 pyhadoop）。

一种选择是掏空例如

 os.system('hdfs dfs -ls %(hdfsPath)s' %locals())

但是有没有更原生的 pyspark 方式来实现这一点？

UPDATE 这不是广播数据的情况，因为每个工作人员将从 hdfs 读取不同的数据。其中一个用例是在每个工作人员中读取一些大型二进制文件（这显然不是广播的情况）。另一种情况是读取包含指令的“命令”文件。我已经在原生 hadoop 和 scala spark 中成功使用了这种模式。

【问题讨论】：

标签： apache-spark hdfs pyspark

【解决方案1】：

解决方案似乎是 subprocess out（没有直接的 python 访问）。将接受的答案和来自以下位置的 cmets 之一拼凑在一起：Python read file as stream from HDFS

cat = subprocess.Popen(["hadoop", "fs", "-cat", "/path/to/myfile"], stdout=subprocess.PIPE)
for line in iter(cat.stdout.readline, ''): 
    print line,   # include the comma

【讨论】：

【解决方案2】：

更本地的 PySpark 方法是使用 sc.textFile() 或其他读取方法读取驱动程序中的数据，并将其作为 RDD 或广播变量传递给工作人员，如果它足够小以适合内存每个执行者。

你能描述一下你的情况吗，我怀疑你真的需要阅读工人中的文件

更新：

简短总结：

直接从大型集群上的工作程序读取文件集可能会杀死名称节点
在大多数情况下，不需要直接从工作人员那里读取单独的文件。您可以为 textFile() 方法使用通配符，或使用 wholeTextFiles() 或 binaryFiles() 方法读取文件集及其名称
在处理具有千兆字节图像的特定情况下，只需将它们放入序列文件并使用sequenceFile() 方法读取它
可以通过直接查询 WebHDFS REST API 来实现使用 Python 直接从 HSFS 读取而无需额外的库，考虑到这正是库所实现的，这有点过头了。另一种选择可能是使用pipe() Spark 方法调用Java 程序读取HDFS 文件并将它们以序列化形式返回到标准输出。另一种选择是通过转义到 shell 将文件从 HDFS 复制到临时空间，然后使用标准读取文件功能读取该文件。就我个人而言，如果我的开发人员实施了我在这里提出的任何方法，我会解雇他们

【讨论】：

更新 OP：每个工人都有不同的数据，所以广播是不合适的。
那是怎么回事呢？为什么不使用带有通配符的textFile() 或wholeTextFiles()？
是的，它不是为此而设计的。这就是为什么我告诉你应该在驱动程序中使用textFile() 或wholeTextFiles()。请澄清您需要此功能的情况，到目前为止，您似乎想弄乱 Spark 概念
对我来说也一样 - 两者都经过认证并与两者合作。从大型集群上运行在 yarn 下的工作人员读取一系列文件会杀死你的 namenode
读几个文件不会。这些文件是每个工人价值 GB 的图像文件。祝你好运！