Oozie工作流互斥调度需求：实现工作流等待触发与并发限制

嘿，这个互斥调度的需求在Oozie集群里其实挺常见的，我给你整理了两种生产环境里验证过的靠谱方案，直接就能上手：

方案1：用Oozie Coordinator原生特性搞定（推荐） #

Oozie Coordinator本身就带了并发控制和依赖检查的能力，刚好能完美匹配你的需求，不用额外搭别的组件：

给workflow1加“运行拦截”逻辑 #

在workflow1的Coordinator配置（coordinator.xml）里，我们要加个前置检查：只要workflow2还在跑，workflow1就不能启动。具体可以这么做：

• 用一个shell脚本调用Oozie CLI查询workflow2的运行状态
• 如果查到workflow2处于RUNNING状态，就让Coordinator每隔10分钟再检查一次，直到workflow2跑完
• 确认workflow2结束后，再提交workflow1的实例

给你贴个配置片段参考：

<coordinator-app name="workflow1-coord" frequency="${coord:days(1)}" start="2024-01-01T00:00Z" end="2024-12-31T00:00Z" timezone="UTC" xmlns="uri:oozie:coordinator:0.5">
    <action>
        <decision name="check-wf2-running">
            <switch>
                <case to="wait">${coord:actionData('check-wf2-status')['status'] eq 'RUNNING'}</case>
                <default to="submit-wf1"/>
            </switch>
        </decision>
        <!-- 调用脚本检查workflow2状态 -->
        <action name="check-wf2-status">
            <shell xmlns="uri:oozie:shell-action:0.3">
                <exec>check_wf2_status.sh</exec>
                <file>hdfs:///user/oozie/scripts/check_wf2_status.sh#check_wf2_status.sh</file>
                <capture-output/>
            </shell>
            <ok to="check-wf2-running"/>
            <error to="fail"/>
        </action>
        <!-- 等待10分钟再重试 -->
        <action name="wait">
            <sleep xmlns="uri:oozie:sleep-action:0.1">
                <minutes>10</minutes>
            </sleep>
            <ok to="check-wf2-status"/>
            <error to="fail"/>
        </action>
        <!-- 提交workflow1 -->
        <action name="submit-wf1">
            <workflow>
                <app-path>hdfs:///user/oozie/workflows/workflow1</app-path>
                <configuration>
                    <property>
                        <name>oozie.wf.rerun.failnodes</name>
                        <value>true</value>
                    </property>
                </configuration>
            </workflow>
            <ok to="end"/>
            <error to="fail"/>
        </action>
    </action>
    <kill name="fail">
        <message>Coordinator failed: ${wf:errorMessage(wf:lastErrorNode())}</message>
    </kill>
    <end name="end"/>
</coordinator-app>

对应的check_wf2_status.sh脚本很简单，就是查Oozie里workflow2的运行实例数：

#!/bin/bash
OOZIE_URL="http://你的oozie服务器地址:11000/oozie"
# 换成你实际的workflow2 Coordinator作业ID
WF2_COORD_JOB_ID="0000001-2401011234567-oozie-oozi-C"

# 统计RUNNING状态的workflow2实例数
RUNNING_COUNT=$(oozie job -oozie $OOZIE_URL -info $WF2_COORD_JOB_ID | grep "RUNNING" | wc -l)

if [ $RUNNING_COUNT -gt 0 ]; then
    echo "status=RUNNING"
else
    echo "status=COMPLETED"
fi

给workflow2加“等待workflow1完成”逻辑 #

和上面的思路一模一样，在workflow2的Coordinator里加个前置检查：先查workflow1有没有在跑，有就等，没有再启动自己。另外记得把workflow2 Coordinator的concurrency设为1，确保同一时间只有一个workflow2在跑，避免自己和自己抢资源。

方案2：用HDFS共享锁实现（适合没开Coordinator的场景） #

如果你的集群没启用Coordinator，或者需要更灵活的控制，那就用共享锁的思路——比如在HDFS上建个锁文件，谁在跑谁占着锁，另一个就等着：

核心逻辑 #

• workflow1启动时，在HDFS指定路径创建wf1.lock文件，跑完就删掉
• workflow2启动前，先检查wf1.lock存在不：存在就等，不存在就创建自己的wf2.lock再启动
• 不管workflow2成功还是失败，最后都要删掉自己的锁，避免死锁

给你贴个workflow2的xml配置片段：

<workflow-app name="workflow2" xmlns="uri:oozie:workflow:0.5">
    <start to="check-wf1-lock"/>
    <!-- 检查workflow1的锁 -->
    <action name="check-wf1-lock">
        <shell xmlns="uri:oozie:shell-action:0.3">
            <exec>check_wf1_lock.sh</exec>
            <file>hdfs:///user/oozie/scripts/check_wf1_lock.sh#check_wf1_lock.sh</file>
            <capture-output/>
        </shell>
        <ok to="decision-lock-check"/>
        <error to="fail"/>
    </action>
    <!-- 判断要不要等 -->
    <decision name="decision-lock-check">
        <switch>
            <case to="wait-wf1">${wf:actionData('check-wf1-lock')['lock_exists'] eq 'true'}</case>
            <default to="create-wf2-lock"/>
        </switch>
    </decision>
    <!-- 等待10分钟重试 -->
    <action name="wait-wf1">
        <sleep xmlns="uri:oozie:sleep-action:0.1">
            <minutes>10</minutes>
        </sleep>
        <ok to="check-wf1-lock"/>
        <error to="fail"/>
    </action>
    <!-- 创建自己的锁 -->
    <action name="create-wf2-lock">
        <shell xmlns="uri:oozie:shell-action:0.3">
            <exec>hdfs dfs -touchz /user/oozie/locks/wf2.lock</exec>
        </shell>
        <ok to="main-workflow"/>
        <error to="fail"/>
    </action>
    <!-- 你的主工作流逻辑 -->
    <action name="main-workflow">
        <map-reduce>
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <configuration>
                <!-- 替换成你的实际MR配置 -->
            </configuration>
        </map-reduce>
        <ok to="cleanup-lock"/>
        <error to="cleanup-lock"/>
    </action>
    <!-- 不管成功失败都删锁 -->
    <action name="cleanup-lock">
        <shell xmlns="uri:oozie:shell-action:0.3">
            <exec>hdfs dfs -rm -f /user/oozie/locks/wf2.lock</exec>
        </shell>
        <ok to="end"/>
        <error to="end"/>
    </action>
    <kill name="fail">
        <message>Workflow failed: ${wf:errorMessage(wf:lastErrorNode())}</message>
    </kill>
    <end name="end"/>
</workflow-app>

check_wf1_lock.sh脚本就是检查HDFS锁文件：

#!/bin/bash
LOCK_PATH="/user/oozie/locks/wf1.lock"

if hdfs dfs -test -e $LOCK_PATH; then
    echo "lock_exists=true"
else
    echo "lock_exists=false"
fi

记得给workflow1也加上创建/删除锁的逻辑哦！

一些要注意的坑 #

• 异常处理：不管用哪种方案，都要处理workflow意外失败的情况——比如workflow1挂了没删锁，或者Coordinator状态异常，不然会导致另一个workflow永远等下去。可以加个超时逻辑，或者定时清理过期锁。
• 权限问题：要确保执行workflow的用户有Oozie查询权限，以及HDFS锁路径的读写权限。
• 等待间隔：别把等待间隔设得太短，不然会频繁查状态浪费资源；也别太长，不然会耽误业务。一般10-15分钟就挺合适的。

内容的提问来源于stack exchange，提问作者Prasad mangalagiri