D-Net

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 java.util.Set;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.openaire.hadoop.utils.HBaseTableUtils;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public class DefineHBaseOpenaireSchemaJobNode extends AbstractHBaseAdminJobNode {

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

	private String schema;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String schemaOverride = StringUtils.isNotBlank(getSchema()) ? getSchema() : asCSV(HBaseTableUtils.listAllColumns());

		log.info("table definition: " + schemaOverride);

		token.getEnv().setAttribute(getTableColumnsParamName(), schemaOverride);

		return Arc.DEFAULT_ARC;

	}

	public String getSchema() {

		return schema;

	}

	public void setSchema(final String schema) {

		this.schema = schema;

	}

}

package eu.dnetlib.msro.workflows.hadoop;

import java.io.IOException;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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.msro.rmi.MSROException;

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

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;

public class PrepareMDStoreImportJobNode extends SimpleJobNode {

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

	@Autowired

	private UniqueServiceLocator serviceLocator;

	private String hdfsPathParam;

	private String hdfsPath;

	private String mappingParam;

	private String mapping;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		token.getEnv().setAttribute(getHdfsPathParam(), getHdfsPath());

		token.getEnv().setAttribute(getMappingParam(), readXslt(getMapping()));

		return Arc.DEFAULT_ARC;

	}

	private String readXslt(final String profileId) throws IOException, MSROException, ISLookUpException {

		if (StringUtils.isBlank(profileId)) throw new MSROException("missing profile id");

		log.info("loading mapping from profile id: " + profileId);

		final String xquery =

				String.format("/RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='%s']/BODY/CONFIGURATION/SCRIPT/CODE/*[local-name()='stylesheet']", profileId);

		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);

	}

	public String getHdfsPathParam() {

		return hdfsPathParam;

	}

	public void setHdfsPathParam(final String hdfsPathParam) {

		this.hdfsPathParam = hdfsPathParam;

	}

	public String getHdfsPath() {

		return hdfsPath;

	}

	public void setHdfsPath(final String hdfsPath) {

		this.hdfsPath = hdfsPath;

	}

	public String getMapping() {

		return mapping;

	}

	public void setMapping(final String mapping) {

		this.mapping = mapping;

	}

	public String getMappingParam() {

		return mappingParam;

	}

	public void setMappingParam(final String mappingParam) {

		this.mappingParam = mappingParam;

	}

}

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 CreateHdfsDirectoryJobNode extends BlackboardJobNode {

	private String cluster;

	private boolean force = 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 {

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

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

		job.getParameters().put("force", String.valueOf(isForce()));

		// 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;

	}

	public boolean isForce() {

		return force;

	}

	public void setForce(boolean force) {

		this.force = force;

	}

}

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

import java.io.IOException;

import java.util.Map;

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.is.lookup.rmi.ISLookUpException;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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.rmi.MSROException;

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;

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;

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

/**

 * Created by claudio on 08/04/16.

 */

public abstract class AbstractHBaseJobNode extends BlackboardJobNode implements ProgressJobNode {

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

	private final String INPUT_HBASE_TABLE_PARAM = "hbaseTable";

	private final String INPUT_EPR_PARAM = "input_epr";

	private final String INPUT_CLUSTER_PARAM = "cluster";

	private final String XSLT_PARAM = "xslt";

	private final String OUTPUT_HBASE_TABLE_PARAM = "table";

	private final String OUTPUT_CLUSTER_PARAM = "cluster";

	private final String SIMULATION_PARAM = "simulation";

	@Autowired

	protected UniqueServiceLocator serviceLocator;

	protected String inputEprParam;

	protected String hbaseTableProperty;

	protected String cluster;

	protected String mapping;

	protected boolean simulation = false;

	protected ProgressProvider progressProvider;

	protected ProcessCountingResultSetFactory processCountingResultSetFactory;

	protected abstract HadoopBlackboardActions getAction();

	@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: " + getAction().toString());

		job.setAction(getAction().toString());

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

		job.getParameters().put(XSLT_PARAM, DnetXsltFunctions.encodeBase64(readXslt(getMapping())));

		job.getParameters().put(OUTPUT_HBASE_TABLE_PARAM, tableName(token));

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

		job.getParameters().put(SIMULATION_PARAM, 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("%s %s objects to HBase table %s, cluster %s", getAction().toString(), count, tableName(token), cluster(token)));

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

			}

		};

	}

	protected String tableName(final NodeToken token) {

		if (token.getEnv().hasAttribute(INPUT_HBASE_TABLE_PARAM)) {

			String table = token.getEnv().getAttribute(INPUT_HBASE_TABLE_PARAM);

			log.debug("found override value in wfEnv for 'hbaseTable' param: " + table);

			return table;

		}

		return getPropertyFetcher().getProperty(getHbaseTableProperty());

	}

	protected String cluster(final NodeToken token) {

		if (token.getEnv().hasAttribute(INPUT_CLUSTER_PARAM)) {

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

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

			return cluster;

		}

		return getCluster();

	}

	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 readXslt(final String profileId) throws IOException, MSROException, ISLookUpException {

		if (StringUtils.isBlank(profileId)) throw new MSROException("missing profile id");

		log.info("loading mapping from profile id: " + profileId);

		final String xquery =

				String.format("/RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value='%s']/BODY/CONFIGURATION/SCRIPT/CODE/*[local-name()='stylesheet']", profileId);

		return serviceLocator.getService(ISLookUpService.class).getResourceProfileByQuery(xquery);

	}

	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;

	}

	@Override

	public ProgressProvider getProgressProvider() {

		return progressProvider;

	}

	public void setProgressProvider(final ProgressProvider progressProvider) {

		this.progressProvider = progressProvider;

	}

	public ProcessCountingResultSetFactory getProcessCountingResultSetFactory() {

		return processCountingResultSetFactory;

	}

	@Required

	public void setProcessCountingResultSetFactory(final ProcessCountingResultSetFactory processCountingResultSetFactory) {

		this.processCountingResultSetFactory = processCountingResultSetFactory;

	}

	public String getMapping() {

		return mapping;

	}

	public void setMapping(final String mapping) {

		this.mapping = mapping;

	}

	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;

import javax.annotation.Resource;

import org.apache.commons.lang.StringUtils;

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

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

/**

 * The Class SetClusterAndTableJobNode.

 */

public class SetClusterAndTableJobNode extends AsyncJobNode {

	/** The cluster. */

	private String cluster;

	/** The table. */

	@Value("${hbase.mapred.datatable}")

	private String table;

	/** The table param. */

	private String tableParam;

	/** The service locator. */

	@Resource

	private UniqueServiceLocator serviceLocator;

	/*

	 * (non-Javadoc)

	 * 

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

	 */

	@Override

	protected String execute(final NodeToken token) throws Exception {

		if (StringUtils.isBlank(getCluster())) throw new IllegalArgumentException("missing cluster name parameter");

		if (StringUtils.isBlank(getTable())) throw new IllegalArgumentException("missing table name parameter");

		if (!serviceLocator.getService(HadoopService.class).existHbaseTable(getCluster(), getTable()))

			throw new IllegalArgumentException(String.format("unexisting table %s on cluster %s", getTable(), getCluster()));

		token.getEnv().setAttribute("cluster", getCluster());

		token.getEnv().setAttribute(getTableParam(), getTable());

		return Arc.DEFAULT_ARC;

	}

	/**

	 * Gets the cluster.

	 *

	 * @return the cluster

	 */

	public String getCluster() {

		return cluster;

	}

	/**

	 * Sets the cluster.

	 *

	 * @param cluster

	 *            the new cluster

	 */

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

	/**

	 * Gets the table param.

	 *

	 * @return the table param

	 */

	public String getTableParam() {

		return tableParam;

	}

	/**

	 * Sets the table param.

	 *

	 * @param tableParam

	 *            the new table param

	 */

	public void setTableParam(final String tableParam) {

		this.tableParam = tableParam;

	}

	public String getTable() {

		return table;

	}

	public void setTable(String table) {

		this.table = table;

	}

}

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

import com.googlecode.sarasvati.NodeToken;

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

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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 + "'");

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

		boolean exists = hadoopService.existHbaseTable(cluster, tableName);

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

		if (exists) {

			final String tableDesc = hadoopService.describeHBaseTableConfiguration(cluster, tableName);

			token.getEnv().setAttribute(getTableConfigurationParamName(), tableDesc);

		}

		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 java.util.List;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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.msro.workflows.nodes.SimpleJobNode;

public class FetchRelClassesJobNode extends SimpleJobNode {

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

	private String relClassesProperty;

	private String relClassesName;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected String execute(final NodeToken token) throws Exception {

		String relClasses = new Gson().toJson(fetchRelClasses());

		log.info("loaded relClasses: " + relClasses);

		token.getEnv().setAttribute(getRelClassesName(), relClasses);

		return Arc.DEFAULT_ARC;

	}

	private List<String> fetchRelClasses() throws ISLookUpException {

		String xquery = getPropertyFetcher().getProperty(getRelClassesProperty());

		log.info("Loading relClasses with xquery: " + xquery);

		return serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xquery);

	}

	public String getRelClassesProperty() {

		return relClassesProperty;

	}

	public void setRelClassesProperty(final String relClassesProperty) {

		this.relClassesProperty = relClassesProperty;

	}

	public String getRelClassesName() {

		return relClassesName;

	}

	public void setRelClassesName(final String relClassesName) {

		this.relClassesName = relClassesName;

	}

}

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

import java.util.Map.Entry;

import java.util.Set;

import javax.annotation.Resource;

import com.google.common.base.Joiner;

import com.google.common.base.Splitter;

import com.google.common.collect.Sets;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.msro.rmi.MSROException;

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

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

public abstract class AbstractHBaseAdminJobNode extends SimpleJobNode {

	/** The Constant log. */

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

	private String tableColumnsParamName = "columns";

	private String tableConfigurationParamName = "tableConf";

	private String hbaseTableProperty;

	private String cluster;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	protected void beforeStart(final NodeToken token) {

		for (Entry<String, String> e : parseJsonParameters(token).entrySet()) {

			token.getEnv().setAttribute(e.getKey(), e.getValue());

		}

	}

	protected String tableName(final NodeToken token) {

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

			String table = token.getEnv().getAttribute("hbaseTable");

			log.debug("found override value in wfEnv for 'hbaseTable' param: " + table);

			return table;

		}

		return getPropertyFetcher().getProperty(getHbaseTableProperty());

	}

	protected String cluster(final NodeToken token) {

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

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

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

			return cluster;

		}

		return getCluster();

	}

	protected Set<String> getColumns(final NodeToken token) throws MSROException {

		String envCols = token.getEnv().getAttribute(getTableColumnsParamName());

		if (StringUtils.isBlank(envCols)) { throw new MSROException("cannot find table description"); }

		log.debug("using columns from env: " + envCols);

		return Sets.newHashSet(Splitter.on(",").omitEmptyStrings().split(envCols));

	}

	protected String asCSV(final Iterable<String> columns) {

		return Joiner.on(",").skipNulls().join(columns);

	}

	public String getCluster() {

		return cluster;

	}

	public void setCluster(final String cluster) {

		this.cluster = cluster;

	}

	public String getHbaseTableProperty() {

		return hbaseTableProperty;

	}

	public void setHbaseTableProperty(final String hbaseTableProperty) {

		this.hbaseTableProperty = hbaseTableProperty;

	}

	public String getTableColumnsParamName() {

		return tableColumnsParamName;

	}

	public void setTableColumnsParamName(final String tableColumnsParamName) {

		this.tableColumnsParamName = tableColumnsParamName;

	}

	public UniqueServiceLocator getServiceLocator() {

		return serviceLocator;

	}

	public String getTableConfigurationParamName() {

		return tableConfigurationParamName;

	}

	public void setTableConfigurationParamName(final String tableConfigurationParamName) {

		this.tableConfigurationParamName = tableConfigurationParamName;

	}

}

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

import java.util.List;

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;

public class GetHBaseTableDescriptionJobNode extends AbstractHBaseAdminJobNode {

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

	@Override

	protected String execute(final NodeToken token) throws Exception {

		final String tableName = tableName(token);

		final String cluster = cluster(token);

		log.info("getting table description: '" + tableName + "' on cluster: '" + cluster + "'");

		final List<String> columns = getServiceLocator().getService(HadoopService.class).describeHbaseTable(cluster, tableName);

		log.debug(String.format("table '%s': " + columns, tableName));

		token.getEnv().setAttribute(getTableColumnsParamName(), asCSV(columns));

		return Arc.DEFAULT_ARC;

	}

}

package eu.dnetlib.msro.workflows.hadoop;

import javax.annotation.Resource;

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.enabling.is.registry.rmi.ISRegistryException;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

public class IndexDSUpdateJobNode extends AsyncJobNode {

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

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	public String execute(final NodeToken token) throws Exception {

		final String dsId = token.getEnv().getAttribute("index_id");

		final String version = token.getEnv().getAttribute("index.feed.timestamp");

		log.info("updating indexDS: " + dsId + " version: " + version);

		updateIndexDS(dsId, version);

		return Arc.DEFAULT_ARC;

	}

	/**

	 * method updates the given indexDataStructureId INDEX_SIZE, INDEX_LAST_UPDATE

	 * 

	 * @param dsId

	 * @param version

	 * @return true if the update was performed successfully, false otherwise

	 * @throws ISRegistryException

	 */

	private boolean updateIndexDS(final String dsId, final String version) throws ISRegistryException {

		final String xquery = "for $x in collection('')/RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + dsId + "']"

				+ " return update value $x//INDEX_LAST_UPDATE with '" + version + "'";

		log.debug("\n\n updating indexDataStructure: " + xquery + "\n\n");

		return serviceLocator.getService(ISRegistryService.class).executeXUpdate(xquery);

	}

}

<?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>dnet45-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>2.0.0</version>

	<scm>

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

	</scm>

	<dependencies>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[4.0.0,5.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>[4.0.0,5.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-client</artifactId>

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

		</dependency>

		<dependency>

			<groupId>eu.dnetlib</groupId>

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

			<version>[6.3.24,7.0.0)</version>

			<exclusions>

				<exclusion>

					<groupId>eu.dnetlib</groupId>

					<artifactId>dnet-hadoop-commons</artifactId>

				</exclusion>

			</exclusions>

		</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>${jackson.version}</version>

		</dependency>

		<dependency>

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

			<artifactId>guava</artifactId>

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

		</dependency>

		<dependency>

			<groupId>junit</groupId>

			<artifactId>junit</artifactId>

			<version>${junit.version}</version>

			<scope>test</scope>

		</dependency>

	</dependencies>

</project>

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

import com.google.gson.Gson;

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

import eu.dnetlib.miscutils.collections.Pair;

public class Similarity {

	private Pair<String, String> pair;

	private Type type;

	public Similarity(final Pair<String, String> pair, final Type type) {

		super();

		this.setPair(pair);

		this.setType(type);

	}

	public Pair<String, String> getPair() {

		return pair;

	}

	public void setPair(final Pair<String, String> pair) {

		this.pair = pair;

	}

	public Type getType() {

		return type;

	}

	public void setType(final Type type) {

		this.type = type;

	}

	@Override

	public String toString() {

		return new Gson().toJson(this, Similarity.class);

	}

}

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 org.springframework.beans.factory.annotation.Value;

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

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.msro.workflows.nodes.SimpleJobNode;

/**

 * This jobNode resolves the original datasource ids to the list of md store ids holding the relative cleaned records.

 *

 * @author claudio

 *

 */

public class MDStoreDatasourceResolverJobNode extends SimpleJobNode {

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

	private String originalDatasourceIdsCSV;

	@Value("${dnet.openaire.dataload.datasource.mdstores.xquery}")

	private String xqueryTemplate;

	@Resource

	private UniqueServiceLocator serviceLocator;

	@Override

	public String execute(final NodeToken token) throws Exception {

		log.info("resolving MDStore ids for datasources: " + getOriginalDatasourceIdsCSV());

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

		for (final String originalId : splitter().split(getOriginalDatasourceIdsCSV())) {

			mdIds.addAll(resolveMdIds(originalId));

		}

		log.info(String.format("adding %s mdStore ids in wf env", mdIds.size()));

		token.getEnv().setAttribute("mdId", new Gson().toJson(mdIds));

		return Arc.DEFAULT_ARC;

	}

	// @Cacheable(value = "mdIds", key = "#acronym")

	public List<String> resolveMdIds(final String id) throws ISLookUpException {

		log.info("Resolving mdID for " + id + ". Cache not used.");

		final String xQuery = String.format(xqueryTemplate, id);

		final List<String> mdIds = serviceLocator.getService(ISLookUpService.class).quickSearchProfile(xQuery);

		return mdIds;

	}

	private Splitter splitter() {

		return Splitter.on(",").trimResults().omitEmptyStrings();

	}

	public String getOriginalDatasourceIdsCSV() {

		return originalDatasourceIdsCSV;

	}

	public void setOriginalDatasourceIdsCSV(final String originalDatasourceIdsCSV) {

		this.originalDatasourceIdsCSV = originalDatasourceIdsCSV;

	}

}

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

import static org.junit.Assert.assertNotNull;

import static org.junit.Assert.assertTrue;

import java.io.IOException;

import java.nio.charset.Charset;

import java.util.Queue;

import org.apache.commons.io.IOUtils;

import org.junit.Before;

import org.junit.Test;

import org.junit.Ignore;

import com.google.common.collect.Lists;

import eu.dnetlib.pace.config.DedupConfig;

@Ignore

public class DedupConfigurationOrchestrationTest {

	public DedupConfigurationOrchestration dco;

	@Before

	public void setUp() throws IOException {

		final Entity e = new Entity("result", "50", "Publication");

		final String actionSetId = "001";

		final Queue<DedupConfig> configurations = Lists.newLinkedList();

		configurations.add(DedupConfig.loadDefault());

		dco = new DedupConfigurationOrchestration(e, actionSetId, configurations);

		assertNotNull(dco);

		assertNotNull(dco.getActionSetId());

		assertNotNull(dco.getEntity());

		assertNotNull(dco.getConfigurations());

	}

	@Test

	public void testSerialization() {

		final String json = dco.toString();

		final DedupConfigurationOrchestration anotherDco = DedupConfigurationOrchestration.fromJSON(json);

		assertNotNull(anotherDco);

		assertTrue(json.equals(anotherDco.toString()));

	}

	@Test

	public void testSerializationOrgs() throws IOException {

		final Entity e = new Entity("organization", "20", "Organization");

		final String actionSetId = "001";

		final Queue<DedupConfig> configurations = Lists.newLinkedList();

		configurations.add(DedupConfig.load(IOUtils.toString(getClass().getResourceAsStream("organisation.conf.json"), Charset.forName("UTF-8"))));

		dco = new DedupConfigurationOrchestration(e, actionSetId, configurations);

		System.out.println(dco.toString());

	}

}

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.dedup;

import java.io.IOException;

import com.google.common.collect.Iterables;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import org.apache.commons.codec.binary.Base64;

import org.apache.commons.io.IOUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.junit.Test;

/**

 * Created by claudio on 05/04/16.

 */

public class SerializationTest {

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

	@Test

	public void test() throws IOException {

		final String data = Iterables.getFirst(IOUtils.readLines(getClass().getResourceAsStream("oaf_data.base64")), "");

		final byte[] oafBytes = Base64.decodeBase64(data);

		Oaf oaf = Oaf.parseFrom(oafBytes);

		JsonFormat jsonFormat = new JsonFormat();

		String asJson = jsonFormat.printToString(oaf);

		log.info(asJson);

	}

}

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;

	}

}

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.data.proto.TypeProtos.Type;

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

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

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<Similarity> combinations = SimilarityMeshBuilder.build(Type.result, list);

		System.out.println(combinations);

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

	}

}

package eu.dnetlib.data.dedup;

import java.util.List;

import java.util.Map;

import java.util.UUID;

import com.google.common.collect.Maps;

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

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

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

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

import eu.dnetlib.miscutils.datetime.DateUtils;

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;

public class DedupUserActionsDAOImpl extends AbstractDedupDAO implements DedupUserActionsDAO {

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

	@Autowired

	private DedupDbDAO dbDAO;

	@Autowired

	private DedupIndexDAO indexDAO;

	@Override

	public Map<String, List<String>> listConfigurations() throws ISLookUpException {

		final Map<String, List<String>> res = Maps.newHashMap();

		final ISLookUpService lookUpService = serviceLocator.getService(ISLookUpService.class);

		final String listEntityTypesXQuery =

				"distinct-values(for $x in //RESOURCE_PROFILE["

						+ ".//RESOURCE_TYPE/@value = 'DedupOrchestrationDSResourceType' and "

						+ ".//CONFIGURATION/@enabled='true'] return $x//ENTITY/@name/string())";

		for (final String entityType : lookUpService.quickSearchProfile(listEntityTypesXQuery)) {

			final String xquery =

					String.format(

							"for $x in //RESOURCE_PROFILE[" +

									".//RESOURCE_TYPE/@value = 'DedupOrchestrationDSResourceType' and .//ENTITY/@name='%s' ] " +

									"return $x//ACTION_SET/@id/string()", entityType);

			res.put(entityType, lookUpService.quickSearchProfile(xquery));

		}

		if (log.isDebugEnabled()) {

			log.debug("found configurations: " + res);

		}

		return res;

	}

	@Override

	public OafResult search(final String type, final String userQuery, final String actionSet, final int start, final int rows, final String fields)

			throws Exception {

		return indexDAO.search(type, userQuery, actionSet, start, rows, fields);

	}

	@Override

	public OafResult searchById(final String actionSet, final String type, final String objidentifier, final List<String> fields) throws Exception {

		return indexDAO.searchById(actionSet, type, objidentifier, fields);

	}

	@Override

	public boolean commit(final SimilarityGroup group) throws Exception {

		if (StringUtils.isBlank(group.getActionSet())) throw new IllegalArgumentException("missing actionset");

		group.setId(UUID.randomUUID().toString());

		group.setDate(InputDocumentFactory.getParsedDateField(DateUtils.now_ISO8601()));

		return dbDAO.commit(group) && indexDAO.commit(group);

	}

}

package eu.dnetlib.data.dedup;

import com.google.common.collect.Lists;

import eu.dnetlib.data.proto.OafProtos.Oaf;

import com.google.common.base.Function;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.junit.Before;

import org.junit.Test;

import java.util.Map;

public class DedupInspectorFunctionsTest {

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

    private String orgFromIndex = "<record>CAESigMIFCK7ARK4ARIlCiNJbnRlcm5hdGlvbmFsIFRlbm5pcyBGZWRlcmF0aW9uIEx0ZCoHCgVm\r\nYWxzZTIHCgVmYWxzZToHCgVmYWxzZUIHCgVmYWxzZUoHCgVmYWxzZVIHCgVmYWxzZVoHCgVmYWxz\r\nZWIHCgVmYWxzZWoHCgVmYWxzZXIHCgVmYWxzZYIBNAoCR0ISDlVuaXRlZCBLaW5nZG9tGg5kbmV0\r\nOmNvdW50cmllcyIOZG5ldDpjb3VudHJpZXNCMnJjdWtfX19fX19fXzo6ODYxOTIxQzMtNjcyMy00\r\nOUMwLUIyQ0UtRDcyODJBOUMxRTk2SkkKMTEwfG9wZW5haXJlX19fXzo6YWIyZDMzMTA3NDFlYTgw\r\nZDNiODcyNmY2NTE1MDI4NTgSFFJlc2VhcmNoIENvdW5jaWxzIFVLWgoyMDE3LTExLTA0YjEyMHxy\r\nY3VrX19fX19fX186OjVjYjFmZTZhYjg1NDcwMTBiMjJkOTAyN2U3MjUyMzVkagoyMDE5LTA1LTE5\r\nIqsBCAEQARoDMC45IiRkZWR1cC1zaW1pbGFyaXR5LW9yZ2FuaXphdGlvbi1zaW1wbGUqegoic3lz\r\naW1wb3J0OmNyb3Nzd2FsazplbnRpdHlyZWdpc3RyeRIic3lzaW1wb3J0OmNyb3Nzd2FsazplbnRp\r\ndHlyZWdpc3RyeRoXZG5ldDpwcm92ZW5hbmNlX2FjdGlvbnMiF2RuZXQ6cHJvdmVuYW5jZV9hY3Rp\r\nb25z\r\n\n</record>";

    private DedupIndexDAO dao;

    @Before

    public void setUp() {

        dao = new DedupIndexDAO();

    }

    @Test

    public void test_1() {

        Oaf oaf = dao.getXml2OafFunction().apply(orgFromIndex);

        System.out.println(oaf);

        Map<String, String> map = dao.getOaf2FieldMapFunction("organization", Lists.newArrayList("legalname", "legalshortname", "country", "websiteurl", "provenance")).apply(oaf);

        System.out.println(map);

    }

}

package eu.dnetlib.functionality.modular.ui.workflows.values;

import java.util.List;

import java.util.Map;

import javax.annotation.Resource;

import com.google.common.collect.Lists;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

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

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

public class ListHBaseMappingTitleValues extends ValidNodeValuesFetcher {

	@Resource

	private UniqueServiceLocator serviceLocator;

	private String sourceFormat;

	@Override

	protected List<DnetParamValue> obtainValues(final Map<String, String> params) throws Exception {

		final String xquery =

				"for $x in /RESOURCE_PROFILE["

						+ ".//RESOURCE_TYPE/@value='TransformationRuleDSResourceType' and "

						+ ".//SOURCE_METADATA_FORMAT/@name = '" + getSourceFormat() + "' and "

						+ ".//SOURCE_METADATA_FORMAT/@layout = 'store' and "

						+ ".//SOURCE_METADATA_FORMAT/@interpretation = 'cleaned'] "

						+ "return concat($x//RESOURCE_IDENTIFIER/@value, ' @@@ ', $x//SCRIPT/TITLE/text())";

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

		final List<DnetParamValue> values = Lists.newArrayList();

		for (final String s : result) {

			final String[] arr = s.split("@@@");

			values.add(new DnetParamValue(arr[0].trim(), arr[1].trim()));

		}

		return values;

	}

	public String getSourceFormat() {

		return sourceFormat;

	}

	@Required

	public void setSourceFormat(final String sourceFormat) {

		this.sourceFormat = sourceFormat;

	}

}

package eu.dnetlib.msro.workflows.actions;

import java.util.List;

import java.util.Map;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

import com.googlecode.sarasvati.Arc;

import com.googlecode.sarasvati.NodeToken;

import eu.dnetlib.actionmanager.set.RawSet;

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

import eu.dnetlib.enabling.locators.UniqueServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.workflows.dedup.conf.DedupConfigurationOrchestration;

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

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;

/**

 * The Class PrepareConfiguredActionSetJobNode.

 */

public class PrepareConfiguredActionSetJobNode extends SimpleJobNode {

	/**

	 * logger.

	 */

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

	/**