老程序员博客

编程生涯的最后一舞

当前位置：首页 > 技术 > 正文内容

Oozie工作流引擎核心机制与实践指南

访客 技术 2026年7月8日 1

Oozie概述

Apache Oozie是一个基于有向无环图（DAG）的工作流调度系统，用于管理Hadoop作业的依赖关系与执行顺序。其核心由控制流节点（如start、end、fork、join）和动作节点（如MapReduce、Shell、Hive）构成，支持复杂业务场景的自动化编排。

官方文档：http://oozie.apache.org/

主要功能模块

  • Workflow：定义顺序执行或并行分支（fork/join）的作业流程。
  • Coordinator：基于时间或数据事件触发Workflow的定时执行。
  • Bundle：将多个Coordinator应用绑定为逻辑单元统一管理。
  • job.properties：集中配置全局参数，如HDFS路径、数据库连接等。

配置示例（Bundle）详见：参考链接

工作流生命周期

Oozie中的工作流作业存在以下状态转换：

状态说明
PREP作业已创建但未启动，处于预备阶段。
RUNNING作业处于执行中，可能因异常终止或挂起。
SUSPENDED运行中的作业被挂起，需恢复或杀死操作才能继续。
SUCCEEDED成功到达end节点，流程正常完成。
KILLED在PREP、RUNNING或SUSPENDED状态下被强制终止。
FAILED因未捕获异常导致作业异常终止。

控制流节点

控制节点负责管理流程的拓扑结构，包括：

  • start：工作流入口。
  • end：工作流结束点。
  • kill：强制终止并输出错误信息。
  • decision：基于条件选择执行路径（类似if-else）。
  • fork/join：并行分支与同步汇合。

常用动作节点

Ssh Action

<action name="ssh_task">
    <ssh>
        <host>user@192.168.1.100</host>
        <command>/usr/bin/python3</command>
        <args>/home/data/script.py</args>
        <args>--config=prod</args>
        <capture-output/>
    </ssh>
    <ok to="next_node"/>
    <error to="fail_node"/>
</action>

Email Action

<action name="alert_email">
    <email xmlns="uri:oozie:email-action:0.1">
        <to>admin@company.com</to>
        <subject>Job ${jobName} Failed on ${yesterday}</subject>
        <body>Error: ${wf:errorMessage(wf:lastErrorNode())}</body>
    </email>
    <ok to="end"/>
    <error to="end"/>
</action>

Shell Action（通过SSH执行）

<action name="run_shell">
    <ssh xmlns="uri:oozie:ssh-action:0.2">
        <host>192.168.10.241</host>
        <command>/bin/bash</command>
        <arg>${sshUri}/preparam.sh</arg>
        <arg>${mysql_host}</arg>
        <arg>${mysql_user}</arg>
        <capture-output/>
    </ssh>
    <ok to="next"/>
    <error to="fail"/>
</action>

Hive Action

<action name="hive_etl">
    <hive xmlns="uri:oozie:hive-action:0.2">
        <job-tracker>node01:8021</job-tracker>
        <name-node>node02:8020</name-node>
        <prepare>
            <delete path="${output_dir}"/>
        </prepare>
        <configuration>
            <property>
                <name>mapred.compress.map.output</name>
                <value>true</value>
            </property>
        </configuration>
        <script>transform.q</script>
        <param>Input=/data/source</param>
        <param>Output=${output_dir}</param>
    </hive>
    <ok to="post_process"/>
    <error to="cleanup"/>
</action>

Sqoop Action

<action name="sqoop_import">
    <sqoop xmlns="uri:oozie:sqoop-action:0.2">
        <job-tracker>node01:8021</job-tracker>
        <name-node>node02:8020</name-node>
        <arg>import</arg>
        <arg>--connect</arg>
        <arg>jdbc:mysql://dbhost:3306/mydb</arg>
        <arg>--table</arg>
        <arg>orders</arg>
        <arg>--target-dir</arg>
        <arg>hdfs://cluster/data/orders</arg>
        <arg>-m</arg>
        <arg>4</arg>
    </sqoop>
    <ok to="validate"/>
    <error to="cleanup"/>
</action>

FS Action（文件操作）

<action name="hdfs_ops">
    <fs>
        <delete path='/tmp/backup'/>
        <mkdir path='/data/archive'/>
        <move source='/data/tmp' target='/data/final'/>
        <chmod path='/data/final' permissions='755' dir-files='true'/>
        <touchz path='/data/.done'/>
    </fs>
    <ok to="next"/>
    <error to="fail"/>
</action>

Java Action

执行单一Map任务的自定义Java程序。

<action name="java_processor">
    <java>
        <job-tracker>cluster-rm:8032</job-tracker>
        <name-node>cluster-nn:8020</name-node>
        <main-class>com.example.DataTransformer</main-class>
        <java-opts>-Xmx2048m -Dconfig=prod</java-opts>
        <arg>--input</arg>
        <arg>hdfs://data/input</arg>
        <arg>--output</arg>
        <arg>hdfs://data/output</arg>
        <capture-output/>
    </java>
    <ok to="end"/>
    <error to="fail"/>
</action>

Sub-workflow（嵌入子流程）

<action name="child_workflow">
    <sub-workflow>
        <app-path>${wf_base}/etl_pipeline</app-path>
        <propagate-configuration/>
    </sub-workflow>
    <ok to="merge"/>
    <error to="fail"/>
</action>

常见调度文件模板

workflow.xml

<workflow-app xmlns="uri:oozie:workflow:0.5" name="${jobName}">
    <global>
        <job-tracker>${jobTracker}</job-tracker>
        <name-node>${nameNode}</name-node>
        <configuration>
            <property>
                <name>mapred.job.queue.name</name>
                <value>${queueName}</value>
            </property>
            <property>
                <name>oozie.launcher.mapred.job.queue.name</name>
                <value>${launcherQueue}</value>
            </property>
        </configuration>
    </global>
    <start to="step1"/>
    <action name="step1">
        <ssh xmlns="uri:oozie:ssh-action:0.2">
            <host>192.168.1.50</host>
            <command>/bin/bash</command>
            <arg>/app/scripts/init.sh</arg>
            <capture-output/>
        </ssh>
        <ok to="parallel_start"/>
        <error to="kill_job"/>
    </action>
    <fork name="parallel_start">
        <path start="sub_task_a"/>
        <path start="sub_task_b"/>
    </fork>
    <action name="sub_task_a">
        <sub-workflow>
            <app-path>${wf_base}/task_a</app-path>
            <propagate-configuration/>
        </sub-workflow>
        <ok to="join_point"/>
        <error to="kill_job"/>
    </action>
    <action name="sub_task_b">
        <sub-workflow>
            <app-path>${wf_base}/task_b</app-path>
            <propagate-configuration/>
        </sub-workflow>
        <ok to="join_point"/>
        <error to="kill_job"/>
    </action>
    <join name="join_point" to="final_step"/>
    <action name="final_step">
        <ssh>
            <host>192.168.1.50</host>
            <command>/bin/bash</command>
            <arg>${ssh_path}/postprocess.sh</arg>
            <capture-output/>
        </ssh>
        <ok to="end"/>
        <error to="kill_job"/>
    </action>
    <kill name="kill_job">
        <message>Job failed: ${wf:errorMessage(wf:lastErrorNode())}</message>
    </kill>
    <end name="end"/>
</workflow-app>

coordinator.xml

<coordinator-app name="${jobName}" frequency="0 30 1 * *" start="${start}" end="${end}" timezone="GMT+0800"
    xmlns="uri:oozie:coordinator:0.4">
    <action>
        <workflow>
            <app-path>${wfPath}</app-path>
            <configuration>
                <property>
                    <name>ds</name>
                    <value>${coord:formatTime(coord:dateOffset(coord:nominalTime(), -1, 'DAY'), 'yyyy-MM-dd')}</value>
                </property>
                <property>
                    <name>mysql_host</name>
                    <value>${mysql_host}</value>
                </property>
                <property>
                    <name>mysql_user</name>
                    <value>${mysql_user}</value>
                </property>
            </configuration>
        </workflow>
    </action>
</coordinator-app>

job.properties

nameNode=hdfs://mycluster
jobTracker=hadoop02:8042
queueName=default
launcherQueue=oozie
jobName=monthly_etl
mysql_host=192.168.1.200
mysql_user=etl_user
mysql_pwd=secret123
gn=prod
timezone=GMT+8
oozie.use.system.libpath=true
ssh_path=/app/scripts
wfPath=${nameNode}/user/oozie/workflows/etl
baseUri=${nameNode}/user/oozie/workflows/etl
oozie.coord.application.path=${wfPath}
start=2023-01-01T03:30+0800
end=2099-12-31T23:00+0800

EL表达式详解

常用时间运算表达式：

  • 计算昨天：${coord:formatTime(coord:dateOffset(coord:nominalTime(), -1, 'DAY'), 'yyyy-MM-dd')}
  • 计算上月：${coord:formatTime(coord:dateOffset(coord:nominalTime(), -1, 'MONTH'), 'yyyy-MM')}

注意：时间单位（DAY, MONTH）需大写。

表达式功能
${coord:minutes(n)}返回n分钟后的时间点
${coord:hours(n)}返回n小时后的时间点
${coord:days(n)}返回n天后的时间点
${coord:months(n)}返回n个月后的时间点
${coord:endOfDays(n)}返回n天后的当天最晚时间
${coord:endOfMonths(1)}返回当月最后一天的最晚时间
${coord:current(n)}返回第n个dataset实例的创建时间
${coord:dataIn(String)}解析输入事件的URI
${coord:dataOut(String)}解析输出事件的URI
${coord:offset(n, unit)}基于名义时间偏移n个单位（MINUTE/HOUR/DAY/MONTH/YEAR）
${coord:hoursInDay(n)}第n天的小时数（n可为负）
${coord:daysInMonth(n)}第n个月的天数（n可为负）
${coord:tzOffset()}时区偏移分钟数
${coord:latest(n)}最近第n个可用的dataset实例
${coord:future(n, limit)}未来第n个dataset实例（limit限制个数）
${coord:nominalTime()}基于频率计算的名义时间（如每天固定时刻）
${coord:actualTime()}动作实际创建的时间
${coord:user()}启动此协调器的用户名
${coord:dateOffset(base, instance, unit)}计算 base + instance * unit 的时间
${coord:formatTime(timestamp, pattern)}按pattern格式化时间戳

参考资料：详细说明

标签: Apache OozieWorkflowCoordinatorSqoop
返回列表

上一篇：AWK文本处理完全指南

下一篇：.NET Core 通用主机设计与实现机制解析

相关文章

Linux crontab 详解

1) crontab 是什么cron 是 Linux 的定时任务守护进程；crontab 是用来编辑/查看“按时间周期执行命令”的表（cron table）。常见两类：用户 crontab：每个用户一份（crontab -e 编辑）系统级 crontab / cron.d：可指定执行用户（/etc/crontab、/etc/cron.d/*）2) crontab 时间...

富文本里可以允许的 HTML 属性

一、所有标签默认允许的安全属性（极少）class        （可选）id           （通常建议禁用）title️ 注意：id 容易被滥用做锚点注入，很多系统直接禁用class 允许的话最好只允许固定前缀（如 editor-*）二、a 标签允许属性<a href="" t...

Mac 安装 Node.js 指南

方法一：通过官网安装包（最简单，适合初学者）如果你只是想快速安装并开始使用，这是最直接的方法。访问 Node.js 官网。页面会显示两个版本：LTS (Recommended For Most Users)：长期支持版，最稳定。建议选这个。Current：最新特性版，包含最新功能但可能不够稳定。下载 .pkg 安装包并运行。按照安装向导点击“下一步”即可完成。方法二：使用 Homebrew 安装（...

Dom\HTML_NO_DEFAULT_NS 的副作用：自动加闭合标签

在使用Dom\HTMLDocument时，Dom\HTML_NO_DEFAULT_NS 将禁止在解析过程中设置元素的命名空间, 此设置是为了与DOMDocument向后兼容而存在的。当使用它时，已知的一个副作用就是：自动加闭合标签例如 </img> 为什么会这样？当你使用：Dom\HTML_NO_DEFAULT_NS文档会变成 无命名空间模式，此时内部更接近 XML...

Laravel 事件和监听器创建

在 Laravel 中，使用 Artisan 命令创建 Events（事件） 和 Listeners（监听器） 是非常高效的。你可以通过以下几种方式来实现：1. 手动创建单个 Event如果你只想创建一个事件类，可以使用 make:event 命令：Bashphp artisan make:event UserRegistered执行后，文件将生成在 app/Even...

自定义域名解析神器 dnsmasq

什么是 dnsmasq？dnsmasq 是一个轻量级、功能强大的网络服务工具，专为小型和中等规模网络设计。它是一个综合的网络基础设施解决方案[1]。dnsmasq 能做什么？功能说明应用场景DNS 转发与缓存将 DNS 查询转发到上游服务器（ISP、Google DNS 等），并在本地缓存结果加快 DNS 查询速度，减少外部 DNS 流量本地 DNS解析本地网络设备的主机名，无需编辑&n...

发表评论

访客

◎欢迎参与讨论，请在这里发表您的看法和观点。

Copyright © agingcoder.cn

Powered By Z-BlogPHP. Theme by TOYEAN.