我有一个大规模的日志处理问题，我必须在 hadoop 集群上运行。任务是将日志的每一行输入一个可执行的“cmd”，并检查结果以决定是否保留这行日志。

由于“cmd”程序打开了一个非常大的字典，我无法为日志的每一行调用该程序。我想让它保持运行并向它提供所需的输入。我当前的解决方案使用 python 的子进程模块，代码如下:

import sys
from subprocess import Popen, PIPE

def main():
    pp = Popen('./bqc/bqc/bqc_tool ./bqc/bqc/bqc_dict/ ./bqc/bqc/word_dict/ flag', shell=True, stdin=PIPE, stdout=PIPE, stderr=PIPE)

    for line in sys.stdin:
        lstr = line.strip().split('\t')
        if len(lstr) != 7:
            continue
        pp.stdin.write('%s\n' % lstr[5])
        pp.stdin.flush()
        out = pp.stdout.readline()
        lout = out.strip().split('\t')
        if len(lout) == 3 and lout[1] == '401':
            print line.strip()

if __name__ == '__main__':
    main()

上面的代码在我的本地机器上测试时可以找到。 将作业提交到 hadoop 时用作映射器。我没有使用reducer，配置如下。

hadoop streaming \
-input /path_to_input \
-output /path_to_output \
-mapper  "python/python2.7/bin/python27.sh ./mapper.py" \
-cacheArchive /path_to_python/python272.tar.gz#python \
-cacheArchive /path_to_cmd/bqc.tar.gz#bqc \
-file ./mapper.py \
-jobconf mapred.job.name="JobName" \
-jobconf mapred.job.priority=HIGH

bqc.tar.gz 中的文件如下所示:

bqc/
bqc/bqc_tool
bqc/bqc_dict/
bqc/word_dict/

在我看来，“-cacheArchive/path_to_cmd/bqc.tar.gz#bqc\”行应该提取 tar 文件并将它们提取到名为 bqc 的文件夹中。

但是提交到hadoop集群时失败，报错如下:

    Traceback (most recent call last):
      File "./mapper.py", line 19, in
        main()
      File "./mapper.py", line 11, in main
        pp.stdin.write('%s\n' % lstr[5])
    IOError: [Errno 32] Broken pipe
    java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1
        at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:335)
        at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:590)
        at org.apache.hadoop.streaming.PipeMapper.map(PipeMapper.java:152)
        at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:54)
        at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:18)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:388)
        at org.apache.hadoop.mapred.Child.main(Child.java:194)
    java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1
        at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:335)
        at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:590)
        at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:163)
        at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:61)
        at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:18)
        at org.apache.hadoop.mapred.MapTask.run(MapTask.java:388)
        at org.apache.hadoop.mapred.Child.main(Child.java:194)

有人知道吗？任何帮助将不胜感激!

谢谢!

扎卡里

最佳答案

谜底揭晓!应该是hadoop的内存限制导致命令加载不成功。该命令需要大约 2G 内存，hadoop 配置为每个节点允许大约 800MB。

关于python - Hadoop 流中损坏的 python 管道，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/11344422/