带有 oozie 工作流程的猪

Posted 2023-04-18

【中文标题】带有 oozie 工作流程的猪

【英文标题】：Pig with oozie workflow

【发布时间】：2014-06-17 22:57:26

【问题描述】：

我有一个包装外壳脚本，它读取输入文件并多次调用猪脚本。这在 crontab 上运行良好。但是，在尝试使用 oozie 引擎时不起作用。如何在 oozie 中实现循环。如果这不是正确的方法，有人可以建议替代方法吗？我需要能够从输入文件中读取一些参数，然后为每一行（和相应的读取参数）多次调用猪脚本或猪动作。

---

读行时

do
    seedidprefix=$(echo $line |awk -v OFS='\t' 'print $3;')
    seedaudprefix=$(echo $line |awk -v OFS='\t' 'print $2;')
    adgrpid=$(echo $line |awk -v OFS='\t' 'print $1;')
    nwtype=$(echo $line |awk -v OFS='\t' 'print $4;')
    chainedseedid=$(echo $line |awk -v OFS='\t' 'print $5;')

if [ $nwtype == "ADX" ]; then

    hadoop jar pig.jar -param segid=$seedaudprefix -param nw=$nwtype -param date=$MODELDATE -param seedid=$seedidprefix -param audienceId=$adgrpid -param chainedseedid=$chainedseedid $TOP_DIR/histo_bkup_adx.pig

else

    hadoop jar pig.jar -param segid=$seedaudprefix -param nw=$nwtype -param date=$MODELDATE -param seedid=$seedidprefix -param audienceId=$adgrpid -param chainedseedid=$chainedseedid $TOP_DIR/histo_bkup.pig

fi

done<$campaign_file

---

【问题讨论】：

你为什么要循环播放？你能举个数据例子吗？

【参考方案1】：

我遇到了类似的问题 - 我需要多次执行猪脚本。

对我有用的是在 Python 中使用 Embedded Pig（[参见文档][1]）。

Oozie 工作流有一个 pig 动作，其中 script 指向 python 文件。 还使用 file 标记来标记需要重复执行的 pig 脚本，以便将其加载到工作目录（控制器所在的文件也是如此，在您的情况下为 shell脚本，读取输入）。

Oozie 脚本看起来像，

<workflow>
    ...        
    <action name="action_1" retry-max="0" retry-interval="1">
        <pig>        
            <script>$path_src/my_controller.py</script>
            <file>$path_src/data.txt#data.txt</file>
            <file>$path_src/my_script.pig#my_script.py</file>
        </pig>
        <ok to="Done"/>
        <error to="Fail"/>
    </action>
    ...
</workflow>

python 脚本看起来像，

#!/usr/bin/python
from org.apache.pig.scripting import Pig
if __name__ == '__main__' :
    # Load pig parameters - assuming each parameter is in the format KEY = Value
    pigParams = 
    ipFile = open( 'data.txt', 'r' )
    for line in ipFile :
        if len( line.strip() ) == 0 :
            continue
        fields = line.split( '=' )
        pigParams[ fields[ 0 ].strip() ] = fields[ 1 ].strip()
    ipFile.close()

    for i in range( 10 ) :
        pigScript = Pig.compileFromFile( 'my_script.pig' )
        pigScript.bind( pigParams ).run()

希望这会有所帮助。