Hadoop Streaming Job 在 python 中失败答案

【问题标题】：Hadoop Streaming Job failed in pythonHadoop Streaming Job 在 python 中失败
【发布时间】：2011-05-04 21:41:13
【问题描述】：

我有一个用 Python 编写的 mapreduce 作业。该程序在linux env中测试成功，但在Hadoop下运行失败。

这里是作业命令：

hadoop  jar $HADOOP_HOME/contrib/streaming/hadoop-0.20.1+169.127-streaming.jar \
   -input /data/omni/20110115/exp6-10122 -output /home/yan/visitorpy.out \
   -mapper SessionMap.py   -reducer  SessionRed.py  -file SessionMap.py \
   -file  SessionRed.py

Session*.py的模式是755，#!/usr/bin/env python是*.py文件的最上面一行。 Mapper.py 是：

#!/usr/bin/env python
import sys
 for line in sys.stdin:
         val=line.split("\t")
         (visidH,visidL,sessionID)=(val[4],val[5],val[108])
         print "%s%s\t%s" % (visidH,visidL,sessionID)

日志中的错误：

java.io.IOException: Broken pipe
    at java.io.FileOutputStream.writeBytes(Native Method)
    at java.io.FileOutputStream.write(FileOutputStream.java:260)
    at java.io.BufferedOutputStream.write(BufferedOutputStream.java:105)
    at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:65)
    at java.io.BufferedOutputStream.write(BufferedOutputStream.java:109)
    at java.io.DataOutputStream.write(DataOutputStream.java:90)
    at org.apache.hadoop.streaming.io.TextInputWriter.writeUTF8(TextInputWriter.java:72)
    at org.apache.hadoop.streaming.io.TextInputWriter.writeValue(TextInputWriter.java:51)
    at org.apache.hadoop.streaming.PipeMapper.map(PipeMapper.java:110)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:50)
    at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:36)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:358)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:307)
    at org.apache.hadoop.mapred.Child.main(Child.java:170)
    at org.apache.hadoop.streaming.PipeMapper.map(PipeMapper.java:126)
    at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:50)
    at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:36)
    at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:358)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:307)
    at org.apache.hadoop.mapred.Child.main(Child.java:170)

【问题讨论】：

标签： python hadoop mapreduce

【解决方案1】：

我遇到了同样的问题并且想知道，因为当我在测试数据上测试我的映射器和减速器时，它会运行。但是当我通过 hadoop map-reduce 运行相同的测试集时，我曾经遇到过同样的问题。

如何在本地测试您的代码：

cat <test file> | python mapper.py | sort | python reducer.py

经过进一步调查，我发现我的 mapper.py 脚本中没有包含“shebang line”。

#!/usr/bin/python

请将上面的行添加为您的python脚本的第一行，并在此之后留一个空行。

如果您需要了解更多关于'shebang line'的信息，请阅读Why do people write #!/usr/bin/env python on the first line of a Python script?

【讨论】：

【解决方案2】：

您可以在 hadoop Web 界面中找到 python 错误消息（例如回溯）和您的脚本写入 stderr 的其他内容。它有点隐藏，但您会在流媒体为您提供的链接中找到它。您单击“映射”或“减少”，然后单击任何任务，然后在“全部”列中的任务日志中

【讨论】：

【解决方案3】：

最后我修复了这个错误，这是我学到的教训。 1）原始代码没有对坏数据进行错误处理。当我在一个小数据集上测试代码时，我没有发现问题。 2）为了处理空字段/变量，我发现在 Python 中测试 None 和空字符串有点棘手。我个人喜欢函数 len(strVar) ，它易于阅读且有效。 3）这种情况下的hadoop命令是正确的。不知怎的，模式为644的*.py可以在我使用的环境中成功运行。

【讨论】：

如果您认为这回答了问题，您应该选择它作为答案。这将使可能面临类似问题的其他人更容易。

【解决方案4】：

Hadoop 流式传输 - Hadoop 1.0.x

我遇到了同样的“断管”问题。问题是我的减速器中的“中断”语句。所以，一切都很顺利，直到“休息”。之后，正在运行的减速器停止运行打印“Broken pipe”错误。此外，另一个减速器开始运行，与之前的命运相同。这个圈子一直在循环。

如果我理解正确，当 reducer 开始从 stdin 读取时（这是我的情况，在 for 循环中），那么它必须读取所有内容。即使你关闭了标准输入（os.close(0) 我试过），你也不能“破坏”这个操作。

【讨论】：

【解决方案5】：

我今天在使用 Hadoop 1.0.1 时遇到了同样的问题。幸运的是，我已经通过以下方式解决了它：

hadoop ... -mapper $cwd/mapper.py -reducer $cwd/reducer.py ...

（我的 Python 脚本在当前目录中）。现在看来绝对路径是必要的。

最好的！

【讨论】：

【解决方案6】：

脏输入可能会导致此问题。

尝试使用 try{} 来避免这种情况。

#!/usr/bin/env python
import sys
for line in sys.stdin:
    try:
         val=line.split("\t")
         (visidH,visidL,sessionID)=(val[4],val[5],val[108])
         print "%s%s\t%s" % (visidH,visidL,sessionID)
    except Exception  as e:
        pass

【讨论】：

【解决方案7】：

Python + Hadoop 在一些不应该的细节上很棘手。看看here。

尝试用双引号将您的输入路径括起来。（-输入“/data/omni/20110115/exp6-10122”）

【讨论】：

【解决方案8】：

一种可能的解决方案是包括“python”，即：

-mapper  "python ./mapper.py"
-reducer "python ./reducer.py"

【讨论】：