我的目标是读取 hdfs 上由 csv 数据组成的二进制(gpg 加密)文件。我的方法——遵循this answer -- 定义一个 Python 函数来读取和解密 gpg 文件,生成每一行,并将该函数作为 flatMap
应用于并行文件列表。
本质上,Python 函数会生成一个子进程,该子进程使用 hadoop
读取文件,并将结果通过管道传输到 gpg
进行解密。在本地模式下运行 Spark 时,这工作得很好。然而,分布式运行(yarn-client
),简单的行计数会返回 0
,主要是因为 Python 认为 stdout
管道始终处于关闭状态。
问题似乎是子进程涉及两个命令之间的管道。当我删除后者(只是加密文件的行数)时,行数与我在命令行上得到的内容相匹配。我尝试了多种方法,但结果都相同。
这是 Python 函数:
import subprocess as sp
def read_gpg_file_on_hdfs(filename):
# Method 1:
p = sp.Popen('hadoop fs -cat {} | gpg -d'.format(filename), shell=True,
stdout=sp.PIPE)
# Method 2:
p1 = sp.Popen(['hadoop', 'fs', '-cat', filename], stdout=sp.PIPE)
p = sp.Popen(['gpg', '-d'], stdin=p1.stdout, stdout=sp.PIPE)
p1.stdout.close()
# Method 3:
p = sp.Ppen('gpg -d <(hadoop fs -cat {})'.format(filename), shell=True,
stdout=sp.PIPE, stderr=sp.PIPE)
for line in p.stdout:
yield line.strip()
这是 Spark 命令:
sc.parallelize(['/path/to/file.gpg']).flatMap(read_gpg_file_on_hdfs).count()
现在我知道 PySpark 使用管道与 Spark 进行通信,但我不了解细节,我不知道这是否会影响我想要做的事情。我的问题是是否有办法完成我想做的事情。
请注意,我使用的是 Spark 1.2.1 分布式版本(MapR 的最新版本)。另外,我考虑过使用 binaryFiles
,但这对于大型 gpg 文件会失败,我有时会遇到这种情况。
提前致谢!
最佳答案
事实证明,gpg
命令实际上是问题所在。据推测,它与如何在本地模式与分布式模式下启动子进程的细节有关,但在本地模式下,gpg
的 homedir
设置正确。但是,当以分布式模式启动时,homedir
指向错误的目录,第二个子进程立即失败。此错误消息似乎没有记录在任何地方,因此 stdout
只是作为空字符串返回。
关于python - 涉及带有管道的子进程的映射步骤在 PySpark 中失败,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/30878500/