Creating a Custom Integration Knowledge Module to Backup Table before inserting new rows - ODI11g

Today I will be modifying an existing IKM SQL Control Append and implement + test it in ODI11g.

*This case study was demonstrated via ODI Substitution API in oracle's website, for further references direct to this link:*

https://docs.oracle.com/middleware/1212/odi/ODIKD/ikm.htm#ODIKD992

To implement this, I will duplicate the original IKM SQL Control Append Knowledge Module and add additional steps.

2 Additional steps will be added to the Knowledge Module:

• Drop Backup Table (if it already exists one);
• Create Backup Table of the target's table before inserting new rows, tables will be suffixed with 'Target table's name'_BKP.

I will name this IKM as 'IKM SQL CONTROL APPEND/ wCreateBKP':

No need to specify target technology.

In Details tab, each step of the knowledge module is specified, and the Order is ranked in a top down approach, so the upper most step starts the first, and so on.

Typically the additional step for Dropping an existing BKP_ table with the target's table name should be the first step in this knowledge module, additionally after this step we want to Create a BKP_ prefixed table with all the existing records in the target table.

So the dependency here is that both of the steps should execute before Truncate target table or Delete target table step. Hence I will add both of these steps as the first to execute in the KM.

The typical SQL query for dropping a table in Oracle is:

• Drop table 'Table_name'_BKP;

And the typical SQL query for creating a table postfixed with '_BKP' as a copy of another table including rows is:

• Create table 'Table_name'||'_BKP' as Select * from 'Table_name';

ODI KM's use substitution API's. As cited in oracle's website (link above): KMs are written as templates by using the Oracle Data Integrator substitution API. The API methods are java methods that return a string value. They all belong to a single object instance named "odiRef". The same method may return different values depending on the type of KM that invokes it. That's why they are classified by type of KM.

Query:

• Drop table 'Table_name'_BKP;
• Drop table <%=odiRef.getTable("L","TARG_NAME","A")%>_BKP

And,

• Create table 'Table_name'||'_BKP' as Select * from 'Table_name';
• Create table <%=odiRef.getTable("L","TARG_NAME","A")%>_BKP as select * from <%=odiRef.getTable("L","TARG_NAME","A")%>

Now that we have the substitution API's ready, create two additional steps in Details tab of KM's using the green plus '+' on the top right corner, and rearrange the order using the up and down arrows on the top right corner:

New steps in KM

Note that when Dropping the backup table (if existing) the checkbox for ignore errors is marked, meaning that if the _BKP backup table doesn't already exist, do not throw an error and proceed with Creating it.

Open the step for Drop Backup Table and make sure to check that box and paste the substitution API's code under Command on Target section. No need to specify Technology of Context whatsoever.

Drop Backup Table (if existing)

Navigate to Create Backup Table step, this time I won't check Ignore Errors checkbox because I wan't to throw an error whenever a table cannot be created, make sure to paste the code under Command on Target Section.

Create Backup Table

Now that both of these steps are created, I want to create two Options under this Knowledge module, so if I want to disable this feature of the KM, I can do that by selecting it as False.

Navigate to Knowledge Modules in Designer Tab, Expand Knowledge Modules, expand IKM SQL CONTROL APPEND/ wCreateBKP and right click on it to add a new option:

New Option

When Creating the options, check for Position, the Options Position should refer to the Order position of the step inside Details tab, in my case Dropping the Backup table should have position 10 and Creating it should have position 20.

Edit Option

Edit Option

Now that we have the Knowledge module in place, it is time to test it.

To demonstrate it's execution and step I will use a simple interface which uses a file to populate an Oracle table.

Under flow tab of this interface, I will choose the custom IKM we've built, and I will make sure that DROP_BACKUP_TBL option and CREATE_BACKUP_TBL options are selected as true (which it should be by default) and Truncate as True.

Flow

It is time to execute the interface and check the steps.

Execution steps

As you can observe, Drop Backup Table has a warning because a previous table with _BKP name did not exist, and then Create Backup Table executed correctly.

If I run it again, it won't show any warning on Drop Backup Table because now a _BKP table exists:

Execution steps

How to perform Group By in ODI11g interfaces

Aggregation is a must in Data Warehousing and Data Integration, as of ODI 12c there exists a designated tool for Group by operations, but in ODI 11g you have to perform a workaround.

Aggregate functions in SQL include:

MIN

MAX

SUM

COUNT

AVG

To perform a query with the above functions, a group by statement has to be used in order to aggregate data.

I will be using a random source file created randomly with ID and different numerical attributes:

Source file set

And a target table created manually in Oracle with identical attributes such as source file:

Target table

Now I will create an interface (GROUP_BY_EXP) which uses the data store's created from above file and table:

Initial Interface

Assuming that I want to find the SUM of Height grouped by ID only, typically this will be achieved with this query:

Mapping only ID and Height in Mappings expressions this is pretty easy:

Mapping Expressions

Executing this interface will generate this code, which indeed is exactly what we expect it to generate:

Code generated in first run

What if we map all of the columns on the target table? And we need the same task Sum(Height) and Group by ID? 

Let's firstly map all of the columns:

Mapping Expressions

Let's execute it:

Code generated in second run

Now this brings an issue because in our case we need only ID used in Group By expression.

In order to avoid this, each of unneeded columns in group by statement: NAME, LENGTH, WIDTH should have aggregate function used in expression, if that does not pose a problem in whatever you are using, then encapsulate each column with MIN/MAX:

Mapping Expressions

Let's execute it:

Code generated in third run

As you can see NAME, HEIGHT and WIDTH are removed from Group By statement whenever there's an aggregate function used on them.

Note: This is a workaround, data should be analyzed after the execution due to the fact that MIN/MAX functions might generate data which can cause trouble on your business logic.

ODI 11g - Updating existing records - Inserting new ones (based upon a key)

Today's post will be a simple but very useful business case in Data Integration and Data warehousing purposes.

- Populate a table will new records
- If the record is fully duplicate, reject it.
- If the record is a duplicate on a key, but there is a particular field changing, update that record

In Oracle Data Integrator 11g this is easily achieved through an Integration Knowledge Module known as IKM Oracle Incremental Update (MERGE) which comes built in with ODI11g.

To use it, there must be a key declared either in database side, or within ODI.

The example which I will cover will include declaring a UK within ODI.

-  Firstly make sure that source datastore and target datastores are created in ODI.







I reverse engineered this table inside ODI:








Create a new Interface and define source and target:






As mentioned above, a key has to be declared, in my case, an Employee is uniquely defined with EMPID, so EMPID is the column which will be used to determine whether a record exists or not!


Navigate to Quick-Edit tab and define EMPID as UK:




Now we are almost set, but we also need to define the knowledge module in the Flow tab:

Click on top of SrcSet (In my case it is a file so an LKM is needed to load it into an Oracle RDBMS table) - Select LKM File to SQL


Click on top of Target - Select IKM Oracle Incremental Update (Merge) specifically.


Make sure that you disable Flow Control (False) because CKM is not being used and it will throw an error otherwise:





The target table is currently empty, in the first run every record in the source set will be inserted, run it and insert every record if there is any:










Expanding Merge rows step, it can be seen that all of the 7 records have been inserted:

Automating email reports using ODI Packages - Accessing Data Sources with changing name mask using ODI Variables

A very important feature which ODI offers is ODI Packages. ODI offers many elements which can be used to automate different processes.

Today I will try to demonstrate the usage of ODI Packages, scheduling runs for those packages and using variables to dynamically access different data sources every day.

In the end, different statistics and automatic emails can be generated to achieve basically everything which can be achieved through an SQL Query in a database :).

Scenario:

- Read a .TXT file which name changes everyday with current date. (test20190811.txt)
- Populate an oracle table everyday except for Saturday and Sunday
- Generate a .csv file from that table with name and date mask (test20190811.csv)
- Attach that file to an email and send that email everyday to a recipient.
- Show row number of the populated table in email.



Firstly we need to create a variable which obtains the value of current date.
That can be done through a simple query in Oracle:

SELECT to_char(sysdate,'YYYYMMDD') from dual;

The variable will be named VAR1, and the Schema provided can be any schema since I am not selecting a particular table but I am selecting from dual which is a built in oracle table. (Otherwise if you need to select from a specific table, the schema where that table is created has to be provided).




Make sure to test the value of the variable using the refresh logo on top right corner, after clicking it, navigate to history tab on the left top corner and check the value if it corresponds to today's date and desired format (in my case YYYYMMDD).


Now that we have created the variable which will do the work, we have to create a new interface which will read the file and populate an non-existing oracle table which will be created during the first run of the interface.

As mentioned, I will be reading a source file .txt named test20190811 (date is the file mask which will change everyday according to current date). Create a new data store which will be used as a source in the interface, make sure that at Resource Name, give the static part of the name, and then for file mask address the variable created, variables in ODI are addressed using "#", in this case #VAR1.

Make sure to write this part manually and don't use the search icon on the right:

After that, define file format in Files tab and make sure to add Columns manually using + sign in Columns tab. (It is not possible to reverse engineer the file because RKM will not obtain the variable value as of 11g.)

After defining manually all of the columns, create a new interface and add the created data store as source, right click on it and Add to target. All of existing columns from source will be mapped into target. 

(Note that the table can be created in database manually and reverse engineered as a new data store and dragged into target tab).

In my case LKM will be assigned LKM File to SQL, IKM SQL Control Append.

Run the interface and check the operator:


Now that we have the interface which loads (File to table) we have to generate a new .csv file containing data from this table and send it as an email attachment and provide statistics within email for every load.

In one of my previous posts, I have shown how to generate a file from an Oracle table, I will generate a .xlsx using ODI package elements this time which is quite interesting.

Let's assemble our elements in a package.

Create a new package, I will name it EMAIL_AUTOMATION.

The first element that I will drag is VAR1 variable, we have to refresh that variable everyday (rememberthat variables have to be refreshed, otherwise they will keep the old value), due to this fact we need to load a new file which updates is name mask everyday so a refreshing is a must:

Drag VAR1 into the diagram, it will be set as a Refresh Variable, which is exactly what we need.



After having refreshed the variable, interface has to be run, we assume that the file will be generated in the path where Interface reads the source file (If it doesn't, we can use ODI File Copy from Toolbox provided and specify the copy path before running the interface but I won't be covering that topic). Drag the interface into the diagram. Connect both points using the Ok arrow which indicates that if one step executes correctly continue with the other step:



After this step file generation is required, I will do that not using an interface but using ODI SQL Unload Element from toolbox,

-Target File (full path of the file which will be generated (I want to name it file#VAR1.csv))
-JDBC Driver: oracle.jdbc.OracleDriver
-JDBC URL: jdbc:oracle:thin@<>:<>;<> (provide database info)
-User: Provide User
-Password: Provide Password
-File Format: Delimited
-SQL QUERY: (select * from <MYSCHEMA>.TARGETTEST), this is the query which "decides" what to output in your file.

Right click on ODI Sql Unload step and select execute step to test it.



Go to the path specified to generate the file and open it:



Having done this correctly you can see the data from the table.


Let's create a new variable which counts the rows in the table, using this query:

SELECT COUNT(*) FROM <MYSCHEMA>.TARGETTEST

Drag this variable into package and make sure to refresh it before the interface, otherwise the row count will not be updated:



As you can see another variable refresh has been added and also an ODISendMail step from Internet folder inside the toolbox on the left, now we need to fill the ODISendMail parameters as below:



- Mail Server (ip address of the mail server)
- From: Any valid email
- To: Any valid email
- Attachment: Full path of the file generated + variable which obtains the dynamically changing mask
- Message Body which consists the date of load, table row count as below:





Save the package and run it (check operator with all of the steps finished):



Wait for the email:




Now lets add a schedule to this package, to do so we must generate a package scenario and also have an existing ODI Agent running.

Generate a scenario under package (Scenario is a frozen code)



Expand the new scenario generated, right click on Scheduling and select new scheduling:





Many constraints can be added, I want it to run every day of the week 7:28 AM.

Enabling Data Quality checks in ODI11g through Flow Control / Isolating errors automatically in an error table

ODI as a tool offers data quality assurance through various scopes.

Today I will demonstrate how to add ODI level constraints and isolate errors which violate a constraint, in error tables that ODI creates automatically (known as E$ tables).

In principle, ODI has 2 kind of data controls known as:

• Flow Control - which validates data before inserting into a target table (validation is done within I$ tables). What happens is that it leaves only the records which do not violate a constraint into the target table.
• Static Control - Validation occurs after inserted in target table, though insertion is done in E$ table for validating records but they won't be deleted from target table.

Enabling these controls in ODI, one must use the IKM's Flow Control/Static Control options of an IKM in the Flow tab of an interface.

But first we need to define what constraints do columns have in order to apply Flow Control.

Todays scenario: 

• Populate a table in which the primary key of the table cannot be null (ID).
• Column height cannot be more than 200
• Column date has to be in format YYYYMMDD (i.e 20191101)

Let's dig into the problem.

Create an interface which populates a table consisting in my case columns mentioned above.

I will be reading a .txt source file which is tab delimited. Data populated in the target table looks like this:

Area hand circled with red, marks the violations which will be declared in the datastore of the target table now. We can observe that the ID is null, Height is bigger that 200 and DATE is not in format of YYYYMMDD.

First let us write the SQL statements which validate the conditions:

As per ID: WHERE ID IS NOT NULL will do the work

As per Height: WHERE to_number(HEIGHT) > 200

As per Date: REGEXP_LIKE(DATE, '^\d{4}(0[1-9]|(1[0-2]))(0[1-9]|[1-2][0-9]|3[0-1])$')

Navigate to Quick-Edit tab and select CN and UK (Check Not Null) for ID:

Note: UK has to be selected for the primary key because flow control requires one.

Now we need to create constraints in target table's datastore. Navigate to: Designer - Models - Model where your datastore resides - Right Click Constraints - New Condition:

In my case, I created 2 Conditions

1. CHECK_HEIGHT

2. CHECK_DATE

Given the names, I also choose Type: Oracle Data Integrator Condition

In the Where area: paste the SQL expressions.

Message: the error message which you want to prompt

Note: In Control tab on top left corner Flow and Static Control can be allowed for these constraints. I have checked both, but in my case I will be using only Flow Control.

Save the constraints and now open the interface, go to Flow tab, select IKM_SQL_Control_Append and mark the option FLOW_CONTROL to True.

Under Control Tab next to Flow you can select the CKM which does the controlling.
Besides selecting the CKM it allows you to disable or enable specific constraints (True or False).




Also the maximum number of errors allowed can be specified, meaning that if that number is exceeded, interface will not throw warnings instead it will throw an error!

Run the interface and check the Operator:

In DBMS you can see that a new table has been created in the schema, the E$ table consists of the constraint violating records:

Only one record did not violate the constraint and that record is inserted into the target table: