Ejecutar un Job (trabajo) programáticamente con Kettle (Data Integrator)

Pentaho Data Integration también conocido como Kettle es una potente herramienta de Extracción, Transformación y Carga de datos (ETL por sus siglas en inglés) mediante un enfoque innovador basado en metadatos.

En forma gráfica, esta herramienta permite crear transformaciones de datos y trabajos (jobs) para, por ejemplo, pasar datos de una base de datos a otra y/o transformarlos en el proceso. Además permite guardar esas transformaciones y trabajos en un repositorio dentro de una base de datos relacional. Para más información de Kettle visita la página de este proyecto.

Esta herramienta esta hecha en lenguaje java y permite “embeber” o incrustar su tecnología en nuestras propias aplicaciones de escritorio o web. En el siguiente código se muestra cómo ejecutar un job que previamente se diseñó y guardó en el repositorio:

package xxx.bean;

import javax.enterprise.context.RequestScoped;
import javax.faces.application.FacesMessage;
import javax.faces.context.FacesContext;
import javax.faces.event.ActionEvent;
import javax.inject.Named;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleSecurityException;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.ObjectId;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.repository.RepositoryDirectory;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.RepositoryMeta;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.pentaho.di.trans.TransMeta;

/**
 *
 * @author SCyASIG
 */
@Named(value = "jobControladorMB")
@RequestScoped
public class jobControladorMB {

    /**
     * Creates a new instance of jobControladorMB
     */
    public jobControladorMB() {
    }

    public String ejecutaJob() {
        String irA = "";
        FacesContext contexto = FacesContext.getCurrentInstance();
        String nombreJob = contexto.getExternalContext().getRequestParameterMap().get("nombreJob");
        try {
            KettleEnvironment.init();
            DatabaseMeta dbm = new DatabaseMeta("", "Oracle", "Native", "localhost", "nombreInstancia", "1521", "usuario", "password");
            KettleDatabaseRepositoryMeta rm = new KettleDatabaseRepositoryMeta();
            rm.setConnection(dbm);

            KettleDatabaseRepository repo = new KettleDatabaseRepository();

            repo.init(rm);

            repo.connect("admin", "admin");

            RepositoryDirectoryInterface rdi = repo.getUserHomeDirectory();

            JobMeta jobMeta = repo.loadJob(nombreJob, rdi, null, null);

            Job job = new Job(repo, jobMeta);

            job.start();

            job.waitUntilFinished();
            Result result = job.getResult();

            System.out.println(result.getResult());

            repo.disconnect();
            contexto.addMessage(null, new FacesMessage("El trabajo " + nombreJob + " se generó exitosamente"));
        } catch (KettleSecurityException kse) {
            kse.printStackTrace();
            contexto.addMessage(null, new FacesMessage("Ocurrió un error al ejecutar el trabajo" + nombreJob ));

        } catch (KettleException ke) {
            ke.printStackTrace();
            contexto.addMessage(null, new FacesMessage("Ocurrió un error al ejecutar el trabajo" + nombreJob ));
        } 
        return irA; 
    } 
}

El código pertenece a una clase CDI Managed Bean llamado desde una página JSF. Dentro de la instalación de Kettle 4.4.0 se encuentra la documentación de la API.

Espero les sea útil este post. ¡Saludos!