Потоковая передача через скрипт Python с модулями импорта - PullRequest
Новая
       17

Потоковая передача через скрипт Python с модулями импорта

4 голосов
/ 23 ноября 2011

Работа с pigtmp $ pig --version Apache Pig версии 0.8.1-cdh3u1 (rexported) скомпилирован 18 июля 2011 г., 08: 29: 40

У меня есть скрипт на python (c-python), которыйимпортирует другой скрипт, в моем примере оба очень простые:

Пример DATA $ hadoop fs -cat /user/pavel/trivial.log

1   one
2   two
3   three

ПРИМЕР БЕЗ ВКЛЮЧЕНИЯ - работает нормально

пример $ pig -f trivial_stream.pig 

(1,1,one)
()
(1,2,two)
()
(1,3,three)
()

где 1) trivial_stream.pig: 

DEFINE test_stream `test_stream.py` SHIP ('test_stream.py');
A = LOAD 'trivial.log' USING PigStorage('\t') AS (mynum: int, mynumstr: chararray);
C = STREAM A THROUGH test_stream;
DUMP C;

2) test_stream.py 

#! /usr/bin/env python

import sys
import string

for line in sys.stdin:
    if len(line) == 0: continue
    new_line = line
    print "%d\t%s" % (1, new_line)

Поэтому я просто собираю строки одним ключом, ничего особенного.

ПРИМЕР С ВКЛЮЧЕНИЕМ - бомбы!Теперь я хотел бы добавить строку из модуля импорта python, которая находится в том же каталоге, что и test_stream.py.Я пытался отправить модуль импорта разными способами, но получил ту же ошибку (см. Ниже)

1) trivial_stream.pig: 

DEFINE test_stream `test_stream.py` SHIP ('test_stream.py', 'test_import.py');
A = LOAD 'trivial.log' USING PigStorage('\t') AS (mynum: int, mynumstr: chararray);
C = STREAM A THROUGH test_stream;
DUMP C;

2) test_stream.py 

#! /usr/bin/env python

import sys
import string

import test_import

for line in sys.stdin:
    if len(line) == 0: continue
    new_line = ("%s-%s") % (line.strip(), test_import.getTestLine())
    print "%d\t%s" % (1, new_line)

3) test_import.py 

def getTestLine():
    return "test line";

Now 

example$ pig -f trivial_stream.pig

Сообщение об ошибке Backend 

org.apache.pig.backend.executionengine.ExecException: ERROR 2055: Received Error while processing the map plan: 'test_stream.py ' failed with exit status: 1
    at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase.runPipeline(PigMapBase.java:265)
    at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase.cleanup(PigMapBase.java:103)
    at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:146)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:647)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:323)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:270)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1127)
    at org.apache.hadoop.mapred.Child.main(Child.java:264)

Трассировка Pig Stack 

ERROR 2997: Unable to recreate exception from backed error: org.apache.pig.backend.executionengine.ExecException: ERROR 2055: Received Error while processing the map plan: 'test_stream.py ' failed with exit status: 1

org.apache.pig.impl.logicalLayer.FrontendException: ERROR 1066: Unable to open iterator for alias C. Backend error : Unable to recreate exception from backed error: org.apache.pig.backend.executionengine.ExecException: ERROR 2055: Received Error while processing the map plan: 'test_stream.py ' failed with exit status: 1
    at org.apache.pig.PigServer.openIterator(PigServer.java:753)
    at org.apache.pig.tools.grunt.GruntParser.processDump(GruntParser.java:615)
    at org.apache.pig.tools.pigscript.parser.PigScriptParser.parse(PigScriptParser.java:303)
    at org.apache.pig.tools.grunt.GruntParser.parseStopOnError(GruntParser.java:168)
    at org.apache.pig.tools.grunt.GruntParser.parseStopOnError(GruntParser.java:144)
    at org.apache.pig.tools.grunt.Grunt.exec(Grunt.java:90)
    at org.apache.pig.Main.run(Main.java:396)
    at org.apache.pig.Main.main(Main.java:107)
Caused by: org.apache.pig.backend.executionengine.ExecException: ERROR 2997: Unable to recreate exception from backed error: org.apache.pig.backend.executionengine.ExecException: ERROR 2055: Received Error while processing the map plan: 'test_stream.py ' failed with exit status: 1
    at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.Launcher.getErrorMessages(Launcher.java:221)
    at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.Launcher.getStats(Launcher.java:151)
    at org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher.launchPig(MapReduceLauncher.java:337)
    at org.apache.pig.backend.hadoop.executionengine.HExecutionEngine.execute(HExecutionEngine.java:382)
    at org.apache.pig.PigServer.executeCompiledLogicalPlan(PigServer.java:1209)
    at org.apache.pig.PigServer.storeEx(PigServer.java:885)
    at org.apache.pig.PigServer.store(PigServer.java:827)
    at org.apache.pig.PigServer.openIterator(PigServer.java:739)
    ... 7 more

Большое спасибо за вашу помощь!-Pavel

Ответы [ 2 ]

3 голосов
/ 15 сентября 2012

Правильный ответ из комментария выше:

Зависимости не поставляются, если вы хотите, чтобы ваше приложение на python работало с pig, вам нужно его сменить (не забудьте init.py!), Затем включитефайл .tar в заявлении SHIP свиньи.Первое, что вы делаете - распаковываете приложение.Могут быть проблемы с путями, поэтому я бы предложил следующее еще до извлечения tar: sys.path.insert (0, os.getcwd ()).

1 голос
/ 07 августа 2013

Вам нужно добавить текущий каталог к ​​sys.path в вашем test_stream.py: 

#! /usr/bin/env python

import sys
sys.path.append(".")

Таким образом, команда SHIP, которая у вас есть, отправляет скрипт python, но вам просто нужно указать Pythonсмотреть.

...