我正在尝试使用Cloudera 5.5.0实现一个简单的Hadoop map reduce示例地图&reduce步骤应该使用Python 2.6.6 实现
问题：
- 如果脚本是在unix命令行上执行的，那么它们运行得非常好，并产生预期的输出
cat join2*.txt|/join3_mapper.py|sort|/join3_reducer.py
- 但是作为hadoop任务执行脚本非常失败：
hadoop-jar/usr/lib/hadop-mapreduce/hadop-streaming.jar-input/user/cloudera/inputTV/join2_gen*.txt-output/user/coloudera/output_tv-mapper/home/cloudera/join3_mapper.py-rereducer/home/cloudera/join3_reducer.py-numReduce任务1
16/01/06 12:32:32 INFO mapreduce.Job: Task Id : attempt_1452069211060_0026_r_000000_0, Status : FAILED
Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1
at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:325)
at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:538)
at org.apache.hadoop.streaming.PipeReducer.close(PipeReducer.java:134)
at org.apache.hadoop.io.IOUtils.cleanup(IOUtils.java:244)
at org.apache.hadoop.mapred.ReduceTask.runOldReducer(ReduceTask.java:459)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:392)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1671)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)
-
如果hadoop命令是在-numReduceTasks为0的情况下执行的，hadoop作业只执行map步骤，成功结束，输出目录包含map步骤的结果文件。
-
我想reduce步骤一定有问题吧？
- Hue中的stderr日志没有显示任何相关内容：
日志上传时间：2016年1月6日星期三12:33:10-0800日志长度：222log4j:WARN找不到记录器（org.apache.hadop.ipc.Server）的附加程序。log4j:WARN请正确初始化log4j系统。log4j:警告请参阅http://logging.apache.org/log4j/1.2/faq.html#noconfig了解更多信息。
脚本代码：第一个文件：join3_mapp.py
#!/usr/bin/env python
import sys
for line in sys.stdin:
line = line.strip() #strip out carriage return
tuple2 = line.split(",") #split line, into key and value, returns a list
if len(tuple2) == 2:
key = tuple2[0]
value = tuple2[1]
if value == 'ABC':
print('%st%s' % (key, value) )
elif value.isdigit():
print('%st%s' % (key, value) )
第二个文件：join3_reducer.py
#!/usr/bin/env python
import sys
last_key = None #initialize these variables
running_total = 0
abcFound =False;
this_key = None
# -----------------------------------
# Loop the file
# --------------------------------
for input_line in sys.stdin:
input_line = input_line.strip()
# --------------------------------
# Get Next Key value pair, splitting at tab
# --------------------------------
tuple2 = input_line.split("t")
this_key = tuple2[0]
value = tuple2[1]
if value.isdigit():
value = int(value)
# ---------------------------------
# Key Check part
# if this current key is same
# as the last one Consolidate
# otherwise Emit
# ---------------------------------
if last_key == this_key:
if value == 'ABC': # filter for only ABC in TV shows
abcFound=True;
else:
if isinstance(value, (int,long) ):
running_total += value
else:
if last_key: #if this key is different from last key, and the previous
# (ie last) key is not empy,
# then output
# the previous <key running-count>
if abcFound:
print('%st%s' % (last_key, running_total) )
abcFound=False;
running_total = value #reset values
last_key = this_key
if last_key == this_key:
print('%st%s' % (last_key, running_total) )
我尝试了各种不同的方法来向hadoop命令声明输入文件，没有区别，就没有成功。
我做错了什么？提示，想法非常感谢
真是一记幸运的重拳，与那一拳战斗了好几天，我知道我成功了：
由于的本地（unix）执行
cat join2_gen*.txt | ./join2_mapper.py | sort | ./join2_reducer.py
工作正常我有一个想法，使用1个合并的输入文件，而不是提供的6个输入文件，所以：
cat join2_gen*.txt >> mergedinputFile.txt
hdfs dfs -put mergedInputFile.txt /user/cloudera/input
然后再次执行相同的hadoop命令，将输入引导到input文件夹-->中的mergedInputFile，完美结果，没有问题，没有异常。
对我来说，这提出了一个问题：
- 为什么它只处理一个合并的输入文件，而现在却提供较小的6个文件？？还不知道
尝试将所有输入文本文件放在一个目录中，然后将该目录作为输入传递。这样你就不必合并所有的输入文件