Data Integration

The classic OGSA-DAI data integration scenario deals with the issue of performing a SQL join operation between tables in two different databases. In this page, you will learn how to join a set of XML documents from an XML database with a relational table.

This page will guide you through the following steps:

• Select data from an XML database,
• Transform it into WebRowSet format,
• Stream it to another data service using data transport operations,
• Bulk load the data into a new table in the relational database

Below you will find an example solution.

1. Locate a sink data service which connects to a MySQL database resource and a source data service which connects to an eXist database resource.

   String handle = "http://localhost:8080/wsrf/services/ogsadai/DataService";
   String sinkID = "MySQLResource";
   String sourceID = "ExistResource";
   DataService sinkService = GenericServiceFetcher.getInstance().getDataService(handle, sinkID);
   DataService sourceService = GenericServiceFetcher.getInstance().getDataService(handle, sourceID);
   

2. Create a new table in the sink database.

   String tableName = "mytable";
   String createTable = "create table if not exists " + tableName  +
                        " (id INTEGER, name VARCHAR(64), " +
                        "address VARCHAR(128), phone VARCHAR(20))";
   SQLUpdate create = new SQLUpdate( createTable );
   sinkService.perform( create );
   

3. Create a new session for the source request.

   Session session = sourceService.createSession();
   

4. Prepare an XPath query for the source XML database - via xPathStatement (using an XPathQuery object). We will use only the first 500 entries in the littleblackbook collection.

   XPathQuery query = new XPathQuery("/entry[@id<500]");
   

5. Connect the output of the XPathQuery activity object to an XSLTransform activity object in order to transform the resource set format into WebRowSet format (via execution of the xslTransform activity).

   XSLTransform transform = new XSLTransform();
   transform.setXMLInput( query.getOutput() );
   

6. You can use the XSLT document here to translate the resource set into the WebRowSet format. Deliver the XSLT document from the URL above and connect the input of the transform activity to the delivery's output.

   DeliverFromURL deliver = new DeliverFromURL( url );
   transform.setXSLTInput( deliver.getOutput() );
   

7. Connect the output of the XSLTransform activity object to the DTOutputStream object's input. The output stream will provide the data to another data service.

   DTOutputStream outputStream = new DTOutputStream();
   outputStream.setInput(transform.getOutput());
   

8. Then add all activities to the source request

   ActivityRequest sourceRequest = new ActivityRequest();
   sourceRequest.add( deliver );
   sourceRequest.add( query );
   sourceRequest.add( transform );
   sourceRequest.add( outputStream );
   

   and join the existing session that we created earlier:

   sourceRequest.setSessionRequirements(new JoinExistingSession(session));
   

9. You can now start the source request which will provide the data through the output stream.

   sourceService.perform( sourceRequest );
   

10. Construct another request for the sink with two activities: deliverFromDT (using an DeliverFromDT object) and sqlBulkLoadRowSet (using an SQLBulkLoad object). The deliverFromDT will pull data in blocks from the source data service and then pass on the data to the sqlBulkLoadRowSet activity.

    DeliverFromDT deliverFromDT = new DeliverFromDT();
    deliverFromDT.setDataTransportInput(outputStream.getDataTransport());
    deliverFromDT.setDataTransportMode( DataTransportMode.BLOCK );
    SQLBulkLoad bulkload = new SQLBulkLoad( deliverFromDT.getOutput(), tableName );
    ActivityRequest sinkRequest = new ActivityRequest();
    sinkRequest.add( deliverFromDT );
    sinkRequest.add( bulkload );
    

11. Now start the sink request.

    sinkService.perform(sinkRequest);
    

12. After the sink request has returned the number of rows that have been loaded into the table at the sink is available from the SQLBulkLoad object.

    bulkload.getInsertedRowsCount();
    

For a complete example, see OGSA-DAI/examples/src/uk/org/ogsadai/examples/clienttoolkit/DataIntegrationExample.java