mapreducer в hadoop выводит ошибку Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1
Имеется текстовый файл, в котором необходимо подсчитать кол-во слов без учета знаков препинания и перевести все слова к нижнему регистру. файл mapper.py выглядит так:
#!/usr/bin/env python
import sys
for line in sys.stdin:
# remove extra spaces
line = line.strip()
# split strings to words
words = line.split()
list = '''!()-[]{};?@#$%:'"\,./^&*_<>&nbs;'''
for k in words:
if k in list:
words = words.replace(k, "").lower()
# add counter
for word in words:
# output key value
print ('%s\t%s' % (word, 1))
файл reducer.py выглядит так:
#!/usr/bin/env python
import sys
current_word = None
current_count = 0
word = None
# read strings from stdin
for line in sys.stdin:
# remove spaces at hte begining and at the end
line = line.strip()
# split key value
word, count = line.split('\t', 1)
# convert value to int
try:
count = int(count)
except ValueError:
# ignore errors
continue
#Reduce get data after sort
if current_word == word:
current_count += count
else:
if current_word:
# write result to STDOUT
print ('%s\t%s' % (current_word, current_count))
current_count = count
current_word = word
# Don't forget last word
if current_word == word:
print ('%s\t%s' % (current_word, current_count))
при выполнении команды yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar -input input/text -output result -mapper "python mapper.py" -reducer "python reducer.py" -file mapper.py -file reducer.py
выдает следующее:
21/06/01 21:57:13 WARN streaming.StreamJob: -file option is deprecated, please use generic option -files instead.
packageJobJar: [mapper.py, reducer.py] [/usr/hdp/3.1.4.0-315/hadoop-mapreduce/hadoop-streaming-3.1.1.3.1.4.0-315.jar] /var/lib/ambari-agent/tmp/hadoop_java_io_tmpdir/streamjob4068879117326248872.jar tmpDir=null
21/06/01 21:57:14 INFO client.RMProxy: Connecting to ResourceManager at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:8050
21/06/01 21:57:14 INFO client.AHSProxy: Connecting to Application History server at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:10200
21/06/01 21:57:14 INFO client.RMProxy: Connecting to ResourceManager at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:8050
21/06/01 21:57:14 INFO client.AHSProxy: Connecting to Application History server at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:10200
21/06/01 21:57:14 ERROR streaming.StreamJob: Error Launching job : Output directory hdfs://bigdataanalytics2-head-shdpt-v31-1-0.novalocal:8020/user/el_student_4472183/result already exists
Streaming Command Failed!
[el_student_4472183@bigdataanalytics2-head-shdpt-v31-1-0 ~]$ hdfs dfs -cat result/*
cat: `result/*': No such file or directory
[el_student_4472183@bigdataanalytics2-head-shdpt-v31-1-0 ~]$ hdfs dfs -rm -r result
21/06/01 21:57:48 INFO fs.TrashPolicyDefault: Moved: 'hdfs://bigdataanalytics2-head-shdpt-v31-1-0.novalocal:8020/user/el_student_4472183/result' to trash at: hdfs://bigdataanalytics2-head-shdpt-v31-1-0.novalocal:8020/user/el_student_4472183/.Trash/Current/user/el_student_4472183/result1622584668904
[el_student_4472183@bigdataanalytics2-head-shdpt-v31-1-0 ~]$ yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar -input input/text -output result -mapper "python mapper.py" -reducer "python reducer.py" -file mapper.py -file reducer.py
21/06/01 21:57:52 WARN streaming.StreamJob: -file option is deprecated, please use generic option -files instead.
packageJobJar: [mapper.py, reducer.py] [/usr/hdp/3.1.4.0-315/hadoop-mapreduce/hadoop-streaming-3.1.1.3.1.4.0-315.jar] /var/lib/ambari-agent/tmp/hadoop_java_io_tmpdir/streamjob6604620941313183154.jar tmpDir=null
21/06/01 21:57:53 INFO client.RMProxy: Connecting to ResourceManager at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:8050
21/06/01 21:57:53 INFO client.AHSProxy: Connecting to Application History server at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:10200
21/06/01 21:57:53 INFO client.RMProxy: Connecting to ResourceManager at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:8050
21/06/01 21:57:53 INFO client.AHSProxy: Connecting to Application History server at bigdataanalytics2-head-shdpt-v31-1-0.novalocal/10.0.0.7:10200
21/06/01 21:57:54 INFO mapreduce.JobResourceUploader: Disabling Erasure Coding for path: /user/el_student_4472183/.staging/job_1620746155522_0349
21/06/01 21:57:54 INFO mapred.FileInputFormat: Total input files to process : 1
21/06/01 21:57:54 INFO mapreduce.JobSubmitter: number of splits:2
21/06/01 21:57:55 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1620746155522_0349
21/06/01 21:57:55 INFO mapreduce.JobSubmitter: Executing with tokens: []
21/06/01 21:57:55 INFO conf.Configuration: found resource resource-types.xml at file:/etc/hadoop/3.1.4.0-315/0/resource-types.xml
21/06/01 21:57:55 INFO impl.YarnClientImpl: Submitted application application_1620746155522_0349
21/06/01 21:57:55 INFO mapreduce.Job: The url to track the job: http://bigdataanalytics2-head-shdpt-v31-1-0.novalocal:8088/proxy/application_1620746155522_0349/
21/06/01 21:57:55 INFO mapreduce.Job: Running job: job_1620746155522_0349
21/06/01 21:58:00 INFO mapreduce.Job: Job job_1620746155522_0349 running in uber mode : false
21/06/01 21:58:00 INFO mapreduce.Job: map 0% reduce 0%
21/06/01 21:58:09 INFO mapreduce.Job: map 100% reduce 0%
21/06/01 21:58:13 INFO mapreduce.Job: Task Id : attempt_1620746155522_0349_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.reduce(PipeReducer.java:128)
at org.apache.hadoop.mapred.ReduceTask.runOldReducer(ReduceTask.java:445)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:393)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:174)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1730)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:168)
21/06/01 21:58:18 INFO mapreduce.Job: Task Id : attempt_1620746155522_0349_r_000000_1, 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.reduce(PipeReducer.java:128)
at org.apache.hadoop.mapred.ReduceTask.runOldReducer(ReduceTask.java:445)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:393)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:174)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1730)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:168)
21/06/01 21:58:23 INFO mapreduce.Job: Task Id : attempt_1620746155522_0349_r_000000_2, 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.reduce(PipeReducer.java:128)
at org.apache.hadoop.mapred.ReduceTask.runOldReducer(ReduceTask.java:445)
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:393)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:174)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1730)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:168)
21/06/01 21:58:29 INFO mapreduce.Job: map 100% reduce 100%
21/06/01 22:00:05 INFO mapreduce.Job: Job job_1620746155522_0349 failed with state FAILED due to: Task failed task_1620746155522_0349_r_000000
Job failed as tasks failed. failedMaps:0 failedReduces:1 killedMaps:0 killedReduces: 0
21/06/01 22:00:05 INFO mapreduce.Job: Counters: 39
File System Counters
FILE: Number of bytes read=0
FILE: Number of bytes written=8262606
FILE: Number of read operations=0
FILE: Number of large read operations=0
FILE: Number of write operations=0
HDFS: Number of bytes read=554313
HDFS: Number of bytes written=0
HDFS: Number of read operations=6
HDFS: Number of large read operations=0
HDFS: Number of write operations=0
Job Counters
Failed reduce tasks=4
Launched map tasks=2
Launched reduce tasks=4
Data-local map tasks=2
Total time spent by all maps in occupied slots (ms)=31623
Total time spent by all reduces in occupied slots (ms)=46472
Total time spent by all map tasks (ms)=10541
Total time spent by all reduce tasks (ms)=11618
Total vcore-milliseconds taken by all map tasks=10541
Total vcore-milliseconds taken by all reduce tasks=11618
Total megabyte-milliseconds taken by all map tasks=16190976
Total megabyte-milliseconds taken by all reduce tasks=23793664
Map-Reduce Framework
Map input records=5347
Map output records=386101
Map output bytes=7010545
Map output materialized bytes=7784152
Input split bytes=290
Combine input records=0
Spilled Records=386101
Failed Shuffles=0
Merged Map outputs=0
GC time elapsed (ms)=195
CPU time spent (ms)=5410
Physical memory (bytes) snapshot=2232684544
Virtual memory (bytes) snapshot=6493569024
Total committed heap usage (bytes)=2053636096
Peak Map Physical memory (bytes)=1122664448
Peak Map Virtual memory (bytes)=3249889280
File Input Format Counters
Bytes Read=554023
21/06/01 22:00:05 ERROR streaming.StreamJob: Job not successful!
Streaming Command Failed!
при этом команда cat text | ./mapper.py | sort | ./reducer.py отрабатывает, но не убирает лишние знаки и не переводит в нижний регистр.
Но если убрать из mapper.py строки:
list = '''!()-[]{};?@#$%:'"\,./^&*_<>&nbs;'''
for k in words:
if k in list:
words = words.replace(k, "").lower()
тогда работает без ошибок, но не такой результат мне нужен((.
при этом вышеуказанные строки в питоне работает отлично.
Подскажите пжл. в чем причина? почему не работает код?