Loading multiple target tables in single interface (mapping) in Oracle Data Integrator

Loading multiple target tables in single interface (mapping) in Oracle Data Integrator

ODI 12c is released and it has the feature of loading mutiple target tables as part of a single interface. Actually interface is termed as mapping in ODI 12c. So i will refer as mapping going forward in this post.

In this post, I will show how to load all rows of source into multiple target tables. And also based on some condition, how to load few rows of source to one target and other rows into other target. It is only the split condition which decides which rows goes to which target table. 

Lets get started.

1. Load Multilple target tables in a mapping

I am going to use the HR.EMPLOYEES table as source and create two new tables to capture sub sets of columns from the employees table. I call emp_basic and emp_advanced. Here are the DDLs for both source and target tables.

Source:

CREATE TABLE "HR"."EMPLOYEES"

  (

    "EMPLOYEE_ID"  NUMBER(6,0),

    "FIRST_NAME"   VARCHAR2(20 BYTE),

    "LAST_NAME"    VARCHAR2(25 BYTE) CONSTRAINT "EMP_LAST_NAME_NN" NOT NULL ENABLE,

    "EMAIL"        VARCHAR2(25 BYTE) CONSTRAINT "EMP_EMAIL_NN" NOT NULL ENABLE,

    "PHONE_NUMBER" VARCHAR2(20 BYTE),

    "HIRE_DATE" DATE CONSTRAINT "EMP_HIRE_DATE_NN" NOT NULL ENABLE,

    "JOB_ID"         VARCHAR2(10 BYTE) CONSTRAINT "EMP_JOB_NN" NOT NULL ENABLE,

    "SALARY"         NUMBER(8,2),

    "COMMISSION_PCT" NUMBER(2,2),

    "MANAGER_ID"     NUMBER(6,0),

    "DEPARTMENT_ID"  NUMBER(4,0),

    CONSTRAINT "EMP_SALARY_MIN" CHECK (salary > 0) ENABLE,

    CONSTRAINT "EMP_EMAIL_UK" UNIQUE ("EMAIL"),

    CONSTRAINT "EMP_EMP_ID_PK" PRIMARY KEY ("EMPLOYEE_ID"),

    CONSTRAINT "EMP_DEPT_FK" FOREIGN KEY ("DEPARTMENT_ID") REFERENCES "HR"."DEPARTMENTS" ("DEPARTMENT_ID") ENABLE,

    CONSTRAINT "EMP_JOB_FK" FOREIGN KEY ("JOB_ID") REFERENCES "HR"."JOBS" ("JOB_ID") ENABLE,

    CONSTRAINT "EMP_MANAGER_FK" FOREIGN KEY ("MANAGER_ID") REFERENCES "HR"."EMPLOYEES" ("EMPLOYEE_ID") ENABLE

  )

Targets:

CREATE TABLE "DATA_TARGET"."EMP_BASIC"

  (

    "EMP_ID"     NUMBER(16,0) NOT NULL ENABLE,

    "FIRST_NAME" VARCHAR2(50 BYTE),

    "LAST_NAME"  VARCHAR2(50 BYTE)

  )

CREATE TABLE "DATA_TARGET"."EMP_ADV"

  (

    "EMP_ID"     NUMBER(16,0) NOT NULL ENABLE,

    "JOB_ID"     VARCHAR2(20 BYTE),

    "SALARY"     NUMBER(16,0),

    "COMMISSION" NUMBER(16,0)

  )

Lets create the mapping in ODI 12c client.

Go to logical tab and drop the source table hr.employees. You can see two employees table instances with name employees and employees1 dropped there. It is because of the manager_id foreign constraint referring to the self table. ODI 12c client represents this constraint as self join here automatically.

Lets drop the two targets data_target.emp_basic and data_target.emp_adv.

Drag and drop the column names from source to target tables to do the mapping.

Lets review the physical tab for KM details. Select each target table in the physical tab to review the IKM and CKM used. As we can see, ODI uses oracle parallel hint to load the target tables which improves performance for appending the data.

You can also see that IKM and CKM are from Global Knowledge modules. By default ODI 12c client has few KM in global KMs.

Lets run the mapping and see the results.

Lets review the target tables data. we see that all rows from source is populated into both tables for different columns.

2. Splitting records into multiple target tables based on condition

Lets try to split rows from source into multiple targets based on some condition. rows which are matching the condition are going into the respective targets. 

Go to the logical tab and delete the lines from source to target and add split components from components.

connect source to split and split to multiple targets. while deleting the existing lines from source to target, odi will still keep the existing mappings.

Click on split component and in the split component properties, define condition in expression section for each target output1 an output2 so that only rows matching this condition from source are loaded to respective target.

The condition I have defined here are 

OUTPUT1 - Default.EMPLOYEES.EMPLOYEE_ID <= 109

OUTPUT2 - Default.EMPLOYEES.EMPLOYEE_ID >   109

so employee_id less than 109 goes to emp_basic table.

employee_id greater than 109 goes to emp_adv table.

Lets run the interface and check the results. Run is successful, lets verify the results in target tables.

emp_basic table has records till employee id 109.

emp_adv table has records from employee_id 110 as expected.

ODI 12c Installation and ODI 12c Repository upgrade from 11g

ODI 12c Installation and ODI 12c Repository upgrade

ODI 12c version is releases and it has come with lot of exciting features. You can find the new features that has been shipped with 12c version in the link ODI 12c new features of oracle documentation.

1. Multiple Target support - Now mappings can load multiple targets as part of the new flow.

2. Re-usable mappings - We can create mapping once and can be re-used across objects which is very good for huge mapping which involves xml, table, complex files.

3. Step by Step debugger - mapping, package, procedure and scenarios can be debugged step by step.

4. Standalone Agent with Weblogic - Standalone agents can be managed through weblogic management framework

5. XML Improvements - lis, union, substition group, mixed context and annotation are supported in 12c version for XML technology.

Oracle's go-forward strategy is to have data warehousing/data integration to be done through Oracle data integrator and phase-out Oracle Warehouse builder to ODI by providing migration tools. Based on this strategy lot of integration from OWB and ODI features introduced as well as OWB migration to ODI is also made easy.

In this post, I would like to show my experience on 12c ODI installation and upgrade of the repository from 11g to 12c. 

Download the 12c ODI generic version from oracle downloads. I have downloaded 12.1.2 generic version which is the current version when this post is written.

ofm_odi_generic_12.1.2.0.0_disk1_1of1.zip

Extract this zip file. we can see the main jar for installation and patches zip file to apply the patches. Extract the patch zip file. ODI 12c installation has 2 steps.

1. Install ODI 12c jar

2. Apply the patches

Installing ODI 12c

Run the jar file in command prompt for installation. we need to run this jar with admin rights in windows (run as administrator option). Also ODI 12c is certified above 1.7.0_15 or above. Be sure to use latest JDK 1.7.

In windows, to run the command with admin rights, type cmd in run and use cntrl-shift-enter to launch it. select yes in the confirmation for running the commands with admin privilege in the cmd window.

 type the command java -jar odi_121200.jar (make sure java executable is from jdk 7 latest version)

Provide oracle home for the installation

Select enterprise installation option. Complete the wizard to complete the installation.

Apply ODI 12c patches

Read the readme file present in the patch directory for pre-requisites which may be different for your configuration before running opatch.

Verify the OUI inventory before running opatch. It needs to run without errors before applying the patches. Run the command "opatch lsinventory" to check this.

Go to the unzipped directory of odi patches and run the command "opatch napply odi_1212_opatch" to apply the patches. Ignore the error "There is an error with library regeneration" which is coming in the command prompt. Patch readme file says it is as expected and we can ignore it.

Lets check the OUI inventory again to check the applied patches.

Upgrading ODI repository from 11g to 12c

Before upgrading the repository, please make sure of the following so that you dont lose your repository if there are any errrors during upgrade.

1. All affected data is backed up (middleware home, oracle home, master and work repository schemas and data)

2. Datbase version is certified to work with ODI 12c

3. Certification and system requirements are met.

To run the upgrade, Go to <Oracle_Home>/oracle_common/upgrade/bin directory and run the ua command.

Select schemas to upgrade repository.

Select ODI for schema for repository upgrade.

Check the prerequiresites and check them to agree that you have completed them.

Provide sysdba user and master repository schema details.

Select all options for upgrading.

Provide supervisor user details

Please note down the upgrade key detail for future reference and work with support for any issues.

Complete the wizard to complete the upgrade. It will take some time to do the upgrade and confirm the upgrade is successful.

Lets connect to the repository using the new 12c ODI client.

ODI client  odi.exe  or odi64.exe can be found in <Oracle_Home>/odi/studio. Lauch the exe for odi studio client.

Oracle has made the look and feel same for Jdeveloper 12c and ODI 12c client. It is evident that OFM is standardized across tools and infrastructure in 12c.

If asked for import earlier version configs, select yes to carry them to 12c client. You will asked to create wallet for storing passwords. Provide a password of your choice for wallet password. It will be prompted for this wallet password while connecting to repository.

Lets connect to our upgraded repository. Select connect to repository and provide wallet password.

It automatically shows all existing repository connections for you to select and provide password. select the work repository and provide password.

Here is how upgraded repository objects looks. we can see that 11g interfaces are renamed to mapping in 12c (not suprisingly, interface is confusing and mapping term looks straightforward).

we can also see that UI and layout has been revamped in the 12c version. 

I will talk about 12c ODI new features with hands on in my next  posts.

Thank You and share your comments if any.

Running Scenarios asynchronously/parallel in ODI

Running Scenarios asynchronously/parallel in ODI 

In this post, I will demonstrate how to run interfaces, procedures or packages asynchronously inside a ODI package. 

By default for interfaces, procedures or packages, ODI doesn't provide an option to run them asynchronously (run it in parallel along with the initiating session) . To run them asynchronously, we need to covert interface, procedure or package to scenarios and scenarios can be run asynchronously inside a package or through command.

In this post, I will demonstrate how to asynchronously run the scenarios through package as well as through command.

1. In my tutorial, I am going to use five interfaces which does table to table loading for regions, countries, locations, departments and employees from hr schema to data_target schema. 

2. I will convert these five interfaces to scenarios and run them asynchronously inside a package and through procedure command.

Following are the five interfaces, I will use in this tutorial. I will also use the Jython execution report procedure (created in my previous post  Jython execution report) to write the execution report to file after the parallel run.

Lets convert all the interfaces to scenarios so that we can run them asynchronously. Right click on each interface and select Generate scenario.

1. Running the scenarios asynchronously inside a package 

Lets create a package and drag and drop the generated scenarios into the package in sequence by connecting them with ok in the regions, countries, locations, departments and employees order. After employee, connect it to wait for child session and then to the execution report procedure.

wait for child session event detection will wait for all child sessions (asynchronous sessions) to finish before putting the control to the next step which execution report procedure here.

I have presented them vertically in the diagram since we are going to make them run in parallel. But we will connect them together with ok sequentially.

Lets configure each scenario in the package diagram to run asynchronously. Select each scenario and select the synchronous/asynchronous option to asynchronous for each scenario in the package. By default this mode is set to synchronous.

Also we can set options for the odi wait for child sessions event detection. If we don't specify a session id, it takes the current session id (which is the package in which it is running)

Keywords property can be used to wait for child sessions which match the selected keyword here. Keyword can be set for each scenario in its properties and we can group the scenarios with keywords and wait for child session will wait only for scenario child sessions which match the keyword.

Lets run the package and see the results. Run is successful.
We can see that child session is created for each asynchronous scenario run but no child session session created for the procedure, it is part of the parent session. Each parellel run is assigned a new session id which is child to the parent session id 29001.

Lets check the execution report to see the timing in which each interface triggered. we could see that all the scenarios started at the same time and waitforchildsession event detector waited until the last scenario parallel run has completed.

2. Running the scenarios asynchronously using command in procedure 

In this case, I want to solve a use case where we should be able to run any number of scenarios dynamically in parallel but we will come to know the scenarios only  at run time.

In this use case, a table will be created at run time with all the scenario names with versions which need to be run in parallel. We can simply read this table in procedure source section and run the scenario asynchronously in target section through procedure binding (executed the command in target section for each row from the source query).

This is how the dynamic table which stores the scenario name and version looks like.

Lets create the procedure with source and target sections.

Source runs the query on the dynamic table with scenario and version names.

For each from source query target start scenario command is executed in asynchronous mode for the scenario.

Lets run the procedure and verify the results. we can see that child session is created for each scenario and it ran in parallel.

Writing execution results to File using Jython in ODI

Writing execution results to File using Jython in ODI

In this post, I am going to show how to write files in customized format by leveraging Jython technology in ODI. Out of the box, ODI provides support for delimted, fixed length, xml and complex files. But if there is a need to produce pdf or custom formated output, we can use Jython to achieve.

To demonstrate this capability, I am taking the example of write execution details of ODI to a custom formated file. The file format will look like this following.

/*------------------------------------------------------------------------------------------------------------------------------------------------*/



                    EXECUTION REPORT FOR SCENARIO : WRITE_EXECUTION_REPORT  ( THU NOV 14 10:34:33 2013 )                    



/*------------------------------------------------------------------------------------------------------------------------------------------------*/



StepNo StepName                                   StepType               BeginningTime     EndTime           Duration      Status     Error_Msg



------ --------------------------------------- ------------------- --------------------- --------------------- ---------- -------------- ----------



0      int_file_table                          Interface           14-NOV-2013 10:34:11  14-NOV-2013 10:34:12  1          Success        None



1      write_execution_report                  Procedure           14-NOV-2013 10:34:12  None                  0          UNKNOWN        None



/*------------------------------------------------------------------------------------------------------------------------------------------------*/

Following are the steps I am going to follow to demonstrate this use case.

1. I will create a procedure which will connect to ODI internal tables and get the execution details for the current session_id

2. I am going to create a file_to_table interface just to have a success scenario and a procedure which raises PL/SQL exception to demonstrate the error scenario.

3. We will create a package and place the interfaces and procedure for success/error scenario and also have the execution report procedure we developed in first step as a last steps for success as well as error scenario to extract the execution details to file with formatting.

Before we write the procedure, Lets understand the ODI internal tables a bit on how they store the execution details.

1. SNP_SESSION - stores the session_id and detailed related to session

2. SNP_SESS_STEP - All interface, package, scenario  and procedure details.

3. SNP_STEP_LOG - This has the steps within each odi object.

Just to put the table relation ships into perspective, showing the sample data for a sample session here. This data is for the below specified package in ODI.

SNP_SESSION  - Has one row per execution at the package level here.

SNP_SESS_STEP - represents objects inside a package with design time details.

SNP_STEP_LOG - represents objects inside a package with run time details (including error_message).

SNP_SESS_TASK - represents the KM, details related to interface,procedure, tools in a package with design time details .

SNP_SESS_TASK_LOG - represents the KM, details related to interface,procedure, tools in a package with run time details (including error message)

In the execution report, I developed I am not showing task level details, instead it stops at the step level. But I wanted to show here that task level details can also be captured in the log.

Execution Report Procedure

Create a procedure and add one command in the details with the following Jython code. Make sure that Jython technology is selected in the target tab. This procedure should be added as a last step in the package so that it takes all the execution details and puts it into formatted file.

This can be used for successful as well as for error conditions for producing the log.

Please find below the complete code for using it in your implementation.

1. It take the connection for current work repository
2. Using the current session_no to retrieve execution details using the query
3. Spools the query data with formatting to a file.

import string
import java.sql as sql
import java.lang as lang
import re
import time
sourceConnection = odiRef.getJDBCConnection("WORKREP")
output_write=open('c:/temp/Execution_Report_<%=odiRef.getSession("SESS_NO")%>.out','w')
sqlstring = sourceConnection.createStatement()
localtime = time.asctime( time.localtime(time.time()) )
print >> output_write, "/*------------------------------------------------------------------------------------------------------------------------------------------------*/\n"
print >> output_write,'%s\n' % (str('  Execution Report for scenario : '+'<%=odiRef.getSession("SESS_NAME")%>'+'  ( '+localtime+' ) ').upper().center(124))
print >> output_write, "/*------------------------------------------------------------------------------------------------------------------------------------------------*/\n"

sqlstmt="select rownum, \
       sess.sess_no, \
       sess.sess_name, \
       step_log.nno, \
       step.step_name, \
       case when step.step_type='F' THEN 'Interface' \
            when step.step_type='VD' THEN 'Variable declaration' \
            when step.step_type='VS' THEN 'Set//Increment variable' \
            when step.step_type='VE' THEN 'Evaluate variable' \
            when step.step_type='V' THEN 'Refresh variable '\
            when step.step_type='T' THEN 'Procedure' \
            when step.step_type='OE' THEN 'OS command' \
            when step.step_type='SE' THEN 'ODI Tool' \
            when step.step_type='RM' THEN 'Reverse-engineer model' \
            when step.step_type='CM' THEN 'Check model' \
            when step.step_type='CS' THEN 'Check sub-model' \
            when step.step_type='CD' THEN 'Check datastore' \
            when step.step_type='JM' THEN 'Journalize model' \
            when step.step_type='JD' THEN 'Journalize datastore' \
            ELSE 'UNKNOWN' END as step_type, \
      to_char(step_log.step_beg,'DD-MON-YYYY HH24:MI:SS') as step_beg, \
      to_char(step_log.step_end,'DD-MON-YYYY HH24:MI:SS') as step_end, \
      step_log.step_dur, \
      case when step_log.step_status='D' THEN 'Success' \
            when step_log.step_status='E' THEN 'Error' \
            when step_log.step_status='Q' THEN 'Queued' \
            when step_log.step_status='W' THEN 'Waiting' \
            when step_log.step_status='M' THEN 'Warning' \
            ELSE 'UNKNOWN' END as step_status, \
      step_log.nb_row, \
      step_log.nb_ins, \
      step_log.nb_upd, \
      step_log.nb_del, \
      step_log.nb_err, \
     dbms_lob.substr( TRANSLATE ( step_log.error_message, 'x'||CHR(10)||CHR(13), 'x'), 600, 1 )  as error_message \
from snp_session sess right outer join snp_sess_step step on sess.sess_no = step.sess_no \
                      inner join snp_step_log step_log on step.sess_no = step_log.sess_no and step.nno = step_log.nno \
where step_log.sess_no  = <%=odiRef.getSession("SESS_NO")%> \
      order by step_log.nno"
print >> output_write, "StepNo StepName                                StepType            BeginningTime  EndTime        Duration   Status  Error_Msg\n"
print >> output_write, "------ --------------------------------------- ------------------- --------------------- --------------------- ---------- -------------- ----------\n"
result=sqlstring.executeQuery(sqlstmt)
while (result.next()):
    rownum=result.getInt("rownum")
    nno=result.getInt("nno")
    step_name=result.getString("step_name")
    step_type=result.getString("step_type")
    step_beg=result.getString("step_beg")
    step_end=result.getString("step_end")
    step_dur=result.getInt("step_dur")
    step_status=result.getString("step_status")
    error_message=result.getString("error_message")
    print >> output_write,'%7s%40s%20s%22s%22s%11s%15s%s\n' % (str(nno).ljust(7),step_name.ljust(40),step_type.ljust(20),str(step_beg).ljust(22),str(step_end).ljust(22),str(step_dur).ljust(11),step_status.ljust(15),str(error_message))
print >> output_write, "/*------------------------------------------------------------------------------------------------------------------------------------------------*/\n"
sourceConnection.close()
output_write.close() 

The only restriction in this method is that we will not able to get the status of the execution report into the file since query is executed while it is run.

If we don't need this procedure details, we can restrict it in the where clause so that you wont see it in the report.

Sample Examples.

I will try to show a successful run as well as error run samples files here.

1. In this package, I am running an interface and on successful run, it produces the log.

/*------------------------------------------------------------------------------------------------------------------------------------------------*/

                    EXECUTION REPORT FOR SCENARIO : WRITE_EXECUTION_REPORT  ( THU NOV 14 10:34:33 2013 )                    

/*------------------------------------------------------------------------------------------------------------------------------------------------*/

StepNo StepName                                StepType            BeginningTime  EndTime        Duration   Status  Error_Msg

------ --------------------------------------- ------------------- --------------------- --------------------- ---------- -------------- ----------

0      int_file_table                          Interface           14-NOV-2013 10:34:11  14-NOV-2013 10:34:12  1          Success        None

1      write_execution_report                  Procedure           14-NOV-2013 10:34:12  None                  0          UNKNOWN        None

/*------------------------------------------------------------------------------------------------------------------------------------------------*/

2. For a error run a sample report is shown here. In this case, I run the interface, run a PL/SQL block which raises exception to produce error and on error run the execution report.

Lets look at the log for the error.

/*------------------------------------------------------------------------------------------------------------------------------------------------*/

                    EXECUTION REPORT FOR SCENARIO : WRITE_EXECUTION_REPORT  ( THU NOV 14 14:50:17 2013 )                    

/*------------------------------------------------------------------------------------------------------------------------------------------------*/

StepNo StepName                                StepType            BeginningTime  EndTime        Duration   Status  Error_Msg

------ --------------------------------------- ------------------- --------------------- --------------------- ---------- -------------- ----------

0      int_file_table                          Interface           14-NOV-2013 14:50:08  14-NOV-2013 14:50:15  7          Success        None

1      proc_call_stored_procedure              Procedure           14-NOV-2013 14:50:15  14-NOV-2013 14:50:16  1          Error          ODI-1226: Step proc_call_stored_procedure fails after 1 attempt(s).ODI-1232: Procedure proc_call_stored_procedure execution fails.ODI-1228: Task proc_call_stored_procedure (Procedure) fails on the target ORACLE connection oracle_xe_hr.Caused By: java.sql.SQLException: ORA-20001: This is a custom errorORA-06512: at "HR.RAISE_EXCEPTION", line 15ORA-06512: at line 2 at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:457) at oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:405) at oracle.jdbc.driver.T4C8Oall.processError(T4C8Oall.java:889) at oracle.jdbc.driver.T4CTTIfun.receive(T

2      write_execution_report                  Procedure           14-NOV-2013 14:50:16  None                  0          UNKNOWN        None

/*------------------------------------------------------------------------------------------------------------------------------------------------*/