Editing the extractor code

Example of the extractor code

The following examples sequentially illustrate the initial code that you need to write and then, the other operations that you can perform on extractors.

1. Write the initial extractor code. Define the extractor, connect, extraction, and disconnect methods.

   Example 1: Initial Code

   package etl.extractor;
   
   import com.neptuny.cpit.etl.Conf;
   import com.neptuny.cpit.etl.DataSetList;
   import com.neptuny.cpit.etl.extractor.Extractor;
   
   public class MyExtractorE extends Extractor {
         @Override
          public void connect(Conf cf) throws Exception {
               
   #...connection stuff goes here...
          }
         @Override
          public DataSetList extract() throws Exception {
   #...extraction stuff goes here...
   
   return null;
          }
         @Override
          public void disconnect() throws Exception {      
   #...disconnection stuff goes here...
   
          }
   }

2. Connect to the database by using the connect method. Review the following example:

   Example 2: Connecting to the database

    /**  
   * Connect to data source  
   * param configuration (Conf)
   */
       @Override
      public void connect(Conf cf) throws Exception{
          this.conf=cf;
          String dburl=conf.getProperty(Conf.PROP_EDBURL); //extract.database.url
          String dbusr=conf.getProperty(Conf.PROP_EDBUSR); //extract.database.user
          String dbpwd=conf.getProperty(Conf.PROP_EDBPWD); //extract.database.password
          dbif = new DBInterface();
          String drv = conf.getProperty(Conf.PROP_EDBDRV); //extract.database.driver
          if(drv==null || "".equals(drv)){
              drv=DBInterface.ORADRV; //If not specified, use ORACLE JDBC driver by default
           }
          try {
              //Connecting to database...
              this.dbif.connect(drv,dburl,dbusr,dbpwd);
           } catch (Exception e) {
              Log.err(e, ETLErrorCode.ETL_FAIL_WRONG_DB_CONN_PARAM);
              throw e;
           }
          Log.put("Connected to database "+dburl,1);
       

   In this example, an instance of the DBInterface method has been used to create the connection and the configuration has been stored in two keys, CONF and DBIF, of the object stack.

3. Prepare the output datasets. The extract method is used to perform the query and apply the business logic to the extracted data. Review the following example:

   Example 3: Preparing output datasets

   If you want to extract the Event count by set metric and the subobject name populated with the order type, you can go to the ETL Datasets view since this metric belongs to the WKLGEN (Generic Business Driver Metrics) dataset. Hence, the output dataset can be built using the following code. 

   
   

   @Override
      public DataSetList extract() throws Exception {
          DataSetList dsList = new DataSetList();
          
          if(dbif==null) return dsList;
                
          DataSet res = new DataSet("WKLGEN");
          this.getConf().getDefChecker().initializeColumns(res);
          
          Log.put("Extracting from data source...");
          dbif.prepare("SELECT timestamp,ordercode,count FROM orders ORDER BY timestamp")
    DataSet extractedSample = dbif.executeQuery();
          
          for(int i=0;i<extractedSample.size();i++){
              String[] row = extractedSample.getRow(i);
              
              String ts      = row[0];
              String op      = row[1];
              String vol     = row[2];
              String hourStr    = ts.substring(11,13);
              int hour = Integer.parseInt(hourStr);
              
              //business logic here
              if (hour<5){
                  op = "TEST";
               } else if ("A12".equals(op)){
                  op = "CREDIT_CARD_PAYMENT";
               } else if (op.startsWith("B")){
                  op = "STOCKS";
               } else {
                  op = "OTHER";
               }
               
              //populate the dataset
              String[] datasetrow = res.newRow();
              res.fillRow("TS",ts,datasetrow);
              res.fillRow("WKLDNM","Orders",datasetrow);
              res.fillRow("DURATION","3600",datasetrow);
              res.fillRow("BYSET_EVENTS",vol,datasetrow);
              res.fillRow("SUBOBJNM",op,datasetrow);
              res.addRow(datasetrow);
              
           }
          
          Log.put("Extraction done. Extracted "+res.size());

   This code example reloads the entire ORDERS table during each run.

4. Implement an append operation. With parsers, the ETL automatically renames previously parsed files and calls the parse method to parse new files. In the case of an extractor, you need to do this manually. 
   The logical choice is to use the lastcounter parameter as the timestamp and save it in the DSStatus object. 

   Example 4: Manually implement an append operation

    //get data source status (for last counter)
          DSStatus status = this.getConf().getStatus();
          if(status==null){
              status = new DSStatus();
              status.setLastCntr("ORDERS","2023-01-01 00:00:00");
              this.getConf().insertStatus(status);
           }
          String lastCounter = status.getLastCntr("ORDERS");
   dbif.prepare("SELECT timestamp,ordercode,count FROM orders WHERE timestamp>? ORDER BY timestamp");
          dbif.setTSParam(1, lastCounter);

5. Update the lastcounter parameter to the final value of its ts variable ( the maximum timestamp value in the output dataset). 

   Example 5: Update the _lastcounter_ parameter

    // update lastcounter with last extracted sample ts...
           status.setLastCntr("ORDERS", newLastCounter);
          
           status.setMsg("ORDERS", "Extract OK. " + res.size() + " records extracted");
           this.getConf().updateStatus(status);
   

6. Implement the disconnect method to release (disconnect) the connection. 

   Example 6: Implement the _disconnect_ method

   /** * Disconnect from data source */
       @Override
      public void disconnect() throws Exception {
          if(dbif!=null){
              dbif.disconnect();
           }
       }