Задание потоковой передачи Hadoop не удалось в python

Question

У меня есть работа с mapreduce, написанная на Python.Программа была успешно протестирована в 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)

Answer 1

У меня возникла та же проблема, и мне было интересно, потому что, когда я тестировал свой маппер и редуктор на тестовых данных, он запускался. Но когда я запускал тот же набор тестов через hadoop map-lower, у меня возникала та же проблема.

Как проверить свой код локально:

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

При дополнительном расследовании я обнаружил, что не включил 'строку shebang' в свой скрипт mapper.py.

#!/usr/bin/python

Пожалуйста, добавьте вышеуказанную строку в качестве первой строки вашего скрипта Python и оставьте пустую строку после этого.

Если вам нужно больше узнать о «линии Шебанга», прочитайте Почему люди пишут #! / Usr / bin / env python в первой строке скрипта Python?

Answer 2

Вы можете найти сообщения об ошибках Python (например, traceback) и другие вещи, записанные вашим скриптом в stderr в веб-интерфейсе hadoop.Он немного скрыт, но вы найдете его в ссылке, которую предоставляет вам потоковая передача.Вы нажимаете «Карта» или «Уменьшить», затем нажимаете на любую задачу, а затем в столбце «Журналы задач» на «Все»

Answer 3

Грязный вход может вызвать эту проблему.

Попробуйте использовать 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

Answer 4

У меня была такая же проблема сегодня, играя с Hadoop 1.0.1. К счастью, я решил это:

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

(Мои скрипты Python были в текущем каталоге). Похоже, сейчас нужны абсолютные пути.

Best!

Answer 5

Потоковая передача Hadoop - Hadoop 1.0.x

У меня была та же проблема с "сломанной трубой".Проблема заключалась в том, что в моем редукторе было выражение «перерыв».Итак, все прошло хорошо до «перерыва».После этого работающий редуктор перестал печатать ошибку «Разбитая труба».Кроме того, начал работать другой редуктор с той же судьбой, что и предыдущий.Этот круг продолжался и продолжался.

Если я правильно понял, когда редуктор начинает читать из стандартного ввода (это был мой случай в цикле for), то он должен читать все.Вы не можете «прервать» эту операцию, даже если закроете стандартный ввод (os.close (0), как я пытался).

Answer 6

Наконец я исправил ошибку, и вот уроки, которые я выучил.1) Исходный код не имеет обработки ошибок на неверных данных.Я не обнаружил проблему при тестировании кода на небольшом наборе данных.2) Для обработки пустых полей / переменных я обнаружил, что в Python немного сложно протестировать None и пустые строки.Лично мне нравится функция len (strVar), которая легко читается и эффективна.3) Команда hadoop в этом случае верна.Каким-то образом * .py с режимом 644 может быть успешно запущен в среде, которую я использую.

Answer 7

Одним из возможных решений является включение "python", т.е.:

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

Answer 8

Python + Hadoop сложен в некоторых деталях, которые не должны быть. Взгляните здесь .

Попробуйте заключить свой путь ввода в двойные кавычки. (-вход "/ data / omni / 20110115 / exp6-10122")