使用导入模块流式传输python脚本

Question

使用pigtmp$ pig——versionApache Pig版本0.8.1-cdh3u1(已报告)编译Jul 18 2011, 08:29:40

我有一个python脚本(c-python)，它导入另一个脚本，在我的例子中都很简单:

数据$ hadoop fs -cat/user/pavel/trivial.log

1   one
2   two
3   three

示例没有INCLUDE -工作正常

$ 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) testrongtream.py

#! /usr/bin/env python
import sys
import string
for line in sys.stdin:
    if len(line) == 0: continue
    new_line = line
    print "%dt%s" % (1, new_line) 

实际上我只是用一个键聚合行，没有什么特别的。

包含炸弹的例子!现在我想添加一个来自python导入模块的字符串，该模块位于与testrongtream.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) testrongtream.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 "%dt%s" % (1, new_line) 

3) test_import.py

def getTestLine():
    return "test line";

现在

example$ pig -f trivial_stream.pig

后端错误信息

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)

猪堆跟踪

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

非常感谢你的帮助!帕维尔

Answer 1

正确答案:

如果你想让你的python应用程序与pig一起工作，你需要tar它(不要忘记init.py的!)，然后在pig的SHIP语句中包含.tar文件。你要做的第一件事是解压缩应用程序。路径可能会有问题，所以我建议在解压缩之前执行以下操作:插入(0,os.getcwd ()) .

Answer 2

您需要将当前目录附加到test_stream.py中的sys.path:

#! /usr/bin/env python
import sys
sys.path.append(".")

因此，您在那里的SHIP命令确实发布了python脚本，但您只需要告诉python在哪里查找。