informaica是一个很强大的ETL工具。其WORKFLOW MANAGER负责对ETL调度流程进行设计与管理和执行！
informatica在在资料库中提供以下表来存储调动流程的相关信息。
以便WORKFLOW MANAGER对用户所设计的调动流程进行管理和执行。
opb_wflow_dep:描述workflow执行步骤相关信息和每个步骤执行的条件信息
opb_wflow_dep_run：描述workflow执行步骤运行时相关信息
opb_wflow_expr ：描述workflow中相关的表达式或条件的相关信息
opb_wflow_perval：描述workflow可持续性变量相关信息
opb_wflow_run:描述workflow运行日志相关信息
opb_wflow_var:描述workflow变量相关信息
opb_task:描述任务对象的基本信息
opb_task_attr:描述任务对象相关的属性的信息
opb_task_inst:描述任务对象实例的基本信息
opb_task_inst_run:描述任务对象实例运行日志相关信息
opb_task_val_list:描述任务对象实例中command信息
WORKFLOW MANAGER系统中常用的有这几个模块，
Command模块，
Session模块，
Waiting_Event模块，
Raising_Event模块，
Assignment模块，
Worklet模块
WORKFLOW MANAGER系统中上述的这些模块统称为任务(Task).如果你对一个模块进行了
复制后新的模块就称作该任务的任务实例（Task_Inst）.
WORKFLOW MANAGER系统中Worklet模块可以有其他非Worklet模块组成。
在WORKFLOW MANAGER系统中一个工资流被称作Workflow,Workflow由各种任务模块组合而成。
同时一个Workflow也是一个任务。
以下是WORKFLOW元数据表的详细说明，
-----------------------------------------------------------------------
--OPB_WFLOW_DEP:描述workflow执行步骤相关信息和每个步骤执行的条件信息
--WORKFOLW MANAGER系统将用户设计的工作流的流程（步骤）信息存放到该表中。
--将每个步骤的执行的条件详细信息存放到opb_wflow_expr中，然后由CONDITION_ID
--关联。如果用户没有设置步骤执行条件则opb_wflow_expr对应记录中的CONDITION
--字段值为空。WORKFOLW MANAGER系统执行一个工作流时，就根据这个表中的步骤信息
--获得所要调度的任务实例标识，然后到opb_task_inst获取具体信息。
-----------------------------------------------------------------------
CREATE TABLE OPB_WFLOW_DEP (
WORKFLOW_ID NUMBER NULL,--工作流步骤归属的工作流标识
FROM_INST_ID NUMBER NULL,--工作流步骤中起始任务实例标识
TO_INST_ID NUMBER NULL,--工作流步骤中终止任务实例标识
CONDITION_ID NUMBER NULL,--工作流步骤可以执行的条件标识
OPB_OBJECT_ID NUMBER NULL
);
-----------------------------------------------------------------------
--OPB_WFLOW_DEP_RUN：描述workflow执行步骤运行时相关信息
--用于关联opb_wflow_run和opb_task_inst_run
--这样这三张表就能完整的描述一个工作流的运行情况了。
-----------------------------------------------------------------------
CREATE TABLE OPB_WFLOW_DEP_RUN (
WORKFLOW_ID NUMBER NULL,--工作流步骤归属的工作流标识
WORKFLOW_RUN_ID NUMBER NULL,--工作流步骤归属的工作流运行实例标识
WORKLET_RUN_ID NUMBER NULL,--工作流步骤归属的WORKLET运行实例标识
FROM_INSTANCE_ID NUMBER NULL,--工作流步骤中起始任务实例标识
TO_INSTANCE_ID NUMBER NULL --工作流步骤可以执行的条件标识
);
-----------------------------------------------------------------------
--OPB_WFLOW_EXPR：描述workflow中相关的表达式或条件的相关信息
--
-----------------------------------------------------------------------
CREATE TABLE OPB_WFLOW_EXPR (
WORKFLOW_ID NUMBER NULL,--表达式归属的工作流标识
CONDITION_ID NUMBER NULL,--条件标识
LINE_NO NUMBER NULL,
CONDITION VARCHAR2(2000) NULL,--条件表达式描述
OPB_OBJECT_ID NUMBER NULL
);
-----------------------------------------------------------------------
--OPB_WFLOW_PERVAL：描述workflow可持续性变量相关信息
--当下次运用到一持续性变量时，WORKFLOW MANAGER系统就会从该表中读取
--该变量的VAR_VALUE值了。当该持续性变量所在的WORKFLOW(或WORKLET)结束后
--系统就会将该持续性变量的当前值保存到该表相应记录的VAR_VALUE字段中
-----------------------------------------------------------------------
CREATE TABLE OPB_WFLOW_PERVAL (
SUBJECT_ID NUMBER NULL,--工作流归属的主题标识
WORKFLOW_ID NUMBER NULL,--变量归属的工作流标识
WORKLET_ID NUMBER NULL,
VARIABLE_ID NUMBER NULL,--变量标识
VAR_VALUE VARCHAR2(2000) NULL,--变量当前值
LAST_SAVED VARCHAR2(30) NULL,
BIT_OPTIONS NUMBER NULL,
OPB_OBJECT_ID NUMBER NULL,
WORKLET_INST_ID NUMBER NULL--变量归属的WORKLET任务实例标识
);
-----------------------------------------------------------------------------
--OPB_WFLOW_RUN：描述workflow运行日志相关信息
--该表中描述运行WROKFLOW的运行环境信息（SERVER_ID、USER_NAME）
--和WORKFLOW的运行结果。
--OPB_WFLOW_RUN和OPB_TASK_INST_RUN这两个表提供了WORKFLOW MONITER系统
--对显示工作流详细运行状态的所需要的所有信息
------------------------------------------------------------------------------
CREATE TABLE OPB_WFLOW_RUN (
SUBJECT_ID NUMBER NULL,--工作流归属主题标识
WORKFLOW_ID NUMBER NULL,--工作流标识
WORKFLOW_RUN_ID NUMBER NULL,--工作流运行实例标识
WORKFLOW_NAME VARCHAR2(240) NULL,--工作流名称
SERVER_ID NUMBER NULL,--运行工作流的INFOMATICA服务器标识
SERVER_NAME VARCHAR2(240) NULL,--运行工作流的INFOMATICA服务器名称
START_TIME DATE NULL,--开始运行工作流时间
END_TIME DATE NULL,--结束运行工作流时间
LOG_FILE VARCHAR2(2000) NULL,--记录工作流运行详细情况的日志文件全名称
RUN_ERR_CODE NUMBER NULL,--工作流运行错误代码（如果工作流正常结束该字段值为空）
RUN_ERR_MSG VARCHAR2(2000) NULL,--工作流运行错误描述（如果工作流正常结束该字段值为空）
RUN_STATUS_CODE NUMBER NULL,--工作流运行状态
USER_NAME VARCHAR2(240) NULL,--运行该工作流所使用的用户名称
RUN_TYPE NUMBER NULL,--工作流运行类型
CODEPAGE_ID NUMBER NULL
);
---------------------------------------------------------------------------------------
--OPB_WFLOW_VAR:描述workflow变量相关信息
--在WORKFLOW MANAGER系统中有系统内值变量和用户自定义变量。
--
----------------------------------------------------------------------------------------
CREATE TABLE OPB_WFLOW_VAR (
SUBJECT_ID NUMBER NULL,--工作流归属主题标识
WORKFLOW_ID NUMBER NULL,--变量归属的工作流任务实例标识(或者变量归属的WORKLET任务实例标识)
VARIABLE_ID NUMBER NULL,--变量标识
VAR_NAME VARCHAR2(240) NULL,--变量名称
VAR_TYPE NUMBER NULL,--变量类型（临时、持久）
VAR_DESC VARCHAR2(2000) NULL,--变量描述
VAR_DATATYPE NUMBER NULL,--变量数据类型
VAR_DEFAULT VARCHAR2(2000) NULL,--变量缺省值
LAST_SAVED VARCHAR2(30) NULL,
OPB_OBJECT_ID NUMBER NULL,
TASK_INST_ID NUMBER NULL,--变量归属任务实例标识（该任务实例为非WORKFLOW/WORKLET任务实例）
TASK_INST_NAME VARCHAR2(240) NULL,--变量归属任务实例名称
BIT_OPTIONS NUMBER NULL
);
-------------------------------------------------------------------------
--OPB_TASK:描述任务对象实例的基本信息
--任务概论是INFORMAICA系统中对所有可以被调度执行的模块的总称，
--包含Command任务，Session任务，Waiting_Event任务，
--Raising_Event任务，Assignment任务，Worklet任务,Workflow任务等。
-------------------------------------------------------------------------
CREATE TABLE OPB_TASK (
SUBJECT_ID NUMBER NULL,--任务归属主题标识
TASK_ID NUMBER NULL,--任务标识
TASK_TYPE NUMBER NULL,--任务类型
TASK_NAME VARCHAR2(240) NULL,--任务名称
TASK_PROPERTY NUMBER NULL,--任务属性
IS_REUSABLE NUMBER NULL,--任务是否可以复用标识
IS_VALID NUMBER NULL,
IS_ENABLED NUMBER NULL,
COMMENTS VARCHAR2(2000) NULL,--任务描述
LAST_SAVED VARCHAR2(30) NULL,
CREATE_INFO NUMBER NULL,
OPB_OBJECT_ID NUMBER NULL,
OBJVERSION NUMBER NULL
);
--------------------------------------------------------------------------
--OPB_TASK_ATTR：描述任务实例对象相关的属性的信息
--因为每个任务可以有多个实例，而每个任务实例的属性也可以各自不同
--所以该表的主键应该是WORKFLOW_ID,INSTANCE_ID,ATTR_ID
---------------------------------------------------------------------------
CREATE TABLE OPB_TASK_ATTR (
WORKFLOW_ID NUMBER NULL,--任务所归宿的工作流标识
INSTANCE_ID NUMBER NULL,--任务实例标识
TASK_ID NUMBER NULL,--任务标识
TASK_TYPE NUMBER NULL,--任务类型
REF_SESSION_ID NUMBER NULL,
COMPONENT_TYPE NUMBER NULL,
ATTR_ID NUMBER NULL,--任务属性标识
ATTR_VALUE VARCHAR2(2000) NULL,--任务属性值
LINE_NO NUMBER NULL,
GROUP_ID NUMBER NULL,
OPB_OBJECT_ID NUMBER NULL
);
---------------------------------------------------------------------------
--OPB_TASK_INST：描述任务对象实例的基本信息
--在INFORMATICA的WORKFLOW MANAGER系统中，提供对任务对象提供复制。
--当一个任务对象被复制后，就在系统中产生一个实例对象，
--但是如果不被复制就不会产生实例对象。同时每个任务实例对象之间及
--任务实例对象和任务对象之间是相互独立的。为的就是信息复制目的。
--不要和JAVA中的对象和实例之间的关系等同起来。
----------------------------------------------------------------------------
CREATE TABLE OPB_TASK_INST (
WORKFLOW_ID NUMBER NULL,--任务实例所归属的工作流标识
INSTANCE_ID NUMBER NULL,--任务实例标识
TASK_ID NUMBER NULL,--任务标识
TASK_TYPE NUMBER NULL,--任务类型
INSTANCE_NAME VARCHAR2(240) NULL,--任务实例名称
TASK_PROPERTY NUMBER NULL,--任务属性
IS_ENABLED NUMBER NULL,
BIT_OPTIONS NUMBER NULL,
COMMENTS VARCHAR2(2000) NULL,--任务描述
IS_VALID NUMBER NULL,
OPB_OBJECT_ID NUMBER NULL
);
------------------------------------------------------------------
--OPB_TASK_INST_RUN:描述任务对象实例运行日志相关信息
--当运行一个工作流调度时，INFORMATICA系统会记录WORKFLOW中每个任务实例
--运行结果。在WORKFLOW MANAGER中每个任务实例对象的运行状态属性值就是
--从该表中获得的。
-------------------------------------------------------------------
CREATE TABLE OPB_TASK_INST_RUN (
SUBJECT_ID NUMBER NULL,--任务实例归属的主题标识
WORKFLOW_ID NUMBER NULL,--任务实例归属的工作流标识
WORKFLOW_RUN_ID NUMBER NULL,--工作流运行实例标识
WORKLET_RUN_ID NUMBER NULL,--WORKLET运行实例标识
CHILD_RUN_ID NUMBER NULL,
INSTANCE_ID NUMBER NULL,--任务实例标识
INSTANCE_NAME VARCHAR2(240) NULL,--任务实例名称
TASK_ID NUMBER NULL,--任务标识
TASK_TYPE NUMBER NULL,--任务类型
START_TIME DATE NULL,--任务实例开始运行时间
END_TIME DATE NULL,--任务实例结束运行时间
RUN_ERR_CODE NUMBER NULL,--任务实例运行错误代码（如果任务实例正常运行该字段值为空）
RUN_ERR_MSG VARCHAR2(2000) NULL,--任务实例运行错误信息（如果任务实例正常运行该字段值为空）
RUN_STATUS_CODE NUMBER NULL,--任务运行状态代码
TASK_NAME VARCHAR2(240) NULL--任务名称
);
-----------------------------------------------------------------------------------------
--OPB_TASK_VAL_LIST：描述任务对象实例中command信息
--该表主要应用于COMMAND任务，记录COMMAND任务的所有命令行信息。
--WORKFLOW MANAGER系统在执行该COMMAND任务时，按顺序逐一读取该任务的
--命令行，而后以阻塞方式运行该命令行。
--要注意的是：如果复制一个COMMAND任务对象产生了一个COMMAND任务实例对象，
--但是不改变COMMAND任务实例对象的命令行内容时则在OPB_TASK_VAL_LIST中不会出现
--该COMMAND任务实例对象的命令行信息记录的。
-----------------------------------------------------------------------------------------
CREATE TABLE OPB_TASK_VAL_LIST (
TASK_ID NUMBER NULL,--任务标识
SUBJECT_ID NUMBER NULL,--任务实例归属的主题标识
WORKFLOW_ID NUMBER NULL,--任务实例归属的工作流标识（或WORKLET标识）
INSTANCE_ID NUMBER NULL,--任务实例标识
SESSION_ID NUMBER NULL,
COMPONENT_TYPE NUMBER NULL,
TASK_TYPE NUMBER NULL,--任务类型（为COMMAND类型的任务)
DESCRIPTION VARCHAR2(2000) NULL,
PM_VALUE VARCHAR2(2000) NULL,--命令行内容
EXEC_ORDER NUMBER NULL,
VAL_NAME VARCHAR2(240) NULL,--代表命令行内容的变量名称
LINE_NO NUMBER NULL,
OPB_OBJECT_ID NUMBER NULL
);
举例：
下面是一个JAVA的类，该类中描述了如何根据WORKFLOW名称，WORKLET名称和变量名称
来获得和设置相应可持续性变量值的方法。
import java.sql.*;
import java.io.*;
public class varConfig
{
public varConfig(String peizhi,String wflname,String wltname,String varname)
{
this.peizhi=peizhi;
this.wflname=wflname;
this.wltname=wltname;
this.varname=varname;
File fs=new File(peizhi);
try
{
BufferedReader in=new BufferedReader(new FileReader(fs));
//取得配置表中的各个参数
String line="";
while((line=in.readLine())!=null)
{
if(line.trim().startsWith("etldsn"))
{
etldsn=line.substring(line.indexOf("=")+1).trim();
}
if(line.trim().startsWith("etluid"))
{
etluid=line.substring(line.indexOf("=")+1).trim();
}
if(line.trim().startsWith("etlpwd"))
{
etlpwd=line.substring(line.indexOf("=")+1).trim();
}
}
in.close();
}
catch(Exception e)
{
System.out.println(e);
}
}
public boolean init()
{
//根据workflow名称获得相应的workflowID
workflowID=GetWorkflowID(wflname);
System.out.println("workflowID="+workflowID);
if(workflowID.trim().equals(""))
{
return false;
}
if(!wltname.trim().equals("0"))
{
//根据workflowID和workflowlet名称获得workflowlet的ID
String tempId=GetWorkflowLetID(workflowID,wltname);
workflowletID=tempId.trim().substring(0,tempId.trim().indexOf("$"));
workletinstID=tempId.trim().substring(tempId.trim().indexOf("$")+1);
System.out.println("workflowletID="+workflowletID);
System.out.println("workletinstID="+workletinstID);
}
//根据workflowID和workflowletID和Var名称获得VarID
if(!workflowletID.trim().equals(""))
{
varID=GetVariableID(workflowletID,varname);
}
else
{
varID=GetVariableID(workflowID,varname);
}
System.out.println("varID="+varID);
return true;
}
public boolean setVarValue(String varvalue)
{
//根据workflowID和workflowletID和workletinstID和VarID修改相应变量的值
try
{
String SQL="";
CreateConn crtetlconn=new CreateConn(etldsn,etluid,etlpwd);
Connection etlconn=crtetlconn.backConn();
Statement etlwf=etlconn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_UPDATABLE);
if(!workflowletID.trim().equals(""))
{
SQL="update opb_wflow_perval set var_value= '"+varvalue+"' where workflow_id= '"+workflowletID+"' and worklet_inst_id='"+workletinstID+"' and variable_id='"+varID+"'";
}
else
{
SQL="update opb_wflow_perval set var_value= '"+varvalue+"' where workflow_id= '"+workflowID+"' and variable_id='"+varID+"'";
}
System.out.println("update="+SQL);
ResultSet rswf=etlwf.executeQuery(SQL);
etlwf.executeQuery("commit");
rswf.close();
etlwf.close();
etlconn.close();
return true;
}
catch(Exception e)
{
System.out.println(e);
return false;
}
}
public String getVarValue()
{
//获得指定VAR的值
try
{
String SQL="";
String varValue="";
CreateConn crtetlconn=new CreateConn(etldsn,etluid,etlpwd);
Connection etlconn=crtetlconn.backConn();
Statement etlwf=etlconn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_UPDATABLE);
if(!workflowletID.trim().equals(""))
{
SQL="select var_value from opb_wflow_perval where workflow_id= '"+workflowletID+"' and worklet_inst_id='"+workletinstID+"' and variable_id='"+varID+"'";
}
else
{
SQL="select var_value from opb_wflow_perval where workflow_id= '"+workflowID+"' and variable_id='"+varID+"'";
}
System.out.println("varvalue="+SQL);
ResultSet rswf1=etlwf.executeQuery(SQL);
while (rswf1.next())
{
varValue=rswf1.getString(1).trim();
}
rswf1.close();
etlwf.close();
etlconn.close();
return varValue;
}
catch(Exception e)
{
System.out.println(e);
return "";
}
}
public String GetWorkflowID(String wflname)
{
//根据workflow名称获得相应的workflowID
try
{
String SQL="";
String workflowID="";
CreateConn crtetlconn=new CreateConn(etldsn,etluid,etlpwd);
Connection etlconn=crtetlconn.backConn();
Statement etlwf=etlconn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_UPDATABLE);
SQL="select task_ID from opb_task where task_name ='" + wflname +"'";
ResultSet rswf2=etlwf.executeQuery(SQL);
while (rswf2.next())
{
workflowID=rswf2.getString(1).trim();
}
rswf2.close();
etlwf.close();
etlconn.close();
return workflowID;
}
catch(Exception e)
{
System.out.println(e);
return "";
}
}
public String GetWorkflowLetID(String WorkflowID,String WorkflowLetName)
{
//根据workflowID和workflowlet名称获得workflowlet的ID
try
{
String SQL="";
String workflowletID="";
String workletinstID="";
CreateConn crtetlconn=new CreateConn(etldsn,etluid,etlpwd);
Connection etlconn=crtetlconn.backConn();
Statement etlwf=etlconn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_UPDATABLE);
SQL="select task_id,INSTANCE_ID from opb_task_INST where workflow_id ='" + WorkflowID +"' and instance_name= '" + WorkflowLetName +"'" ;
System.out.println("wl="+SQL);
ResultSet rswf3=etlwf.executeQuery(SQL);
while (rswf3.next())
{
workflowletID=rswf3.getString(1).trim();
workletinstID=rswf3.getString(2).trim();
}
rswf3.close();
etlwf.close();
etlconn.close();
return workflowletID+"$"+workletinstID;
}
catch(Exception e)
{
System.out.println(e);
return "";
}
}
public String GetVariableID(String WorkflowID,String VariableName)
{
//根据workflowID和变量名称获得变量ID
try
{
String SQL="";
String varID="";
CreateConn crtetlconn=new CreateConn(etldsn,etluid,etlpwd);
Connection etlconn=crtetlconn.backConn();
Statement etlwf=etlconn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE,ResultSet.CONCUR_UPDATABLE);
SQL="select variable_id from opb_wflow_var where workflow_id ='" + WorkflowID +"' and var_name= '" +VariableName+"'";
System.out.println("Var="+SQL);
ResultSet rswf4=etlwf.executeQuery(SQL);
while (rswf4.next())
{
varID=rswf4.getString(1).trim();
}
rswf4.close();
etlwf.close();
etlconn.close();
return varID;
}
catch(Exception e)
{
System.out.println(e);
return "";
}
}
private String workflowID="";
private String varID="";
private String workflowletID="";
private String workletinstID="";
private String wflname="";
private String wltname="";
private String varname="";
private String peizhi="";
private String etldsn="";
private String etluid="";
private String etlpwd="";
}
联系客服
