如何使用 PIG 加载文件夹中的每个文件？

Posted 2023-04-18

【中文标题】如何使用 PIG 加载文件夹中的每个文件？

【英文标题】：How Can I Load Every File In a Folder Using PIG?

【发布时间】：2011-09-07 20:38:47

【问题描述】：

我有一个每天创建的文件文件夹，它们都存储相同类型的信息。我想制作一个脚本来加载最新的 10 个，将它们联合起来，然后在它们上运行一些其他代码。由于 pig 已经有一个 ls 方法，我想知道是否有一种简单的方法可以让我获取最后 10 个创建的文件，并使用相同的加载器和选项以通用名称加载它们。我猜它看起来像：

REGISTER /usr/local/lib/hadoop/hadoop-lzo-0.4.13.jar;
REGISTER /usr/local/lib/hadoop/elephant-bird-2.0.5.jar;
FOREACH file in some_path:
    file = LOAD 'file' 
    USING com.twitter.elephantbird.pig.load.LzoTokenizedLoader('\\t') 
    AS (i1, i2, i3);

【参考方案1】：

这不是我能够开箱即用的事情，而是可以通过某种包装脚本或帮助脚​​本（bash、perl 等）在脚本之外完成的事情。如果您编写一个名为 last10.sh 的脚本，它将输出您最后 10 个文件，以逗号分隔：

$ ./last10.sh
/input/file38,/input/file39,...,/input/file48

对于最近的 10 个文件，这样的事情应该可以解决问题：

hadoop fs -ls /input/ | sort -k6,7 | tail -n10 | awk 'print $8' | tr '\n' ','

你可以这样做：

$ pig -p files="`last10.sh`" my_mr.pig

然后，在你的猪脚本中，执行：

data = LOAD '$files'
       USING com.twitter.elephantbird.pig.load.LzoTokenizedLoader('\\t')
       AS (i1, i2, i3);

如果文件像这样用逗号分隔，Pig 会加载单独的文件。这相当于做：

data = LOAD '/input/file38,/input/file39,...,/input/file48'
       USING com.twitter.elephantbird.pig.load.LzoTokenizedLoader('\\t')
       AS (i1, i2, i3);

【讨论】：

甜蜜！如果 PIG 为我提供了一种直接执行此操作的方法会更好，但这绝对有效。谢谢！

我同意。 Pig 擅长做分析的东西，但是当涉及到分析之外的任何类型的真正集成时，它并没有太多。我的团队几乎承认我们所有的猪脚本都需要用 bash 包装。

没关系。原来 pig 不喜欢空格，所以像 pig -p files="file1 file2" script.pig 这样的东西不起作用并因“在命令行上遇到意外参数”错误而死。你有解决方法吗？

糟糕！我的错！它喜欢逗号，而不是空格。我正在更新我的答案以用逗号替换换行符。让我知道这是否有效。我知道有些人在file1,file2,file3 这样的路径周围使用 ，但我认为它们具有相同的效果。

酷！我唯一的建议是你在最后添加一个管道到 awk 以摆脱在开头和结尾产生的括号。这将是您在上面的答案并添加：| awk ' print substr($0, 1, length($0)-1) '

【参考方案2】：

Donald Miner 的答案仍然非常有效，但 IMO 现在有更好的方法来解决这个问题，在 Python 中使用 Embedded Pig。 O'Reilly 有一个简短的解释here。还有一个关于为什么这是你想做的事情的演示，以及它是如何工作的here。长话短说，在运行 pig 脚本以确定脚本的各个部分之前，可以访问很多功能。在 Jython 中包装和/或动态生成部分脚本让您这样做。高兴！

【参考方案3】：

我喜欢以上两种方法。只是想为 oozie 爱好者提供更多选择。 oozie 中的 Java 操作在由“oozie.action.output.properties”配置的位置中吐出一个文件，Pig 操作将其传递给 pig 脚本。与上述 2 相比，这绝对不是优雅的解决方案。我在 oozie 中使用 java 调度配置嵌入式猪时遇到了麻烦，所以我不得不采用这个解决方案。

<workflow-app xmlns='uri:oozie:workflow:0.1' name='java-wf'>
<start to='java1' />

<action name='java1'>
    <java>
        <job-tracker>$jobTracker</job-tracker>
        <name-node>$nameNode</name-node>
        <configuration>
           <property>
                <name>mapred.job.queue.name</name>
                <value>$queueName</value>
            </property>
        </configuration>
        <main-class>org.apache.oozie.test.MyTest</main-class>
        <arg>$outputFileName</arg>
        <capture-output/>
    </java>
    <ok to="pig1" />
    <error to="fail" />
</action>


<action name='pig1'>
    <pig>
        <job-tracker>$jobTracker</job-tracker>
        <name-node>$nameNode</name-node>
        <configuration>
            <property>
                <name>mapred.job.queue.name</name>
                <value>$queueName</value>
            </property>
        </configuration>
        <script>script.pig</script>
        <param>MY_VAR=$wf:actionData('java1')['PASS_ME']</param>
    </pig>
    <ok to="end" />
    <error to="fail" />
</action>

<kill name="fail">
    <message>Pig failed, error message[$wf:errorMessage(wf:lastErrorNode())]</message>
</kill>
<end name='end' />