运行 oozie fork 时出现 LeaseExpiredException

Posted 2023-04-18

【中文标题】运行 oozie fork 时出现 LeaseExpiredException

【英文标题】：LeaseExpiredException while running oozie fork

【发布时间】：2013-02-20 12:17:17

【问题描述】：

我们正在尝试运行 Oozie 工作流程，其中 3 个子工作流程使用 fork 并行运行。子工作流包含一个运行本机 map reduce 作业的节点，以及随后运行一些复杂的PIG 作业的两个节点。最后，三个子工作流连接到一个end 节点。

当我们运行这个工作流时，我们会得到LeaseExpiredException。运行PIG 作业时会随机发生异常。它发生的时间没有确定的位置，但每次我们运行WF时都会发生。

此外，如果我们删除 fork 并按顺序运行子工作流，它可以正常工作。但是，我们的期望是让它们在某些执行时间并行运行。

能否请您帮助我理解这个问题以及我们可能会出错的一些指示。我们从hadoop 开发开始，之前没有遇到过这样的问题。

看起来由于多个任务并行运行，其中一个线程关闭了一个零件文件，当另一个线程试图关闭该文件时，它会抛出错误。

以下是 hadoop 日志中异常的堆栈跟踪。

2013-02-19 10:23:54,815 INFO org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher: 57% complete 
2013-02-19 10:26:55,361 INFO org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher: 59% complete
2013-02-19 10:27:59,666 ERROR org.apache.hadoop.hdfs.DFSClient: Exception closing file <hdfspath>/oozie-oozi/0000105-130218000850190-oozie-oozi-W/aggregateData--pig/output/_temporary/_attempt_201302180007_0380_m_000000_0/part-00000 : org.apache.hadoop.ipc.RemoteException: org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException: No lease on <hdfspath>/oozie-oozi/0000105-130218000850190-oozie-oozi-W/aggregateData--pig/output/_temporary/_attempt_201302180007_0380_m_000000_0/part-00000 File does not exist. Holder DFSClient_attempt_201302180007_0380_m_000000_0 does not have any open files.
                at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkLease(FSNamesystem.java:1664)
                at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkLease(FSNamesystem.java:1655)
                at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.completeFileInternal(FSNamesystem.java:1710)
                at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.completeFile(FSNamesystem.java:1698)
                at org.apache.hadoop.hdfs.server.namenode.NameNode.complete(NameNode.java:793)
                at sun.reflect.GeneratedMethodAccessor34.invoke(Unknown Source)
                at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
                at java.lang.reflect.Method.invoke(Method.java:597)
                at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:557)
                at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1439)
                at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:1435)
                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:1278)
                at org.apache.hadoop.ipc.Server$Handler.run(Server.java:1433)

以下是主要工作流程和一个子工作流程的示例。

主要工作流程：

<workflow-app xmlns="uri:oozie:workflow:0.2" name="MainProcess">
<start to="forkProcessMain"/>
<fork name="forkProcessMain">
    <path start="Proc1"/>
    <path start="Proc2"/>
    <path start="Proc3"/>
</fork>
<join name="joinProcessMain" to="end"/>
<action name="Proc1">
    <sub-workflow>
        <app-path>$nameNode$wfPath/proc1_workflow.xml</app-path>
        <propagate-configuration/>
    </sub-workflow>
    <ok to="joinProcessMain"/>
    <error to="fail"/>
</action>   
<action name="Proc2">
    <sub-workflow>
        <app-path>$nameNode$wfPath/proc2_workflow.xml</app-path>
        <propagate-configuration/>
    </sub-workflow>
    <ok to="joinProcessMain"/>
    <error to="fail"/>
</action>   
<action name="Proc3">
    <sub-workflow>
        <app-path>$nameNode$wfPath/proc3_workflow.xml</app-path>
        <propagate-configuration/>
    </sub-workflow>
    <ok to="joinProcessMain"/>
    <error to="fail"/>
</action>   
<kill name="fail">
    <message>WF Failure, 'wf:lastErrorNode()' failed, error message[$wf:errorMessage(wf:lastErrorNode())]</message>
</kill>
<end name="end"/>

子工作流程：

<workflow-app xmlns="uri:oozie:workflow:0.2" name="Sub Process">
<start to="Step1"/>
<action name="Step1">
    <java>
        <job-tracker>$jobTracker</job-tracker>
        <name-node>$nameNode</name-node>
        <prepare>
           <delete path="$step1JoinOutputPath"/>
        </prepare>
        <configuration>
            <property>
                <name>mapred.queue.name</name>
                <value>$queueName</value>
            </property>
        </configuration>
        <main-class>com.absd.mr.step1</main-class>
        <arg>$wf:name()</arg>
        <arg>$wf:id()</arg>
        <arg>$tbMasterDataOutputPath</arg>
        <arg>$step1JoinOutputPath</arg>
        <arg>$tbQueryKeyPath</arg>
        <capture-output/>
    </java>
    <ok to="generateValidQueryKeys"/>
    <error to="fail"/>
</action>
<action name="generateValidQueryKeys">
    <pig>
        <job-tracker>$jobTracker</job-tracker>
        <name-node>$nameNode</name-node>
        <prepare>
           <delete path="$tbValidQuerysOutputPath"/>
        </prepare>
        <configuration>
            <property>
                <name>pig.tmpfilecompression</name>
                <value>true</value>
            </property>
            <property>
                <name>pig.tmpfilecompression.codec</name>
                <value>lzo</value>
            </property>
            <property>
                <name>pig.output.map.compression</name>
                <value>true</value>
            </property>
            <property>
                <name>pig.output.map.compression.codec</name>
                <value>lzo</value>
            </property>
            <property>
                <name>pig.output.compression</name>
                <value>true</value>
            </property>
            <property>
                <name>pig.output.compression.codec</name>
                <value>lzo</value>
            </property>
            <property>
                <name>mapred.compress.map.output</name>
                <value>true</value>
            </property>
        </configuration>
        <script>$pigDir/tb_calc_valid_accounts.pig</script>
        <param>csvFilesDir=$csvFilesDir</param>
        <param>step1JoinOutputPath=$step1JoinOutputPath</param>
        <param>tbValidQuerysOutputPath=$tbValidQuerysOutputPath</param>
        <param>piMinFAs=$piMinFAs</param>
        <param>piMinAccounts=$piMinAccounts</param>
        <param>parallel=80</param>
    </pig>
    <ok to="aggregateAumData"/>
    <error to="fail"/>
</action>
<action name="aggregateAumData">
    <pig>
        <job-tracker>$jobTracker</job-tracker>
        <name-node>$nameNode</name-node>
        <prepare>
           <delete path="$tbCacheDataPath"/>
        </prepare>
        <configuration>
            <property>
                <name>pig.tmpfilecompression</name>
                <value>true</value>
            </property>
            <property>
                <name>pig.tmpfilecompression.codec</name>
                <value>lzo</value>
            </property>
            <property>
                <name>pig.output.map.compression</name>
                <value>true</value>
            </property>
            <property>
                <name>pig.output.map.compression.codec</name>
                <value>lzo</value>
            </property>
            <property>
                <name>pig.output.compression</name>
                <value>true</value>
            </property>
            <property>
                <name>pig.output.compression.codec</name>
                <value>lzo</value>
            </property>
            <property>
                <name>mapred.compress.map.output</name>
                <value>true</value>
            </property>
        </configuration>
        <script>$pigDir/aggregationLogic.pig</script>
        <param>csvFilesDir=$csvFilesDir</param>
        <param>tbValidQuerysOutputPath=$tbValidQuerysOutputPath</param>
        <param>tbCacheDataPath=$tbCacheDataPath</param>
        <param>currDate=$date</param>
        <param>udfJarPath=$nameNode$wfPath/lib</param>
        <param>parallel=150</param>
      </pig>
    <ok to="loadDataToDB"/>
    <error to="fail"/>
</action>   
<kill name="fail">
    <message>WF Failure, 'wf:lastErrorNode()' failed, error message[$wf:errorMessage(wf:lastErrorNode())]</message>
</kill>
<end name="end"/>

【问题讨论】：

我也有同样的问题，如果有人能回答这个问题...

我们遇到了完全相同的问题，即 Hive 查询（oozie 操作和直线）和流式传输。结果：很多数据没有处理。一些（5% 到 20%）的 Streaming 进程被 SIGTERM 杀死。我们使用 YARN 和 CDH-4.1.1，qnd 也在 CDH-4.1.2 和 CDH-4.2.0-rc 上进行了尝试。

@DanishKhan 这里讨论了一个类似的问题：***.com/questions/14170186/…。您使用什么 Hadoop/Oozie 版本？在这种情况下你有自定义的 OutputFormat 吗？

@LorandBendig 感谢您指出线程。是的，我们正在运行并行 MR 作业，这从 fork 结构中可以看出，但我认为，因为它是由相同的工作流产生的，@ 987654334@ 会处理并行和临时文件。我们没有使用Hive，也没有使用CustomOutputFormat。

【参考方案1】：

当我们同时运行三个 pig 操作并且其中一个失败时，我们遇到了同样的错误。该消息错误是工作流程意外停止的结果，因为一项操作失败，工作流程已停止，而其他操作正在尝试继续。 您必须查看状态为 ERROR 的失败操作才能知道发生了什么，而不是查看状态为 KILLED 的操作