目录

创建java项目

添加依赖

自定义action

定义schema

项目打包

oozie-site.xml添加配置

job.properties

workflow.xml

测试

---

本文以oozie-5.2.0.tar.gz为例。

创建java项目

使用IDEA开发工具创建一个java项目，目录结构如图：

 

添加依赖

jdom-1.1.jar
oozie-client-5.2.0.jar
oozie-core-5.2.0.jar

这三个包在oozie安装目录的lib目录下。

 

自定义action

代码如下：

package com.leboop.www;

import org.apache.oozie.action.ActionExecutor;
import org.apache.oozie.action.ActionExecutorException;
import org.apache.oozie.client.WorkflowAction;
import org.apache.oozie.util.XmlUtils;
import org.jdom.Element;
import org.jdom.JDOMException;
import org.jdom.Namespace;

import java.io.PrintWriter;
import java.sql.*;
import java.util.HashSet;
import java.util.Set;


/**
 * Created by leboop on 2020/6/21.
 */
public class SyncMysqlActionExecutor extends ActionExecutor {
    // 自定义节点类型
    private static final String ACTION_TYPE = "sync-mysql";
    // 自定义节点命名空间
    private static final String NAMESPACE = "uri:oozie:sync-mysql-action:0.1";
    // 自定义节点的元素
    private static final String JDBC_URL = "jdbc-url";
    private static final String SQL = "sql";
    private static final String SQL_OUTPUT_FILE_PATH = "sql-output-file-path";
    // mysql驱动
    private static final String DRIVER = "com.mysql.jdbc.Driver";
    private static Set<String> COMPLETED_STATUS = new HashSet<String>();

    // 无参构造
    public SyncMysqlActionExecutor() {
        super(ACTION_TYPE);
    }

    // 静态块
    static {
        COMPLETED_STATUS.add("SUCCEEDED");
        COMPLETED_STATUS.add("KILLED");
        COMPLETED_STATUS.add("FAILED");
        COMPLETED_STATUS.add("FAILED_KILLED");
    }

    @Override
    public void initActionType() {
        super.initActionType();
    }

    public void start(Context context, WorkflowAction workflowAction) throws ActionExecutorException {
        Element actionXml = null;
        try {
            actionXml = XmlUtils.parseXml(workflowAction.getConf());
            Namespace ns = Namespace.getNamespace(NAMESPACE);

            String jdbcUrl = actionXml.getChildTextTrim(JDBC_URL, ns);
            String sql = actionXml.getChildTextTrim(SQL, ns);
            String sqlOutputFilePath = actionXml.getChildTextTrim(SQL_OUTPUT_FILE_PATH, ns);

            runMysql(jdbcUrl, sql, sqlOutputFilePath);

            context.setExecutionData("OK", null);

        } catch (JDOMException e) {
            e.printStackTrace();
        }
    }

    public void end(Context context, WorkflowAction workflowAction) throws ActionExecutorException {
        if (workflowAction.getExternalStatus().equals("OK")) {
            context.setEndData(WorkflowAction.Status.OK, WorkflowAction.Status.OK.toString());
        } else {
            context.setEndData(WorkflowAction.Status.ERROR, WorkflowAction.Status.ERROR.toString());
        }
    }

    public void check(Context context, WorkflowAction workflowAction) throws ActionExecutorException {
        throw new UnsupportedOperationException();
    }

    public void kill(Context context, WorkflowAction workflowAction) throws ActionExecutorException {
        context.setEndData(WorkflowAction.Status.KILLED, "ERROR");
    }


    public boolean isCompleted(String s) {
        return COMPLETED_STATUS.contains(s);
    }


    private void runMysql(String jdbcUrl, String sql, String sqlOutputFilePath) throws ActionExecutorException {
        Connection connect = null;
        Statement statement = null;
        ResultSet resultSet = null;

        try {
            Class.forName(DRIVER);
            connect = DriverManager.getConnection(jdbcUrl);
            statement = connect.createStatement();
            resultSet = statement.executeQuery(sql);
            writeResultSet(resultSet, sqlOutputFilePath);
        } catch (Exception e) {
            throw convertException(e);
        } finally {
            try {
                if (resultSet != null)
                    resultSet.close();
                if (statement != null)
                    statement.close();
                if (connect != null)
                    connect.close();
            } catch (Exception e) {
                throw convertException(e);
            }
        }

    }

    private void writeResultSet(ResultSet resultSet, String sqlOutPutFilePath) throws Exception {
        PrintWriter out;
        if (sqlOutPutFilePath != null && sqlOutPutFilePath.length() > 0) {
            out = new PrintWriter(sqlOutPutFilePath);
        } else {
            out = new PrintWriter(System.out);
        }

        ResultSetMetaData metaData = resultSet.getMetaData();
        while (resultSet.next()) {
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                out.println(metaData.getColumnName(i) + "=" + resultSet.getString(i));
            }
        }
        out.close();
    }

}

 

定义schema

sync_mysql-0.1.xsd 如下：

<?xml version="1.0" encoding="UTF-8"?>
<!--
  Licensed to the Apache Software Foundation (ASF) under one
  or more contributor license agreements.  See the NOTICE file
  distributed with this work for additional information
  regarding copyright ownership.  The ASF licenses this file
  to you under the Apache License, Version 2.0 (the
  "License"); you may not use this file except in compliance
  with the License.  You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.
-->
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"
           xmlns:sync-mysql="uri:oozie:sync-mysql-action:0.1" elementFormDefault="qualified"
           targetNamespace="uri:oozie:sync-mysql-action:0.1">

    <xs:element name="sync-mysql" type="sync-mysql:sync_mysql_type"/>
    <xs:complexType name="sync_mysql_type">
        <xs:sequence>
            <xs:element name="jdbc-url" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="sql" type="xs:string" minOccurs="1" maxOccurs="1"/>
            <xs:element name="sql-output-file-path" type="xs:string" minOccurs="0" maxOccurs="1"/>
        </xs:sequence>
    </xs:complexType>
</xs:schema>

 

项目打包

打瘦包即可，如图：

将包拷贝到oozie安装目录lib下。

 

oozie-site.xml添加配置

内容如下：

<property>
     <name>oozie.service.ActionService.executor.ext.classes</name>
     <value>com.leboop.www.SyncMysqlActionExecutor</value>
 </property>
 <property>
     <name>oozie.service.SchemaService.wf.ext.schemas</name>
     <value>sync_mysql-0.1.xsd</value>
 </property>
重启oozie server。

 

job.properties

nameNode=hdfs://single:9000
jobTracker=single:8032
queueName=default
examplesRoot=jobs
oozie.use.system.libpath=true
oozie.wf.application.path=${nameNode}/user/root/${examplesRoot}/sync-mysql
jdbcURL=jdbc:mysql://192.168.128.1:3306/oozie?user=root&password=root
sql=select * from oozie_sys;
SQL_OUTPUT_PATH=/out/result.txt

需要在Linux上创建/out/result.txt文件。

 

workflow.xml

<?xml version="1.0" encoding="UTF-8"?>
<!--
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at
 http://www.apache.org/licenses/LICENSE-2.0
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
-->
<workflow-app xmlns="uri:oozie:workflow:0.5" name="sync-mysql-wf">
        <start to="sync-mysql-node"/>
        <action name="sync-mysql-node">
                <sync-mysql xmlns="uri:oozie:sync-mysql-action:0.1">
                        <jdbc-url>${jdbcURL}</jdbc-url>
                        <sql>${sql}</sql>
                        <sql-output-file-path>${SQL_OUTPUT_PATH}</sql-output-file-path>
                </sync-mysql>
                <ok to="end"/>
                <error to="fail"/>
        </action>
        <kill name="fail">
                <message>syncMysql failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
        </kill>
        <end name="end"/>
</workflow-app>

需要将workflow.xml拷贝到job.properties配置的oozie.wf.application.path路径下。

 

测试

oozie job -config job.properties -run

如图：

查看日志：

oozie job -info 0000004-200621162902099-oozie-root-W

如图：

 最后会将查询oozie数据库中oozie_sys表的内容：

输出到/out/result.txt文件中，内容如下：

name=db.version
data=4
name=oozie.version
data=5.2.0