D-Net

package eu.dnetlib.msro.workflows.nodes.hadoop;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Map.Entry;

import javax.annotation.Resource;

import com.google.common.collect.Iterables;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.msro.workflows.nodes.BlackboardJobNode;

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.msro.workflows.procs.Token;

import eu.dnetlib.rmi.data.hadoop.HadoopBlackboardActions;

import eu.dnetlib.rmi.data.hadoop.HadoopJobType;

import eu.dnetlib.rmi.data.hadoop.HadoopService;

import eu.dnetlib.rmi.enabling.ISLookUpException;

import eu.dnetlib.rmi.enabling.ISLookUpService;

import eu.dnetlib.rmi.manager.MSROException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class SubmitHadoopJobNode extends BlackboardJobNode {

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(SubmitHadoopJobNode.class);

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String hadoopJob;

	private String cluster;

	private boolean simulation = false;

	private Map<String, Object> jobParams = new HashMap<>();

	@Override

	protected String obtainServiceId(final Env env) {

		return getServiceLocator().getServiceId(HadoopService.class);

	}

	@Override

	protected void prepareJob(final BlackboardJob job, final Token token) throws Exception {

		String type = getJobType(getHadoopJob());

		log.info("submitting job " + getHadoopJob() + " type: " + type);

		job.setAction(type);

		job.getParameters().put("job.name", getHadoopJob());

		job.getParameters().put("cluster", cluster(token));

		job.getParameters().put("simulation", String.valueOf(isSimulation()));

		for (Entry<String, Object> e : getJobParams().entrySet()) {

			try {

				job.getParameters().put(e.getKey(), e.getValue().toString());

			} catch (Throwable ex) {

				throw new MSROException(String.format("missing value for map key '%s'", e.getKey()));

			}

		}

	}

	private String cluster(final Token token) {

		if (token.getEnv().hasAttribute("cluster")) {

			String cluster = token.getEnv().getAttribute("cluster", String.class);

			log.info("found override value in wfEnv for 'cluster' param: " + cluster);

			return cluster;

		}

		return getCluster();

	}

	/**

	 * reads the job type for the given job name

	 *

	 * @param jobName

	 * @return

	 * @throws ISLookUpException

	 */

	private String getJobType(final String jobName) throws ISLookUpException {

		List<String> res =

				serviceLocator.getService(ISLookUpService.class).quickSearchProfile(

						"/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'HadoopJobConfigurationDSResourceType']//HADOOP_JOB[./@name='" + jobName

								+ "']/@type/string()");

		if (res.isEmpty()) { throw new IllegalStateException("unable to find job type for job: " + jobName); }

		final HadoopJobType type = HadoopJobType.valueOf(Iterables.getOnlyElement(res));

		switch (type) {

		case mapreduce:

			return HadoopBlackboardActions.SUBMIT_MAPREDUCE_JOB.toString();

		case admin:

			return HadoopBlackboardActions.SUBMIT_ADMIN_JOB.toString();

		case oozie:

			return HadoopBlackboardActions.SUBMIT_OOZIE_JOB.toString();

		default:

			throw new IllegalStateException("undefined job type: " + type.toString());

		}

	}

	public String getHadoopJob() {

		return hadoopJob;

	}

	public void setHadoopJob(final String hadoopJob) {

		this.hadoopJob = hadoopJob;

	}

	public String getCluster() {

		return cluster;

	}

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

	public boolean isSimulation() {

		return simulation;

	}

	public void setSimulation(final boolean simulation) {

		this.simulation = simulation;

	}

	public Map<String, Object> getJobParams() {

		return jobParams;

	}

	public void setJobParams(final Map<String, Object> jobParams) {

		this.jobParams = jobParams;

	}

}