Hadoop Oozie学习笔记(五) E0720: Fork/join mismatch, node [join_node_name]异常解决

本博客属原创文章,转载请注明出处:http://guoyunsky.iteye.com/blogs/1299770/

 欢迎加入Hadoop超级群: 180941958 

 

    Oozie是个针对Hadoop的工作流,有些自己的语法. 这两天碰到一个异常,查看源码才明白Oozie的join只允许承接fork下来的任务,否则会报以下错误.整个异常如下:

  WARN CallableQueueService$CallableWrapper:528 - USER[-] GROUP[-] TOKEN[-] APP[-] JOB[-] ACTION[-] exception callable [signal], E0720: Fork/join mismatch, node [join_node_name]

org.apache.oozie.command.CommandException: E0720: Fork/join mismatch, node [tianqi_sawlog_transformation_done]

at org.apache.oozie.command.wf.SignalCommand.call(SignalCommand.java:213)

at org.apache.oozie.command.wf.SignalCommand.execute(SignalCommand.java:305)

at org.apache.oozie.command.wf.SignalCommand.execute(SignalCommand.java:59)

at org.apache.oozie.command.Command.call(Command.java:202)

at org.apache.oozie.service.CallableQueueService$CallableWrapper.run(CallableQueueService.java:128)

at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)

at java.lang.Thread.run(Thread.java:662)

Caused by: org.apache.oozie.workflow.WorkflowException: E0720: Fork/join mismatch, node [tianqi_sawlog_transformation_done]

at org.apache.oozie.workflow.lite.JoinNodeDef$JoinNodeHandler.loopDetection(JoinNodeDef.java:44)

at org.apache.oozie.workflow.lite.LiteWorkflowInstance.signal(LiteWorkflowInstance.java:203)

at org.apache.oozie.workflow.lite.LiteWorkflowInstance.signal(LiteWorkflowInstance.java:284)

at org.apache.oozie.command.wf.SignalCommand.call(SignalCommand.java:120)

... 7 more

 

    源码来自org.apache.oozie.workflow.lite.JoinNodeDef,检测这个语法的代码如下:

public void loopDetection(Context context) throws WorkflowException {
    String flag = getLoopFlag(context.getNodeDef().getName());
    if (context.getVar(flag) != null) {
        throw new WorkflowException(ErrorCode.E0709, context.getNodeDef().getName());
    }
    String parentExecutionPath = context.getParentExecutionPath(context.getExecutionPath());
    String forkCount = context.getVar(ForkNodeDef.FORK_COUNT_PREFIX + parentExecutionPath);
    if (forkCount == null) {
        throw new WorkflowException(ErrorCode.E0720, context.getNodeDef().getName());
    }
    int count = Integer.parseInt(forkCount) - 1;
    if (count == 0) {
        context.setVar(flag, "true");
    }
}

public boolean enter(Context context) throws WorkflowException {
    String parentExecutionPath = context.getParentExecutionPath(context.getExecutionPath());
    String forkCount = context.getVar(ForkNodeDef.FORK_COUNT_PREFIX + parentExecutionPath);
    if (forkCount == null) {
        throw new WorkflowException(ErrorCode.E0720, context.getNodeDef().getName());
    }
    int count = Integer.parseInt(forkCount) - 1;
    if (count > 0) {
        context.setVar(ForkNodeDef.FORK_COUNT_PREFIX + parentExecutionPath, "" + count);
        context.deleteExecutionPath();
    }
    else {
        context.setVar(ForkNodeDef.FORK_COUNT_PREFIX + parentExecutionPath, null);
    }
    return (count == 0);
}

 

    可以发现这个两个方法都会通过String forkCount = context.getVar(ForkNodeDef.FORK_COUNT_PREFIX + parentExecutionPath);去获取当前节点

的所有父节点是Fork的个数.如果为空,则通过这行代码throw new WorkflowException(ErrorCode.E0720, context.getNodeDef().getName());抛出这个异常

 

    所以我们今年在使用join节点的时候,一定要承接来自fork或者join本身的节点.下面举几个例子:

    1.错误的例子,由于join并没有承接fork或者join,所以会报以上的错误

<workflow-app xmlns="uri:oozie:workflow:0.1" name="workflow-test">
	<start to="action1">
	<action name="action1">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	<join name="join1" to="end" />
	
	<kill name="fail">
		<message>Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
	</kill>
	
	<end name="end" />
</workflow-app>

 

    2.正确的例子,join承接了fork(join1承接了fork1过来的action1和action2)

<workflow-app xmlns="uri:oozie:workflow:0.1" name="workflow-test">
	<start to="fork1">
	
	<fork name="fork1">

		<path start="action1" />
		<path start="action2" />
	</fork>
	
	<action name="action1">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	
	<action name="action2">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	
	<join name="join1" to="end" />
	
	<kill name="fail">
		<message>Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
	</kill>
	
	<end name="end" />
</workflow-app>

 

   3.正确的例子,join承接了join(join2承接了join1)

<workflow-app xmlns="uri:oozie:workflow:0.1" name="workflow-test">
	<start to="fork1">
	
	<fork name="fork1">
		<path start="action1" />
		<path start="action2" />
	</fork>
	
	<action name="action1">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	
	<action name="action2">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	
	<join name="join1" to="join2" />
	
	<join name="join2" to="end" />
	
	<kill name="fail">
		<message>Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
	</kill>
	
	<end name="end" />
</workflow-app>

   4.正确的例子,join承接了fork下来的所有节点

<workflow-app xmlns="uri:oozie:workflow:0.1" name="workflow-test">
	<start to="fork1">
	
	<fork name="fork1">
		<path start="action1" />
		<path start="action2" />
	</fork>
	
	<action name="action1">
		<!-- do some things-->
		<ok to="action3" />
		<error to="fail" />
	</action>
	
	<action name="action2">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	
	<action name="action3">
		<!-- do some things-->
		<ok to="join1" />
		<error to="fail" />
	</action>
	
	<join name="join1" to="end" />
	
	<kill name="fail">
		<message>Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
	</kill>
	
	<end name="end" />
</workflow-app>

 

   这里只是大概玩了下Oozie,希望能起抛砖引玉...

 

 

更多技术文章、感悟、分享、勾搭，请用微信扫描: