D-Net

package eu.dnetlib.msro.workflows.dedup;

import java.nio.file.FileSystems;

import java.nio.file.Path;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.msro.workflows.nodes.blackboard.BlackboardWorkflowJobListener;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Autowired;

/**

 * Created by claudio on 14/10/15.

 */

public class MinDistSearchHadoopJobNode extends DedupConfigurationAwareJobNode {

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

	private final static String StatusParam = "MinDistSearchHadoopJobNode.status";

	private final static String DepthParam = "mindist_recursion_depth";

	private final static String UpdateCounterParam = "UpdateCounter.UPDATED";

	private final static String DebugParam = "mindist_DEBUG";

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private boolean debug = false;

	private String outPathParam;

	@Override

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

		String depthString = token.getFullEnv().getAttribute(DepthParam);

		log.debug(String.format("found depthParam: '%s'", depthString));

		if (StringUtils.isBlank(depthString)) {

			depthString = "0";

		}

		int depth = Integer.valueOf(depthString);

		final String cluster = token.getEnv().getAttribute("cluster");

		final String outputPath = getPath(token.getEnv().getAttribute("workDir"), depth);

		final HadoopService hadoopService = serviceLocator.getService(HadoopService.class);

		switch (getStatusFromEnv(token)) {

		case DATALOAD:

			setHadoopJob("dedupSimilarity2GraphJob");

			job.getParameters().put("mapred.output.dir", getPath(token.getEnv().getAttribute("workDir"), depth) + "/out");

			hadoopService.createHdfsDirectory(cluster, outputPath, true);

			break;

		case DEPTH_N:

			setHadoopJob("dedupMinDistGraphJob");

			final String newOutputPath = getPath(token.getEnv().getAttribute("workDir"), depth + 1);

			hadoopService.createHdfsDirectory(cluster, newOutputPath, true);

			job.getParameters().put(DepthParam, String.valueOf(depth));

			job.getParameters().put(DebugParam, String.valueOf(isDebug()));

			job.getParameters().put("mapred.input.dir", outputPath + "/out");

			job.getParameters().put("mapred.output.dir", newOutputPath + "/out");

			if (log.isDebugEnabled()) {

				log.debug(String.format("input job parameters: %s", job.getParameters()));

			}

			token.getFullEnv().setAttribute(DepthParam, String.valueOf(depth + 1));

			token.getFullEnv().setAttribute(getOutPathParam(), newOutputPath + "/out");

			break;

		}

		super.prepareJob(job, token);

	}

	private String getPath(final String basePath, final int depth) {

		Path path = FileSystems.getDefault().getPath(basePath, "depth_" + depth);

		return path.toAbsolutePath().toString();

	}

	private STATUS getStatusFromEnv(final NodeToken token) {

		if(StringUtils.isBlank(token.getEnv().getAttribute(StatusParam))) {

			return STATUS.DATALOAD;

		}

		STATUS current = STATUS.DATALOAD;

		try {

			current = STATUS.valueOf(token.getEnv().getAttribute(StatusParam));

			log.debug("found status: " + current.toString());

		} catch (IllegalArgumentException e) {}

		return current;

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Engine engine, final NodeToken token) {

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void onDone(final BlackboardJob job) {

				final STATUS status = getStatusFromEnv(token);

				log.debug("complete phase: " + status);

				switch (status) {

				case DATALOAD:

					token.getFullEnv().setAttribute(StatusParam, STATUS.DEPTH_N.toString());

					token.getFullEnv().setAttribute(DepthParam, "0");

					complete("depth_n");

					break;

				case DEPTH_N:

					if (log.isDebugEnabled()) {

						log.debug(String.format("return job parameters: %s=%s, %s=%s", DepthParam, job.getParameters().get(DepthParam), UpdateCounterParam, job.getParameters().get(UpdateCounterParam)));

					}

					final String counter = job.getParameters().get(UpdateCounterParam);

					if (StringUtils.isBlank(counter)) {

						token.getFullEnv().removeAttribute(StatusParam);

						token.getFullEnv().removeAttribute(DepthParam);

						log.info(String.format("done iteration %s:%s", UpdateCounterParam, 0));

						complete(Arc.DEFAULT_ARC);

					} else {

						log.info(String.format("continue with next iteration %s:%s", UpdateCounterParam, counter));

						complete("depth_n");

					}

					break;

				}

			}

			private void complete(final String arc) {

				engine.complete(token, arc);

				engine.executeQueuedArcTokens(token.getProcess());

			}

		};

	}

	public boolean isDebug() {

		return debug;

	}

	public void setDebug(boolean debug) {

		this.debug = debug;

	}

	public String getOutPathParam() {

		return outPathParam;

	}

	public void setOutPathParam(String outPathParam) {

		this.outPathParam = outPathParam;

	}

	enum STATUS {DATALOAD, DEPTH_N}

}