python - Hadoop 流 : Mapper 'wrapping' a binary executable

Question

我有一个目前在大型大学计算机集群上运行的管道。出于发布目的，我想将其转换为 mapreduce 格式，以便任何人都可以使用 hadoop 集群(例如 amazon webservices (AWS))运行它。该管道目前由一系列 python 脚本组成，这些脚本包装了不同的二进制可执行文件并使用 python 子进程和临时文件模块管理输入和输出。不幸的是，我没有编写二进制可执行文件，其中许多文件要么不采用 STDIN，要么不以“可用”方式发出 STDOUT(例如，仅将其发送到文件)。这些问题就是为什么我用 python 包装了大部分问题。

到目前为止，我已经能够修改我的 Python 代码，这样我就有了一个映射器和一个缩减器，可以在我的本地机器上以标准的“测试格式”运行。

$ cat data.txt | mapper.py | reducer.py

映射器按照它包装的二进制文件所需的方式格式化每一行数据，使用 subprocess.popen 将文本发送到二进制文件(这也允许我屏蔽很多虚假的 STDOUT)，然后收集我想要的 STOUT，并将其格式化为适合 reducer 的文本行。 当我尝试在本地 hadoop 安装上复制命令时出现问题。我可以让映射器执行，但它给出了一个错误，表明它找不到二进制可执行文件。

File "/Users/me/Desktop/hadoop-0.21.0/./phyml.py", line 69, in main() File "/Users/me/Desktop/hadoop-0.21.0/./mapper.py", line 66, in main phyml(None) File "/Users/me/Desktop/hadoop-0.21.0/./mapper.py", line 46, in phyml ft = Popen(cli_parts, stdin=PIPE, stderr=PIPE, stdout=PIPE) File "/Library/Frameworks/Python.framework/Versions/6.1/lib/python2.6/subprocess.py", line 621, in init errread, errwrite) File "/Library/Frameworks/Python.framework/Versions/6.1/lib/python2.6/subprocess.py", line 1126, in _execute_child raise child_exception OSError: [Errno 13] Permission denied

我的 hadoop 命令如下所示:

./bin/hadoop jar /Users/me/Desktop/hadoop-0.21.0/mapred/contrib/streaming/hadoop-0.21.0-streaming.jar \
-input /Users/me/Desktop/Code/AWS/temp/data.txt \
-output /Users/me/Desktop/aws_test \
-mapper  mapper.py \
-reducer  reducer.py \
-file /Users/me/Desktop/Code/AWS/temp/mapper.py \
-file /Users/me/Desktop/Code/AWS/temp/reducer.py \
-file /Users/me/Desktop/Code/AWS/temp/binary

正如我上面提到的，在我看来映射器并不知道二进制文件——也许它没有被发送到计算节点？不幸的是，我真的不知道问题出在哪里。任何帮助将不胜感激。如果能看到一些用 python 编写的包装二进制可执行文件的 hadoop 流式映射器/缩减器，那就太好了。我无法想象我是第一个尝试这样做的人!事实上，这里有另一篇帖子基本上问了同样的问题，但还没有得到回答......

Hadoop/Elastic Map Reduce with binary executable?

Answer 1

经过多次谷歌搜索(等)后，我想出了如何包含您的映射器/缩减器可以访问的可执行二进制文件/脚本/模块。诀窍是先将所有文件上传到 hadoop。

$ bin/hadoop dfs -copyFromLocal /local/file/system/module.py module.py

然后你需要像下面的模板一样格式化你的流命令:

$ ./bin/hadoop jar /local/file/system/hadoop-0.21.0/mapred/contrib/streaming/hadoop-0.21.0-streaming.jar \
-file /local/file/system/data/data.txt \
-file /local/file/system/mapper.py \
-file /local/file/system/reducer.py \
-cacheFile hdfs://localhost:9000/user/you/module.py#module.py \
-input data.txt \
-output output/ \
-mapper mapper.py \
-reducer reducer.py \
-verbose

如果您要链接 python 模块，则需要将以下代码添加到您的映射器/缩减器脚本中:

import sys 
sys.path.append('.')
import module

如果您通过子处理访问二进制文件，您的命令应该如下所示:

cli = "./binary %s" % (argument)
cli_parts = shlex.split(cli)
mp = Popen(cli_parts, stdin=PIPE, stderr=PIPE, stdout=PIPE)
mp.communicate()[0]

希望这对您有所帮助。