D-Net

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

	<parent>

		<groupId>eu.dnetlib</groupId>

		<artifactId>dnet-parent</artifactId>

		<version>1.0.0</version>

		<relativePath />

	</parent>

	<modelVersion>4.0.0</modelVersion>

	<groupId>eu.dnetlib</groupId>

	<artifactId>dnet-deduplication</artifactId>

	<packaging>jar</packaging>

	<version>1.0.1</version>

	<scm>

		<developerConnection>scm:svn:https://svn.driver.research-infrastructures.eu/driver/dnet40/modules/dnet-deduplication/tags/dnet-deduplication-1.0.1</developerConnection>

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-msro-service</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-hadoop-service-rmi</artifactId>

			<version>[1.0.0,2.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-actionmanager-api</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-modular-ui</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>		

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-index-solr-client</artifactId>

			<version>[2.0.0,3.0.0)</version>

		</dependency>	

		<dependency>

			<groupId>eu.dnetlib</groupId>

			<artifactId>dnet-openaireplus-mapping-utils</artifactId>

			<version>[3.0.0,4.0.0)</version>

		</dependency>

		<dependency>

			<groupId>javax.servlet</groupId>

			<artifactId>javax.servlet-api</artifactId>

			<version>${javax.servlet.version}</version>

			<scope>provided</scope>

		</dependency>

		<dependency>

			<groupId>com.fasterxml.jackson.core</groupId>

			<artifactId>jackson-databind</artifactId>

			<version>2.4.3</version>

		</dependency>

		<dependency>

			<groupId>com.google.guava</groupId>

			<artifactId>guava</artifactId>

			<version>${google.guava.version}</version>

		</dependency>

	</dependencies>

</project>

package eu.dnetlib.msro.workflows.dedup;

import java.util.List;

import org.junit.Before;

import org.junit.Test;

import com.google.common.collect.Lists;

import eu.dnetlib.functionality.modular.ui.dedup.SimilarityMeshBuilder;

import eu.dnetlib.miscutils.collections.Pair;

public class SimilarityMeshBuilderTest {

	private List<String> list;

	@Before

	public void setUp() throws Exception {

		list = Lists.newArrayList();

		for (int i = 0; i < 10; i++) {

			list.add(i + "");

		}

	}

	@Test

	public void test() {

		final List<Pair<String, String>> combinations = SimilarityMeshBuilder.build("50", list);

		System.out.println(combinations);

		System.out.println(combinations.size());

	}

}

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

import java.util.List;

import com.google.common.collect.Lists;

import eu.dnetlib.miscutils.collections.Pair;

public class SimilarityMeshBuilder {

	public static List<Pair<String, String>> build(final String typePrefix, final List<String> list) {

		final List<Pair<String, String>> res = Lists.newArrayList();

		for (int i = 0; i < list.size(); i++) {

			for (int j = 0; j < list.size(); j++) {

				if (!list.get(i).equals(list.get(j))) {

					final String source = typePrefix + "|" + list.get(i);

					final String target = typePrefix + "|" + list.get(j);

					res.add(new Pair<String, String>(source, target));

				}

			}

		}

		return res;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import java.io.StringReader;

import java.util.Iterator;

import java.util.List;

import java.util.Queue;

import javax.annotation.Resource;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.antlr.stringtemplate.StringTemplate;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

import org.dom4j.DocumentException;

import org.dom4j.Node;

import org.dom4j.io.SAXReader;

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

import com.google.common.collect.Lists;

import com.google.common.collect.Queues;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.enabling.resultset.IterableResultSetFactory;

import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;

import eu.dnetlib.miscutils.collections.Pair;

import eu.dnetlib.msro.workflows.hadoop.utils.SimilarityMeshBuilder;

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

public class BuildSimilarityMeshJobNode extends AsyncJobNode {

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

	/** The result set factory. */

	@Resource(name = "iterableResultSetFactory")

	private IterableResultSetFactory resultSetFactory;

	/** The result set client factory. */

	@Resource(name = "resultSetClientFactory")

	private ResultSetClientFactory resultSetClientFactory;

	private StringTemplate similarity;

	private String inputEprParam;

	private String outputEprParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String inputEpr = token.getEnv().getAttribute(getInputEprParam());

		final Iterator<String> rsClient = resultSetClientFactory.getClient(inputEpr).iterator();

		final Queue<Object> queue = Queues.newLinkedBlockingQueue();

		final SAXReader reader = new SAXReader();

		if (rsClient.hasNext()) {

			populateQueue(queue, reader, rsClient.next());

		}

		final W3CEndpointReference eprOut = resultSetFactory.createIterableResultSet(new Iterable<String>() {

			@Override

			public Iterator<String> iterator() {

				return new Iterator<String>() {

					@Override

					public boolean hasNext() {

						synchronized (queue) {

							return !queue.isEmpty();

						}

					}

					@Override

					@SuppressWarnings("unchecked")

					public String next() {

						synchronized (queue) {

							final Object o = queue.poll();

							while (queue.isEmpty() && rsClient.hasNext()) {

								populateQueue(queue, reader, rsClient.next());

							}

							return buildSimilarity((Pair<String, String>) o);

						}

					}

					@Override

					public void remove() {

						throw new UnsupportedOperationException();

					}

				};

			}

		});

		token.getEnv().setAttribute(getOutputEprParam(), eprOut.toString());

		return Arc.DEFAULT_ARC;

	}

	private void populateQueue(final Queue<Object> q, final SAXReader r, final String xml) {

		try {

			final Document d = r.read(new StringReader(xml));

			final String groupid = d.valueOf("//FIELD[@name='id']");

			final List<?> items = d.selectNodes("//FIELD[@name='group']/ITEM");

			final String entitytype = d.valueOf("//FIELD[@name='entitytype']");

			final List<String> group = Lists.newArrayList();

			for (final Object id : items) {

				group.add(((Node) id).getText());

			}

			// compute the full mesh

			final String type = String.valueOf(Type.valueOf(entitytype).getNumber());

			final List<Pair<String, String>> mesh = SimilarityMeshBuilder.build(type, group);

			// total += mesh.size();

			if (log.isDebugEnabled()) {

				log.debug(String.format("built mesh for group '%s', size %d", groupid, mesh.size()));

			}

			for (final Pair<String, String> p : mesh) {

				if (log.isDebugEnabled()) {

					log.debug(String.format("adding to queue: %s", p.toString()));

				}

				q.add(p);

			}

		} catch (final DocumentException e) {

			log.error("invalid document: " + xml);

		}

	}

	private String buildSimilarity(final Pair<String, String> p) {

		final StringTemplate template = new StringTemplate(getSimilarity().getTemplate());

		template.setAttribute("source", p.getKey());

		template.setAttribute("target", p.getValue());

		final String res = template.toString();

		return res;

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getOutputEprParam() {

		return outputEprParam;

	}

	public void setOutputEprParam(final String outputEprParam) {

		this.outputEprParam = outputEprParam;

	}

	public StringTemplate getSimilarity() {

		return similarity;

	}

	@Required

	public void setSimilarity(final StringTemplate similarity) {

		this.similarity = similarity;

	}

}

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

import java.io.IOException;

import java.util.Map;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

import eu.dnetlib.enabling.resultset.rmi.ResultSetException;

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

import eu.dnetlib.miscutils.functional.xml.DnetXsltFunctions;

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

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

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

import eu.dnetlib.msro.workflows.resultset.ProcessCountingResultSetFactory;

import eu.dnetlib.msro.workflows.util.ProgressProvider;

import eu.dnetlib.msro.workflows.util.ResultsetProgressProvider;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public class DeleteHBaseRecordsJobNode extends BlackboardJobNode implements ProgressJobNode {

	private static final Log log = LogFactory.getLog(DeleteHBaseRecordsJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private String inputEprParam;

	private String hbaseTableProperty;

	private String cluster;

	private String xslt;

	private boolean simulation = false;

	private ProgressProvider progressProvider;

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("Invoking blackboard method");

		job.setAction(HadoopBlackboardActions.DELETE_EPR_HBASE.toString());

		job.getParameters().put("input_epr", DnetXsltFunctions.encodeBase64(prepareEpr(token)));

		job.getParameters().put("xslt", DnetXsltFunctions.encodeBase64(prepareXslt()));

		job.getParameters().put("table", getPropertyFetcher().getProperty(getHbaseTableProperty()));

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

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

	}

	@Override

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

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void populateEnv(final Env env, final Map<String, String> responseParams) {

				final String count = responseParams.get("count");

				log.info(String.format("Deleted %s objects from HBase table %s, cluster %s", count, getPropertyFetcher().getProperty(getHbaseTableProperty()),

						getCluster()));

				env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + getName() + ":count", count);

			}

		};

	}

	private String prepareEpr(final NodeToken token) throws ResultSetException {

		final String epr = token.getEnv().getAttribute(inputEprParam);

		final ResultsetProgressProvider resultsetProgressProvider = processCountingResultSetFactory.createProgressProvider(token.getProcess(), epr);

		setProgressProvider(resultsetProgressProvider);

		return resultsetProgressProvider.getEpr().toString();

	}

	private String prepareXslt() throws IOException {

		return (xslt == null) || xslt.isEmpty() ? "" : IOUtils.toString(getClass().getResourceAsStream(xslt));

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getHbaseTableProperty() {

		return hbaseTableProperty;

	}

	public void setHbaseTableProperty(final String hbaseTableProperty) {

		this.hbaseTableProperty = hbaseTableProperty;

	}

	@Required

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public void setProgressProvider(final ProgressProvider progressProvider) {

		this.progressProvider = progressProvider;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	public String getXslt() {

		return xslt;

	}

	public void setXslt(final String xslt) {

		this.xslt = xslt;

	}

	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;

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

/**

 * The Class DropHBaseTableJobNode.

 */

public class DropHBaseTableJobNode extends AbstractHBaseAdminJobNode {

	/** The Constant log. */

	private static final Log log = LogFactory.getLog(DropHBaseTableJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	/*

	 * (non-Javadoc)

	 * 

	 * @see eu.dnetlib.msro.workflows.nodes.SimpleJobNode#execute(com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String tableName = tableName(token);

		final String cluster = cluster(token);

		log.info("Dropping hbase table '" + tableName + "' on cluster: '" + cluster + "'");

		getServiceLocator().getService(HadoopService.class).dropHbaseTable(cluster, tableName);

		return Arc.DEFAULT_ARC;

	}

}

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

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

public class ExistHBaseTableJobNode extends AbstractHBaseAdminJobNode {

	private static final Log log = LogFactory.getLog(ExistHBaseTableJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private String existOutNode;

	private String dontExistOutNode;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String tableName = tableName(token);

		final String cluster = cluster(token);

		log.info("checking table existance: '" + tableName + "' on cluster: '" + cluster + "'");

		boolean exists = getServiceLocator().getService(HadoopService.class).existHbaseTable(cluster, tableName);

		log.info("table '" + tableName + "' exists: " + exists);

		return exists ? getExistOutNode() : getDontExistOutNode();

	}

	public String getExistOutNode() {

		return existOutNode;

	}

	public void setExistOutNode(final String existOutNode) {

		this.existOutNode = existOutNode;

	}

	public String getDontExistOutNode() {

		return dontExistOutNode;

	}

	public void setDontExistOutNode(final String dontExistOutNode) {

		this.dontExistOutNode = dontExistOutNode;

	}

}

package eu.dnetlib.msro.workflows.hadoop;

import com.googlecode.sarasvati.NodeToken;

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

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

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

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

public class DeleteHdfsPathJobNode extends BlackboardJobNode {

	private String cluster;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		job.setAction(HadoopBlackboardActions.DELETE_HDFS_PATH.toString());

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

		// The "path" parameter is set by the following call

		job.getParameters().putAll(parseJsonParameters(token));

	}

	public String getCluster() {

		return cluster;

	}

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

}

package eu.dnetlib.msro.workflows.hadoop;

import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Iterables;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class FindIndexJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(FindIndexJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Resource

	private UniqueServiceLocator serviceLocator;

	/**

	 * if non null, overrides format from env.

	 */

	private String mdFormat;

	/**

	 * if non null, overrides format from env.

	 */

	private String layout;

	/**

	 * if non null, overrides format from env.

	 */

	private String interpretation;

	/**

	 * {@inheritDoc}

	 * 

	 * @see com.googlecode.sarasvati.mem.MemNode#execute(com.googlecode.sarasvati.Engine, com.googlecode.sarasvati.NodeToken)

	 */

	@Override

	public String execute(final NodeToken token) {

		final String envFormat = token.getFullEnv().getAttribute("format");

		final String envLayout = token.getFullEnv().getAttribute("layout");

		final String envInterpretation = token.getFullEnv().getAttribute("interpretation");

		final String format = handleOverride(token, "format", envFormat, getMdFormat());

		final String layout = handleOverride(token, "layout", envLayout, getLayout());

		final String interp = handleOverride(token, "interpretation", envInterpretation, getInterpretation());

		String mdRef = format + "-" + layout + "-" + interp;

		log.info("searching index for [" + mdRef + "]");

		final String indexId = findIndex(format, layout, interp);

		token.getEnv().setAttribute("index_id", indexId);

		if (indexId == null || indexId.isEmpty()) {

			log.info("no index was found for [" + mdRef + "]");

			return "notFound";

		} else {

			log.info("index found for [" + mdRef + "]: " + indexId);

			return "found";

		}

	}

	private String findIndex(final String format, final String layout, final String interp) {

		final String xquery =

				"for $x in /RESOURCE_PROFILE[.//RESOURCE_TYPE/@value='IndexDSResourceType']" + "return $x[.//METADATA_FORMAT = '" + format

						+ "' and .//METADATA_FORMAT_LAYOUT='" + layout + "' and .//METADATA_FORMAT_INTERPRETATION = '" + interp

						+ "']//RESOURCE_IDENTIFIER/@value/string()";

		try {

			log.info("xquery: " + xquery);

			List<String> ids = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

			log.info("found indexDS ids: " + ids);

			if (ids == null || ids.isEmpty()) { return null; }

			if (ids.size() > 1) { throw new IllegalStateException("found more than one index of given format: " + format + ", layout: " + layout

					+ ", interpretation: " + interp); }

			return Iterables.getOnlyElement(ids);

		} catch (Exception e) {

			return null;

		}

	}

	private String handleOverride(final NodeToken token, final String name, final String env, final String override) {

		if (override != null) {

			token.getEnv().setAttribute(name, override);

		}

		return override != null ? override : env;

	}

	public String getMdFormat() {

		return mdFormat;

	}

	public void setMdFormat(final String mdFormat) {

		this.mdFormat = mdFormat;

	}

	public String getLayout() {

		return layout;

	}

	public void setLayout(final String layout) {

		this.layout = layout;

	}

	public String getInterpretation() {

		return interpretation;

	}

	public void setInterpretation(final String interpretation) {

		this.interpretation = interpretation;

	}

}

package eu.dnetlib.msro.workflows.hadoop;

import java.util.List;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Iterables;

import com.googlecode.sarasvati.NodeToken;

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

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

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

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

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;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

	protected void prepareJob(final BlackboardJob job, final NodeToken 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()));

		job.getParameters().putAll(parseJsonParameters(token));

	}

	private String cluster(final NodeToken token) {

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

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

			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;

	}

}

# temporary name

dnet.dedup.db.name = dnet_openaireplus

dnet.dedup.index.format = STDL

dnet.dedup.index.collection = STDL-index-dedup

dnet.openaire.model.relclasses.xquery = distinct-values(for $x in /RESOURCE_PROFILE[./HEADER/RESOURCE_TYPE/@value='OntologyDSResourceType']//TERM return concat($x/@code/string(), ":", $x//RELATION[./@type = 'inverseOf']/@code/string()))

<?xml version="1.0" encoding="UTF-8"?>

<RESOURCE_PROFILE xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <HEADER>

        <RESOURCE_IDENTIFIER value="ce304c65-5836-4cf0-9a48-53472b9f6f35_V29ya2Zsb3dEU1Jlc291cmNlcy9Xb3JrZmxvd0RTUmVzb3VyY2VUeXBl"/>

        <RESOURCE_TYPE value="WorkflowDSResourceType"/>

        <RESOURCE_KIND value="WorkflowDSResources"/>

        <RESOURCE_URI value=""/>

        <DATE_OF_CREATION value="2006-05-04T18:13:51.0Z"/>

    </HEADER>

    <BODY>

        <WORKFLOW_NAME>Reset HBase</WORKFLOW_NAME>

        <WORKFLOW_TYPE>Data Load</WORKFLOW_TYPE>

        <WORKFLOW_PRIORITY>30</WORKFLOW_PRIORITY>

        <CONFIGURATION start="manual">

			<NODE name="checkTable" type="CheckHBaseTable" isStart="true">

				<DESCRIPTION>check hbase table</DESCRIPTION>

				<PARAMETERS>

					<PARAM name="hbaseTableProperty" type="string" managedBy="system" required="true">hbase.mapred.datatable</PARAM>

					<PARAM name="cluster" type="string" managedBy="system" required="true">DM</PARAM>

					<PARAM name="tableColumnsParamName" type="string" managedBy="system" required="true">hTableColumns</PARAM>

					<PARAM name="existOutNode" type="string" managedBy="system" required="true">drop</PARAM>

					<PARAM name="dontExistOutNode" type="string" required="true" managedBy="system">define</PARAM>					

				</PARAMETERS>

				<ARCS>

					<ARC to="drop" name="drop" />

					<ARC to="define" name ="define" />

				</ARCS>

			</NODE>

			<NODE name="drop" type="DropHBaseTable">

				<DESCRIPTION>drop hbase table</DESCRIPTION>

				<PARAMETERS>

					<PARAM name="hbaseTableProperty" type="string" managedBy="system" required="true">hbase.mapred.datatable</PARAM>

					<PARAM name="cluster" type="string" managedBy="system" required="true">DM</PARAM>

				</PARAMETERS>

				<ARCS>

					<ARC to="define" />

				</ARCS>

			</NODE>        

			<NODE name="define" type="DefineHBaseOpenaireSchema">

				<DESCRIPTION>define OpenAIRE hbase table</DESCRIPTION>

				<PARAMETERS>

					<PARAM name="tableColumnsParamName" type="string" managedBy="system" required="true">hTableColumns</PARAM>

					<PARAM name="hbaseTableProperty" type="string" managedBy="system" required="true">hbase.mapred.datatable</PARAM>

					<PARAM name="cluster" type="string" managedBy="system" required="true">DM</PARAM>

				</PARAMETERS>

				<ARCS>

					<ARC to="create" />

				</ARCS>

			</NODE>

			<NODE name="create" type="CreateHBaseTable">

				<DESCRIPTION>create hbase table</DESCRIPTION>

				<PARAMETERS>

					<PARAM name="hbaseTableProperty" type="string" managedBy="system" required="true">hbase.mapred.datatable</PARAM>

					<PARAM name="cluster" type="string" managedBy="system" required="true">DM</PARAM>

					<PARAM name="tableColumnsParamName" type="string" managedBy="system" required="true">hTableColumns</PARAM>

				</PARAMETERS>

				<ARCS>

					<ARC to="success" />

				</ARCS>

			</NODE>

        </CONFIGURATION>

        <STATUS />

    </BODY>

</RESOURCE_PROFILE>

/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'DedupConfigurationDSResourceType']//CONFIGURATION[@id='$configCode$']//ENTITY[@name = '$entityType$']/PACE/text()

--DEDUP SPECIFIC

CREATE TABLE groups (

    id character varying(255) NOT NULL,

    entitytype character varying(255) NOT NULL,

    date timestamp without time zone default now(),

    _dnet_resource_identifier_ character varying(2048) DEFAULT ((('temp_'::text || md5((clock_timestamp())::text)) || '_'::text) || md5((random())::text)),

    PRIMARY KEY(id)

);

CREATE TABLE similarity_groups (

    groupid character varying(255) NOT NULL REFERENCES groups(id) ON DELETE CASCADE,

    objidentifier character varying(255) NOT NULL REFERENCES entities(id) ON DELETE CASCADE,

    _dnet_resource_identifier_ character varying(2048) DEFAULT ((('temp_'::text || md5((clock_timestamp())::text)) || '_'::text) || md5((random())::text)),

    PRIMARY KEY(groupid, objidentifier)

);

CREATE TABLE dissimilarities (

    id1 character varying(255) NOT NULL REFERENCES entities(id) ON DELETE CASCADE,

    id2 character varying(255) NOT NULL REFERENCES entities(id) ON DELETE CASCADE,

    _dnet_resource_identifier_ character varying(2048) DEFAULT ((('temp_'::text || md5((clock_timestamp())::text)) || '_'::text) || md5((random())::text)),

    PRIMARY KEY(id1, id2)

);

CREATE TABLE entities (

    id character varying(255) NOT NULL,

    entitytype character varying(255) NOT NULL,

    _dnet_resource_identifier_ character varying(2048) DEFAULT ((('temp_'::text || md5((clock_timestamp())::text)) || '_'::text) || md5((random())::text)),

    PRIMARY KEY(id)

);

/RESOURCE_PROFILE[.//RESOURCE_TYPE/@value = 'DedupConfigurationDSResourceType']//CONFIGURATION[@id='$configCode$']//ENTITY[@name = '$entityType$']/WORKFLOW/text()

select e.entitytype, d.id1, d.id2 from dissimilarities d join entities e on (d.id1 = e.id)

package eu.dnetlib.msro.workflows.hadoop;

import com.googlecode.sarasvati.NodeToken;

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

public class ReuseHdfsRecordsJobNode extends SimpleJobNode {

	private boolean reuseMdRecords;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		return String.valueOf(isReuseMdRecords());

	}

	public boolean isReuseMdRecords() {

		return reuseMdRecords;

	}

	public void setReuseMdRecords(final boolean reuseMdRecords) {

		this.reuseMdRecords = reuseMdRecords;

	}

}

<?xml version="1.0" encoding="UTF-8"?>

<xsl:stylesheet version="1.0"

	xmlns:xsl="http://www.w3.org/1999/XSL/Transform" xmlns:dc="http://purl.org/dc/elements/1.1/"

	xmlns:dr="http://www.driver-repository.eu/namespace/dr" xmlns:dri="http://www.driver-repository.eu/namespace/dri" 

	xmlns:oaa="http://namespace.openaire.eu/oaa" xmlns:oaf="http://namespace.openaire.eu/oaf"

	xmlns:dnet="eu.dnetlib.data.transform.xml.DNetDbToHbaseXsltFunctions">

	<xsl:output omit-xml-declaration="yes" indent="yes"/>

	<xsl:template match="/">

		<xsl:variable name="type" select="/ROW/FIELD[@name = 'entitytype']" />

		<xsl:variable name="source" select="dnet:oafSimpleId($type, /ROW/FIELD[@name = 'id1'])" />

		<xsl:variable name="target" select="dnet:oafSimpleId($type, /ROW/FIELD[@name = 'id2'])" />

		<ROWS>

			<xsl:if test="string-length($source) &gt; 0 and string-length($target) &gt; 0">

				<ROW key="{$source}" columnFamily="resultResult_dedupSimilarity_isSimilarTo">

					<QUALIFIER name="{$target}" type="base64"></QUALIFIER>

				</ROW>

				<ROW key="{$target}" columnFamily="resultResult_dedupSimilarity_isSimilarTo">

					<QUALIFIER name="{$source}" type="base64"></QUALIFIER>

				</ROW>				

			</xsl:if>

		</ROWS>

	</xsl:template>

</xsl:stylesheet>

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

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

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

public class DedupGrouperJobNode extends DedupConfigurationAwareJobNode {

	// TODO factor out this constant, it should be a configuration parameter

	public static final int DEDUP_GROUPER_MAX_LOOPS = 10;

	public static final String DEDUP_GROUPER_LOOPER = "dedup.grouper.looper";

	public static final String DEDUP_GROUPER_CURR_WRITTEN_RELS = "dedup.grouper.written.rels";

	public static final String DEDUP_GROUPER_PREV_WRITTEN_RELS = "dedup.grouper.prev.written.rels";

	private static final Log log = LogFactory.getLog(DedupGrouperJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private class DedupBlackboardWorkflowJobListener extends BlackboardWorkflowJobListener {

		public DedupBlackboardWorkflowJobListener(final Engine engine, final NodeToken token) {

			super(engine, token);

		}

		@Override

		protected void onDone(final BlackboardJob job) {

			final int times = currentIteration(getToken());

			final String curr = job.getParameters().get(DEDUP_GROUPER_CURR_WRITTEN_RELS);

			if (times == 0) {

				getToken().getFullEnv().setAttribute(DEDUP_GROUPER_PREV_WRITTEN_RELS, -1);

			}

			if ((times >= DEDUP_GROUPER_MAX_LOOPS) || isStable(getToken(), curr)) {

				super.complete(job, "done");

			} else {

				log.info("incrementing dedup.grouper.looper to " + (times + 1));

				getToken().getFullEnv().setAttribute(DEDUP_GROUPER_LOOPER, times + 1);

				getToken().getFullEnv().setAttribute(DEDUP_GROUPER_PREV_WRITTEN_RELS, curr);

				super.complete(job, Arc.DEFAULT_ARC);

			}

		}

	}

	private int currentIteration(final NodeToken token) {

		try {

			final String sTimes = token.getFullEnv().getAttribute(DEDUP_GROUPER_LOOPER);

			log.info("read dedup.grouper.looper from fullEnv: '" + sTimes + "'");

			return Integer.parseInt(sTimes);

		} catch (final NumberFormatException e) {

			log.info("got empty dedup.grouper.looper, initializing to 0");

			return 0;

		}

	}

	private boolean isStable(final NodeToken token, final String sCurr) {

		final String sPrev = token.getFullEnv().getAttribute(DEDUP_GROUPER_PREV_WRITTEN_RELS);

		log.info("Comparing written rels, prev=" + sPrev + ", curr=" + sCurr);

		try {

			final boolean b = Integer.parseInt(sCurr) == Integer.parseInt(sPrev);

			if (b) {

				log.info("  --- The number of written rels is STABLE");

			}

			return b;

		} catch (final Exception e) {

			log.error("Invalid parsing of written rels counters - curr: " + sCurr + ", prev: " + sPrev);

			return false;

		}

	}

	@Override

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

		return new DedupBlackboardWorkflowJobListener(engine, token);

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.proto.DedupSimilarityProtos.DedupSimilarity;

import eu.dnetlib.data.proto.RelTypeProtos.RelType;

import eu.dnetlib.data.proto.RelTypeProtos.SubRelType;

import eu.dnetlib.data.proto.TypeProtos.Type;

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

import eu.dnetlib.msro.rmi.MSROException;

public class DedupSimilarityToActionsJobNode extends DedupConfigurationAwareJobNode {

	private static final Log log = LogFactory.getLog(DedupSimilarityToActionsJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	@Override

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

		super.prepareJob(job, token);

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

		if (StringUtils.isBlank(entityType))

			throw new MSROException("unable to find wf param: entityType");

		String cf = "_" + SubRelType.dedupSimilarity + "_" + DedupSimilarity.RelName.isSimilarTo;

		switch (Type.valueOf(entityType)) {

		case organization:

			cf = RelType.organizationOrganization + cf;

			break;

		case person:

			cf = RelType.personPerson + cf;

			break;

		case result:

			cf = RelType.resultResult + cf;

			break;

		default:

			throw new MSROException("invalid parameter entityType: " + entityType);

		}

		log.info("using similarity CF: " + cf);

		job.getParameters().put("similarityCF", cf);

		token.getEnv().setAttribute("similarityCF", cf);

	}

}

package eu.dnetlib.msro.workflows.hadoop;

import java.util.Map;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Engine;

import com.googlecode.sarasvati.NodeToken;

import com.googlecode.sarasvati.env.Env;

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

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

import eu.dnetlib.enabling.resultset.rmi.ResultSetException;

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

import eu.dnetlib.miscutils.functional.xml.DnetXsltFunctions;

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

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

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

import eu.dnetlib.msro.workflows.resultset.ProcessCountingResultSetFactory;

import eu.dnetlib.msro.workflows.util.ProgressProvider;

import eu.dnetlib.msro.workflows.util.ResultsetProgressProvider;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public class StoreHdfsRecordsJobNode extends BlackboardJobNode implements ProgressJobNode {

	private static final Log log = LogFactory.getLog(StoreHdfsRecordsJobNode.class); // NOPMD by marko on 11/24/08 5:02 PM

	private String inputEprParam;

	private String hdfsPathParam;

	private String cluster;

	private ProgressProvider progressProvider;

	private ProcessCountingResultSetFactory processCountingResultSetFactory;

	@Override

	protected String obtainServiceId(final NodeToken token) {

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

	}

	@Override

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

		log.info("Invoking blackboard method");

		job.setAction(HadoopBlackboardActions.IMPORT_EPR_HDFS.toString());

		job.getParameters().put("input_epr", DnetXsltFunctions.encodeBase64(prepareEpr(token)));

		job.getParameters().put("path", token.getEnv().getAttribute(getHdfsPathParam()));

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

	}

	@Override

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

		return new BlackboardWorkflowJobListener(engine, token) {

			@Override

			protected void populateEnv(final Env env, final Map<String, String> responseParams) {

				env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + getName() + ":count", responseParams.get("count"));

			}

		};

	}

	private String prepareEpr(final NodeToken token) throws ResultSetException {

		final String epr = token.getEnv().getAttribute(inputEprParam);

		final ResultsetProgressProvider resultsetProgressProvider = processCountingResultSetFactory.createProgressProvider(token.getProcess(), epr);

		setProgressProvider(resultsetProgressProvider);

		return resultsetProgressProvider.getEpr().toString();

	}

	public String getInputEprParam() {

		return inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	@Required

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public void setProgressProvider(final ProgressProvider progressProvider) {

		this.progressProvider = progressProvider;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	public String getCluster() {

		return cluster;

	}

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

	public String getHdfsPathParam() {

		return hdfsPathParam;

	}

	public void setHdfsPathParam(final String hdfsPathParam) {

		this.hdfsPathParam = hdfsPathParam;

	}

}

package eu.dnetlib.msro.workflows.actions;

import java.io.StringReader;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Attribute;

import org.dom4j.Document;

import org.dom4j.Element;

import org.dom4j.io.SAXReader;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

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

public class UpdateSetsJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		@SuppressWarnings("unchecked")

		final List<Map<String, String>> sets = new Gson().fromJson(token.getEnv().getAttribute("sets"), List.class);

		final String lastUpdate = DateUtils.now_ISO8601();

		for (Map<String, String> set : sets) {

			// update only the enabled sets.

			if (isEnabled(set)) {

				log.info("updating set: " + set.toString());

				addLatestRawSet(set, lastUpdate);

			} else {

				log.info("skip set update: " + set.toString());

			}

		}

		return Arc.DEFAULT_ARC;

	}

	private boolean isEnabled(final Map<String, String> set) {

		return set.containsKey("enabled") && set.get("enabled").equals("true");

	}

	public void addLatestRawSet(final Map<String, String> set, final String lastUpdate) throws MSROException {

		final String q = "for $x in collection('/db/DRIVER/ActionManagerSetDSResources/ActionManagerSetDSResourceType') where $x//SET/@id = '" + set.get("set")

				+ "' return $x";

		try {

			final String profile = serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(q);

			final Document doc = new SAXReader().read(new StringReader(profile));

			final String profId = doc.valueOf("//RESOURCE_IDENTIFIER/@value");

			final Element latest = (Element) doc.selectSingleNode("//RAW_SETS/LATEST");

			final Element expired = ((Element) doc.selectSingleNode("//RAW_SETS")).addElement("EXPIRED");

			for (Object o : latest.attributes()) {

				final Attribute a = (Attribute) o;

				expired.addAttribute(a.getName(), a.getValue());

			}

			latest.addAttribute("id", set.get("rawset"));

			latest.addAttribute("creationDate", set.get("creationDate"));

			latest.addAttribute("lastUpdate", lastUpdate);

			serviceLocator.getService(ISRegistryService.class).updateProfile(profId, doc.asXML(), "ActionManagerSetDSResourceType");

		} catch (Exception e) {

			String msg = "Error updating profile of set: " + set;

			log.error(msg);

			throw new MSROException(msg, e);

		}

	}

}

package eu.dnetlib.msro.workflows.dedup;

import org.apache.commons.lang.StringUtils;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.proto.TypeProtos.Type;

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

public class DedupConfigurationSetterJobNode extends AsyncJobNode {

	public static final String CONF_SEPARATOR = ":";

	private String entityType;

	private String dedupConfigSequence;

	private String dedupConfigSequenceParam;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isBlank(getEntityType())) throw new IllegalArgumentException("missing entity type parameter");

		if (StringUtils.isBlank(getDedupConfigSequence())) throw new IllegalArgumentException("missing configuration sequence");

		token.getEnv().setAttribute("entityType", getEntityType());

		token.getEnv().setAttribute("entityTypeId", Type.valueOf(getEntityType()).getNumber());

		token.getEnv().setAttribute(getDedupConfigSequenceParam(), getDedupConfigSequence());

		return Arc.DEFAULT_ARC;

	}

	public String getEntityType() {

		return entityType;

	}

	public void setEntityType(final String entityType) {

		this.entityType = entityType;

	}

	public String getDedupConfigSequence() {

		return dedupConfigSequence;

	}

	public void setDedupConfigSequence(final String dedupConfigSequence) {

		this.dedupConfigSequence = dedupConfigSequence;

	}

	public String getDedupConfigSequenceParam() {

		return dedupConfigSequenceParam;

	}

	public void setDedupConfigSequenceParam(final String dedupConfigSequenceParam) {

		this.dedupConfigSequenceParam = dedupConfigSequenceParam;

	}

}

package eu.dnetlib.msro.workflows.dedup;

import static java.lang.String.format;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.data.provision.index.rmi.IndexService;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpDocumentNotFoundException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

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

import eu.dnetlib.functionality.index.solr.feed.InputDocumentFactory;