D-Net

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

import java.nio.file.FileSystems;

import java.nio.file.Path;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

import eu.dnetlib.msro.workflows.graph.Arc;

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

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

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

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

import org.apache.commons.lang3.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 DedupConfigurationLoaderJobNode {

	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;

	private String workDir;

	@Override

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

		String depthString = token.getEnv().getAttribute(DepthParam, String.class);

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

		if (StringUtils.isBlank(depthString)) {

			depthString = "0";

		}

		int depth = Integer.valueOf(depthString);

		final String outputPath = getPath(getWorkDir(), depth);

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

		switch (getStatusFromEnv(token.getEnv())) {

		case DATALOAD:

			setHadoopJob("dedupSimilarity2GraphJob");

			job.getParameters().put("mapred.output.dir", getPath(getWorkDir(), depth) + "/out");

			hadoopService.createHdfsDirectory(getCluster(), outputPath, true);

			break;

		case DEPTH_N:

			setHadoopJob("dedupMinDistGraphJob");

			final String newOutputPath = getPath(getWorkDir(), depth + 1);

			hadoopService.createHdfsDirectory(getCluster(), 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.getEnv().setAttribute(DepthParam, String.valueOf(depth + 1));

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

			break;

		}

		super.prepareJob(job, token);

	}

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

		log.info("got basePath: " + basePath);

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

		final String path = fsPath.toAbsolutePath().toString();

		log.info("built outputPath: " + path);

		return path;

	}

	private STATUS getStatusFromEnv(final Env env) {

		if (StringUtils.isBlank(env.getAttribute(StatusParam, String.class))) {

			return STATUS.DATALOAD;

		}

		STATUS current = STATUS.DATALOAD;

		try {

			current = STATUS.valueOf(env.getAttribute(StatusParam, String.class));

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

		} catch (IllegalArgumentException e) {}

		return current;

	}

	@Override

	protected BlackboardWorkflowJobListener generateBlackboardListener(final Token token) {

		return new BlackboardWorkflowJobListener(token) {

			@Override

			protected void onDone(final BlackboardJob job) {

				final STATUS status = getStatusFromEnv(token.getEnv());

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

				switch (status) {

				case DATALOAD:

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

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

					token.release("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.getEnv().removeAttribute(StatusParam);

						token.getEnv().removeAttribute(DepthParam);

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

						token.release(Arc.DEFAULT_ARC);

					} else {

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

						token.release("depth_n");

					}

					break;

				}

			}

		};

	}

	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;

	}

	public String getWorkDir() {

		return workDir;

	}

	public void setWorkDir(final String workDir) {

		this.workDir = workDir;

	}

	enum STATUS {DATALOAD, DEPTH_N}

}