merge upstream

dhp-build/dhp-build-assembly-resources/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-build</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
 
     <artifactId>dhp-build-assembly-resources</artifactId>

dhp-build/dhp-build-properties-maven-plugin/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-build</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
 
     <artifactId>dhp-build-properties-maven-plugin</artifactId>

dhp-build/dhp-code-style/pom.xml

@@ -1,13 +1,11 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
+<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/xsd/maven-4.0.0.xsd">
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>eu.dnetlib.dhp</groupId>
     <artifactId>dhp-code-style</artifactId>
-    <version>1.1.7-SNAPSHOT</version>
+    <version>1.1.8-SNAPSHOT</version>
 
     <packaging>jar</packaging>
 

dhp-build/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>eu.dnetlib.dhp</groupId>
 		<artifactId>dhp</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.8-SNAPSHOT</version>
 	</parent>
 	<artifactId>dhp-build</artifactId>
 	<packaging>pom</packaging>

dhp-common/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>eu.dnetlib.dhp</groupId>
 		<artifactId>dhp</artifactId>
-		<version>1.1.7-SNAPSHOT</version>
+		<version>1.1.8-SNAPSHOT</version>
 		<relativePath>../</relativePath>
 	</parent>
 

dhp-schemas/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>
 

dhp-schemas/src/main/java/eu/dnetlib/dhp/schema/common/ModelConstants.java

@@ -49,6 +49,9 @@ public class ModelConstants {
 	public static final String HAS_PARTICIPANT = "hasParticipant";
 	public static final String IS_PARTICIPANT = "isParticipant";
 
+	public static final String UNKNOWN = "UNKNOWN";
+	public static final String NOT_AVAILABLE = "not available";
+
 	public static final Qualifier PUBLICATION_DEFAULT_RESULTTYPE = qualifier(
 		PUBLICATION_RESULTTYPE_CLASSID, PUBLICATION_RESULTTYPE_CLASSID,
 		DNET_RESULT_TYPOLOGIES, DNET_RESULT_TYPOLOGIES);

dhp-workflows/dhp-actionmanager/pom.xml

@@ -1,11 +1,10 @@
 <?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"
+<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/xsd/maven-4.0.0.xsd">
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-workflows</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <artifactId>dhp-actionmanager</artifactId>
 

dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/migration/MigrateActionSet.java

@@ -84,8 +84,11 @@ public class MigrateActionSet {
 		final List<Path> sourcePaths = getSourcePaths(sourceNN, isLookUp);
 		log
 			.info(
-				"paths to process:\n{}",
+				"paths to process:\n{}", sourcePaths
-				sourcePaths.stream().map(p -> p.toString()).collect(Collectors.joining("\n")));
+					.stream()
+					.map(p -> p.toString())
+					.collect(Collectors.joining("\n")));
+
 		for (Path source : sourcePaths) {
 
 			if (!sourceFS.exists(source)) {
@@ -119,9 +122,8 @@ public class MigrateActionSet {
 			}
 		}
 
-		props
+		final String targetPathsCsv = targetPaths.stream().map(p -> p.toString()).collect(Collectors.joining(","));
-			.setProperty(
+		props.setProperty(TARGET_PATHS, targetPathsCsv);
-				TARGET_PATHS, targetPaths.stream().map(p -> p.toString()).collect(Collectors.joining(",")));
 		File file = new File(System.getProperty("oozie.action.output.properties"));
 
 		try (OutputStream os = new FileOutputStream(file)) {

dhp-workflows/dhp-actionmanager/src/main/java/eu/dnetlib/dhp/actionmanager/migration/ProtoConverter.java

@@ -1,12 +1,10 @@
 
 package eu.dnetlib.dhp.actionmanager.migration;
 
-import static eu.dnetlib.data.proto.KindProtos.Kind.entity;
+import static eu.dnetlib.dhp.schema.common.ModelConstants.*;
-import static eu.dnetlib.data.proto.KindProtos.Kind.relation;
-import static eu.dnetlib.data.proto.TypeProtos.*;
-import static eu.dnetlib.data.proto.TypeProtos.Type.*;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -21,10 +19,6 @@ import eu.dnetlib.dhp.schema.oaf.*;
 
 public class ProtoConverter implements Serializable {
 
-	public static final String UNKNOWN = "UNKNOWN";
-	public static final String NOT_AVAILABLE = "not available";
-	public static final String DNET_ACCESS_MODES = "dnet:access_modes";
-
 	public static Oaf convert(OafProtos.Oaf oaf) {
 		try {
 			switch (oaf.getKind()) {
@@ -64,6 +58,7 @@ public class ProtoConverter implements Serializable {
 			case result:
 				final Result r = convertResult(oaf);
 				r.setInstance(convertInstances(oaf));
+				r.setExternalReference(convertExternalRefs(oaf));
 				return r;
 			case project:
 				return convertProject(oaf);
@@ -94,13 +89,44 @@ public class ProtoConverter implements Serializable {
 		i.setHostedby(mapKV(ri.getHostedby()));
 		i.setInstancetype(mapQualifier(ri.getInstancetype()));
 		i.setLicense(mapStringField(ri.getLicense()));
-		i.setUrl(ri.getUrlList());
+		i
+			.setUrl(
+				ri.getUrlList() != null ? ri
+					.getUrlList()
+					.stream()
+					.distinct()
+					.collect(Collectors.toCollection(ArrayList::new)) : null);
 		i.setRefereed(mapStringField(ri.getRefereed()));
 		i.setProcessingchargeamount(mapStringField(ri.getProcessingchargeamount()));
 		i.setProcessingchargecurrency(mapStringField(ri.getProcessingchargecurrency()));
 		return i;
 	}
 
+	private static List<ExternalReference> convertExternalRefs(OafProtos.Oaf oaf) {
+		ResultProtos.Result r = oaf.getEntity().getResult();
+		if (r.getExternalReferenceCount() > 0) {
+			return r
+				.getExternalReferenceList()
+				.stream()
+				.map(e -> convertExtRef(e))
+				.collect(Collectors.toList());
+		}
+		return Lists.newArrayList();
+	}
+
+	private static ExternalReference convertExtRef(ResultProtos.Result.ExternalReference e) {
+		ExternalReference ex = new ExternalReference();
+		ex.setUrl(e.getUrl());
+		ex.setSitename(e.getSitename());
+		ex.setRefidentifier(e.getRefidentifier());
+		ex.setQuery(e.getQuery());
+		ex.setQualifier(mapQualifier(e.getQualifier()));
+		ex.setLabel(e.getLabel());
+		ex.setDescription(e.getDescription());
+		ex.setDataInfo(ex.getDataInfo());
+		return ex;
+	}
+
 	private static Organization convertOrganization(OafProtos.Oaf oaf) {
 		final OrganizationProtos.Organization.Metadata m = oaf.getEntity().getOrganization().getMetadata();
 		final Organization org = setOaf(new Organization(), oaf);

dhp-workflows/dhp-aggregation/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp-workflows</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <artifactId>dhp-aggregation</artifactId>
     

dhp-workflows/dhp-broker-events/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/model/Event.java

@@ -0,0 +1,104 @@
+
+package eu.dnetlib.dhp.broker.model;
+
+import java.util.Map;
+
+public class Event {
+
+	private String eventId;
+
+	private String producerId;
+
+	private String topic;
+
+	private String payload;
+
+	private Long creationDate;
+
+	private Long expiryDate;
+
+	private boolean instantMessage;
+
+	private Map<String, Object> map;
+
+	public Event() {
+	}
+
+	public Event(final String producerId, final String eventId, final String topic, final String payload,
+		final Long creationDate, final Long expiryDate,
+		final boolean instantMessage,
+		final Map<String, Object> map) {
+		this.producerId = producerId;
+		this.eventId = eventId;
+		this.topic = topic;
+		this.payload = payload;
+		this.creationDate = creationDate;
+		this.expiryDate = expiryDate;
+		this.instantMessage = instantMessage;
+		this.map = map;
+	}
+
+	public String getProducerId() {
+		return this.producerId;
+	}
+
+	public void setProducerId(final String producerId) {
+		this.producerId = producerId;
+	}
+
+	public String getEventId() {
+		return this.eventId;
+	}
+
+	public void setEventId(final String eventId) {
+		this.eventId = eventId;
+	}
+
+	public String getTopic() {
+		return this.topic;
+	}
+
+	public void setTopic(final String topic) {
+		this.topic = topic;
+	}
+
+	public String getPayload() {
+		return this.payload;
+	}
+
+	public void setPayload(final String payload) {
+		this.payload = payload;
+	}
+
+	public Long getCreationDate() {
+		return this.creationDate;
+	}
+
+	public void setCreationDate(final Long creationDate) {
+		this.creationDate = creationDate;
+	}
+
+	public Long getExpiryDate() {
+		return this.expiryDate;
+	}
+
+	public void setExpiryDate(final Long expiryDate) {
+		this.expiryDate = expiryDate;
+	}
+
+	public boolean isInstantMessage() {
+		return this.instantMessage;
+	}
+
+	public void setInstantMessage(final boolean instantMessage) {
+		this.instantMessage = instantMessage;
+	}
+
+	public Map<String, Object> getMap() {
+		return this.map;
+	}
+
+	public void setMap(final Map<String, Object> map) {
+		this.map = map;
+	}
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/model/EventFactory.java

@@ -0,0 +1,140 @@
+
+package eu.dnetlib.dhp.broker.model;
+
+import java.text.ParseException;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.time.DateUtils;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.schema.oaf.Author;
+import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import eu.dnetlib.dhp.schema.oaf.Result;
+import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+
+public class EventFactory {
+
+	private final static String PRODUCER_ID = "OpenAIRE";
+
+	private static final int TTH_DAYS = 365;
+
+	private final static String[] DATE_PATTERNS = {
+		"yyyy-MM-dd"
+	};
+
+	public static Event newBrokerEvent(final Result source, final Result target, final UpdateInfo<?> updateInfo) {
+
+		final long now = new Date().getTime();
+
+		final Event res = new Event();
+
+		final Map<String, Object> map = createMapFromResult(target, source, updateInfo);
+
+		final String payload = createPayload(target, updateInfo);
+
+		final String eventId = calculateEventId(
+			updateInfo.getTopic(), target.getOriginalId().get(0), updateInfo.getHighlightValueAsString());
+
+		res.setEventId(eventId);
+		res.setProducerId(PRODUCER_ID);
+		res.setPayload(payload);
+		res.setMap(map);
+		res.setTopic(updateInfo.getTopic());
+		res.setCreationDate(now);
+		res.setExpiryDate(calculateExpiryDate(now));
+		res.setInstantMessage(false);
+		return res;
+	}
+
+	private static String createPayload(final Result result, final UpdateInfo<?> updateInfo) {
+		final OpenAireEventPayload payload = new OpenAireEventPayload();
+		// TODO
+
+		updateInfo.compileHighlight(payload);
+
+		return payload.toJSON();
+	}
+
+	private static Map<String, Object> createMapFromResult(final Result oaf, final Result source,
+		final UpdateInfo<?> updateInfo) {
+		final Map<String, Object> map = new HashMap<>();
+
+		final List<KeyValue> collectedFrom = oaf.getCollectedfrom();
+		if (collectedFrom.size() == 1) {
+			map.put("target_datasource_id", collectedFrom.get(0).getKey());
+			map.put("target_datasource_name", collectedFrom.get(0).getValue());
+		}
+
+		final List<String> ids = oaf.getOriginalId();
+		if (ids.size() > 0) {
+			map.put("target_publication_id", ids.get(0));
+		}
+
+		final List<StructuredProperty> titles = oaf.getTitle();
+		if (titles.size() > 0) {
+			map.put("target_publication_title", titles.get(0));
+		}
+
+		final long date = parseDateTolong(oaf.getDateofacceptance().getValue());
+		if (date > 0) {
+			map.put("target_dateofacceptance", date);
+		}
+
+		final List<StructuredProperty> subjects = oaf.getSubject();
+		if (subjects.size() > 0) {
+			map
+				.put(
+					"target_publication_subject_list",
+					subjects.stream().map(StructuredProperty::getValue).collect(Collectors.toList()));
+		}
+
+		final List<Author> authors = oaf.getAuthor();
+		if (authors.size() > 0) {
+			map
+				.put(
+					"target_publication_author_list",
+					authors.stream().map(Author::getFullname).collect(Collectors.toList()));
+		}
+
+		// PROVENANCE INFO
+		map.put("trust", updateInfo.getTrust());
+		final List<KeyValue> sourceCollectedFrom = source.getCollectedfrom();
+		if (sourceCollectedFrom.size() == 1) {
+			map.put("provenance_datasource_id", sourceCollectedFrom.get(0).getKey());
+			map.put("provenance_datasource_name", sourceCollectedFrom.get(0).getValue());
+		}
+		map.put("provenance_publication_id_list", source.getOriginalId());
+
+		return map;
+	}
+
+	private static String calculateEventId(final String topic, final String publicationId, final String value) {
+		return "event-"
+			+ DigestUtils.md5Hex(topic).substring(0, 6) + "-"
+			+ DigestUtils.md5Hex(publicationId).substring(0, 8) + "-"
+			+ DigestUtils.md5Hex(value).substring(0, 8);
+	}
+
+	private static long calculateExpiryDate(final long now) {
+		return now + TTH_DAYS * 24 * 60 * 60 * 1000;
+	}
+
+	private static long parseDateTolong(final String date) {
+		if (StringUtils.isBlank(date)) {
+			return -1;
+		}
+		try {
+			return DateUtils.parseDate(date, DATE_PATTERNS).getTime();
+		} catch (final ParseException e) {
+			return -1;
+		}
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/GenerateEventsApplication.java

@@ -0,0 +1,112 @@
+
+package eu.dnetlib.dhp.broker.oa;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.broker.model.Event;
+import eu.dnetlib.dhp.broker.model.EventFactory;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingAbstract;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingAuthorOrcid;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingOpenAccess;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingPid;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingProject;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingPublicationDate;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMissingSubject;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMoreOpenAccess;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMorePid;
+import eu.dnetlib.dhp.broker.oa.util.EnrichMoreSubject;
+import eu.dnetlib.dhp.broker.oa.util.UpdateInfo;
+import eu.dnetlib.dhp.common.HdfsSupport;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class GenerateEventsApplication {
+
+	private static final Logger log = LoggerFactory.getLogger(GenerateEventsApplication.class);
+
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	public static void main(final String[] args) throws Exception {
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
+			IOUtils
+				.toString(
+					GenerateEventsApplication.class
+						.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/merge_claims_parameters.json")));
+		parser.parseArgument(args);
+
+		final Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String graphPath = parser.get("graphPath");
+		log.info("graphPath: {}", graphPath);
+
+		final String eventsPath = parser.get("eventsPath");
+		log.info("eventsPath: {}", eventsPath);
+
+		final SparkConf conf = new SparkConf();
+		runWithSparkSession(conf, isSparkSessionManaged, spark -> {
+			removeOutputDir(spark, eventsPath);
+			generateEvents(spark, graphPath, eventsPath);
+		});
+
+	}
+
+	private static void removeOutputDir(final SparkSession spark, final String path) {
+		HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
+	}
+
+	private static void generateEvents(final SparkSession spark, final String graphPath, final String eventsPath) {
+		// TODO
+	}
+
+	private List<Event> generateEvents(final Result... children) {
+		final List<Event> list = new ArrayList<>();
+
+		for (final Result source : children) {
+			for (final Result target : children) {
+				if (source != target) {
+					list
+						.addAll(
+							findUpdates(source, target)
+								.stream()
+								.map(info -> EventFactory.newBrokerEvent(source, target, info))
+								.collect(Collectors.toList()));
+				}
+			}
+		}
+
+		return list;
+	}
+
+	private List<UpdateInfo<?>> findUpdates(final Result source, final Result target) {
+		final List<UpdateInfo<?>> list = new ArrayList<>();
+		list.addAll(EnrichMissingAbstract.findUpdates(source, target));
+		list.addAll(EnrichMissingAuthorOrcid.findUpdates(source, target));
+		list.addAll(EnrichMissingOpenAccess.findUpdates(source, target));
+		list.addAll(EnrichMissingPid.findUpdates(source, target));
+		list.addAll(EnrichMissingProject.findUpdates(source, target));
+		list.addAll(EnrichMissingPublicationDate.findUpdates(source, target));
+		list.addAll(EnrichMissingSubject.findUpdates(source, target));
+		list.addAll(EnrichMoreOpenAccess.findUpdates(source, target));
+		list.addAll(EnrichMorePid.findUpdates(source, target));
+		list.addAll(EnrichMoreSubject.findUpdates(source, target));
+		return list;
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingAbstract.java

@@ -0,0 +1,31 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingAbstract extends UpdateInfo<String> {
+
+	public static List<EnrichMissingAbstract> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMissingAbstract(final String highlightValue, final float trust) {
+		super("ENRICH/MISSING/ABSTRACT", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getAbstracts().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingAuthorOrcid.java

@@ -0,0 +1,31 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingAuthorOrcid extends UpdateInfo<String> {
+
+	public static List<EnrichMissingAuthorOrcid> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMissingAuthorOrcid(final String highlightValue, final float trust) {
+		super("ENRICH/MISSING/AUTHOR/ORCID", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		// TODO
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingOpenAccess.java

@@ -0,0 +1,32 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.Instance;
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingOpenAccess extends UpdateInfo<Instance> {
+
+	public static List<EnrichMissingOpenAccess> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMissingOpenAccess(final Instance highlightValue, final float trust) {
+		super("ENRICH/MISSING/OPENACCESS_VERSION", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getInstances().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue().getUrl();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingPid.java

@@ -0,0 +1,32 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.broker.objects.Pid;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPid extends UpdateInfo<Pid> {
+
+	public static List<EnrichMissingPid> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMissingPid(final Pid highlightValue, final float trust) {
+		super("ENRICH/MISSING/PID", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getPids().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue().getType() + "::" + getHighlightValue().getValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingProject.java

@@ -0,0 +1,33 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.broker.objects.Project;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingProject extends UpdateInfo<Project> {
+
+	public static List<EnrichMissingProject> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMissingProject(final Project highlightValue, final float trust) {
+		super("ENRICH/MISSING/PROJECT", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getProjects().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue().getFunder() + "::" + getHighlightValue().getFundingProgram()
+			+ getHighlightValue().getCode();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingPublicationDate.java

@@ -0,0 +1,31 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingPublicationDate extends UpdateInfo<String> {
+
+	public static List<EnrichMissingPublicationDate> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMissingPublicationDate(final String highlightValue, final float trust) {
+		super("ENRICH/MISSING/PUBLICATION_DATE", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().setPublicationdate(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMissingSubject.java

@@ -0,0 +1,36 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMissingSubject extends UpdateInfo<String> {
+
+	public static List<EnrichMissingSubject> findUpdates(final Result source, final Result target) {
+		// MESHEUROPMC
+		// ARXIV
+		// JEL
+		// DDC
+		// ACM
+
+		return Arrays.asList();
+	}
+
+	private EnrichMissingSubject(final String subjectClassification, final String highlightValue, final float trust) {
+		super("ENRICH/MISSING/SUBJECT/" + subjectClassification, highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getSubjects().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMoreOpenAccess.java

@@ -0,0 +1,32 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.Instance;
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMoreOpenAccess extends UpdateInfo<Instance> {
+
+	public static List<EnrichMoreOpenAccess> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMoreOpenAccess(final Instance highlightValue, final float trust) {
+		super("ENRICH/MORE/OPENACCESS_VERSION", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getInstances().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue().getUrl();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMorePid.java

@@ -0,0 +1,32 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.broker.objects.Pid;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMorePid extends UpdateInfo<Pid> {
+
+	public static List<EnrichMorePid> findUpdates(final Result source, final Result target) {
+		// return Arrays.asList(new EnrichMissingAbstract("xxxxxxx", 0.9f));
+		return Arrays.asList();
+	}
+
+	private EnrichMorePid(final Pid highlightValue, final float trust) {
+		super("ENRICH/MORE/PID", highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getPids().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue().getType() + "::" + getHighlightValue().getValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/EnrichMoreSubject.java

@@ -0,0 +1,36 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+import eu.dnetlib.dhp.schema.oaf.Result;
+
+public class EnrichMoreSubject extends UpdateInfo<String> {
+
+	public static List<EnrichMoreSubject> findUpdates(final Result source, final Result target) {
+		// MESHEUROPMC
+		// ARXIV
+		// JEL
+		// DDC
+		// ACM
+
+		return Arrays.asList();
+	}
+
+	private EnrichMoreSubject(final String subjectClassification, final String highlightValue, final float trust) {
+		super("ENRICH/MORE/SUBJECT/" + subjectClassification, highlightValue, trust);
+	}
+
+	@Override
+	public void compileHighlight(final OpenAireEventPayload payload) {
+		payload.getHighlight().getSubjects().add(getHighlightValue());
+	}
+
+	@Override
+	public String getHighlightValueAsString() {
+		return getHighlightValue();
+	}
+
+}

dhp-workflows/dhp-broker-events/src/main/java/eu/dnetlib/dhp/broker/oa/util/UpdateInfo.java

@@ -0,0 +1,36 @@
+
+package eu.dnetlib.dhp.broker.oa.util;
+
+import eu.dnetlib.broker.objects.OpenAireEventPayload;
+
+public abstract class UpdateInfo<T> {
+
+	private final String topic;
+
+	private final T highlightValue;
+
+	private final float trust;
+
+	protected UpdateInfo(final String topic, final T highlightValue, final float trust) {
+		this.topic = topic;
+		this.highlightValue = highlightValue;
+		this.trust = trust;
+	}
+
+	public T getHighlightValue() {
+		return highlightValue;
+	}
+
+	public float getTrust() {
+		return trust;
+	}
+
+	public String getTopic() {
+		return topic;
+	}
+
+	abstract public void compileHighlight(OpenAireEventPayload payload);
+
+	abstract public String getHighlightValueAsString();
+
+}

dhp-workflows/dhp-dedup-openaire/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>dhp-dedup-openaire</artifactId>

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCreateMergeRels.java

@@ -137,10 +137,14 @@ public class SparkCreateMergeRels extends AbstractSparkAction {
 	}
 
 	private Relation rel(String source, String target, String relClass, DedupConfig dedupConf) {
+
+		String entityType = dedupConf.getWf().getEntityType();
+
 		Relation r = new Relation();
 		r.setSource(source);
 		r.setTarget(target);
 		r.setRelClass(relClass);
+		r.setRelType(entityType + entityType.substring(0, 1).toUpperCase() + entityType.substring(1));
 		r.setSubRelType("dedup");
 
 		DataInfo info = new DataInfo();

dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java

@@ -86,7 +86,8 @@ public class SparkPropagateRelation extends AbstractSparkAction {
 			mergedIds,
 			FieldType.TARGET,
 			getFixRelFn(FieldType.TARGET))
-				.filter(SparkPropagateRelation::containsDedup);
+				.filter(SparkPropagateRelation::containsDedup)
+				.distinct();
 
 		Dataset<Relation> updated = processDataset(
 			processDataset(rels, mergedIds, FieldType.SOURCE, getDeletedFn()),

dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/SparkDedupTest.java

@@ -12,12 +12,14 @@ import java.io.Serializable;
 import java.net.URISyntaxException;
 import java.nio.file.Paths;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.JavaPairRDD;
 import org.apache.spark.api.java.JavaRDD;
 import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.function.MapFunction;
 import org.apache.spark.api.java.function.PairFunction;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Encoders;
@@ -450,6 +452,22 @@ public class SparkDedupTest implements Serializable {
 		assertEquals(updated, deletedbyinference);
 	}
 
+	@Test
+	@Order(6)
+	public void testRelations() throws Exception {
+		testUniqueness("/eu/dnetlib/dhp/dedup/test/relation_1.json", 12, 10);
+		testUniqueness("/eu/dnetlib/dhp/dedup/test/relation_2.json", 10, 2);
+	}
+
+	private void testUniqueness(String path, int expected_total, int expected_unique) {
+		Dataset<Relation> rel = spark.read()
+				.textFile(getClass().getResource(path).getPath())
+				.map((MapFunction<String, Relation>) s -> new ObjectMapper().readValue(s, Relation.class), Encoders.bean(Relation.class));
+
+		assertEquals(expected_total, rel.count());
+		assertEquals(expected_unique, rel.distinct().count());
+	}
+
 	@AfterAll
 	public static void finalCleanUp() throws IOException {
 		FileUtils.deleteDirectory(new File(testOutputBasePath));

dhp-workflows/dhp-dedup-openaire/src/test/resources/eu/dnetlib/dhp/dedup/test/relation_1.json

@@ -0,0 +1,12 @@
+{"collectedfrom":[{"key":"10|driver______::bee53aa31dc2cbb538c10c2b65fa5824","value":"DOAJ-Articles"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|doajarticles::40c7b1dfa18c3693d374dafd21ef852f","subRelType":"provision","target":"10|doajarticles::618df40624078491acfd93ca3ff6921c"}
+{"collectedfrom":[{"key":"10|driver______::bee53aa31dc2cbb538c10c2b65fa5824","value":"DOAJ-Articles"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|doajarticles::0b4e756a73338f60b84de98d080f6422","subRelType":"provision","target":"10|doajarticles::6d01e689db13b6977b411f4170b6143b"}
+{"collectedfrom":[{"key":"10|driver______::bee53aa31dc2cbb538c10c2b65fa5824","value":"DOAJ-Articles"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|doajarticles::fe2f7c9d350b9c5aa658ec384d761e33","subRelType":"provision","target":"10|doajarticles::9b8a956b0703854ba79e52ddf7dc552e"}
+{"collectedfrom":[{"key":"10|driver______::bee53aa31dc2cbb538c10c2b65fa5824","value":"DOAJ-Articles"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|doajarticles::a116734108ba011ef715b012f095e3f5","subRelType":"provision","target":"10|doajarticles::c5de04b1a35da2cc4468e299bc9ffa16"}
+{"collectedfrom":[{"key":"10|openaire____::47ce9e9f4fad46e732cff06419ecaabb","value":"OpenDOAR"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|opendoar____::8b83abbbcad5496fe43cda88d0045aa4","subRelType":"provision","target":"10|opendoar____::6855456e2fe46a9d49d3d3af4f57443d"}
+{"collectedfrom":[{"key":"10|openaire____::47ce9e9f4fad46e732cff06419ecaabb","value":"OpenDOAR"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|opendoar____::88034de0247d9d36e22783e9319c5ba3","subRelType":"provision","target":"10|opendoar____::c17028c9b6e0c5deaad29665d582284a"}
+{"collectedfrom":[{"key":"10|openaire____::47ce9e9f4fad46e732cff06419ecaabb","value":"OpenDOAR"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|opendoar____::dfb21c796f33e9acf505cc960a3d8d2c","subRelType":"provision","target":"10|opendoar____::dfa037a53e121ecc9e0926800c3e814e"}
+{"collectedfrom":[{"key":"10|openaire____::21f8a223b9925c2f87c404096080b046","value":"Registry of Research Data Repository"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|re3data_____::b526b1aa1562038881a31be59896985f","subRelType":"provision","target":"10|re3data_____::2e457773b62df3534cc04441bf406a70"}
+{"collectedfrom":[{"key":"10|openaire____::21f8a223b9925c2f87c404096080b046","value":"Registry of Research Data Repository"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|re3data_____::6b306183bc051b5aaa5376f2fab6e6e5","subRelType":"provision","target":"10|re3data_____::6371ff9ee1ec7073416cb83c868b10a3"}
+{"collectedfrom":[{"key":"10|openaire____::21f8a223b9925c2f87c404096080b046","value":"Registry of Research Data Repository"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|re3data_____::0f697c2543a43bc0da793bf78ecd4996","subRelType":"provision","target":"10|re3data_____::770ef1f8eb03f174c0add746523c6f28"}
+{"collectedfrom":[{"key":"10|openaire____::21f8a223b9925c2f87c404096080b046","value":"Registry of Research Data Repository"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|re3data_____::0f697c2543a43bc0da793bf78ecd4996","subRelType":"provision","target":"10|re3data_____::770ef1f8eb03f174c0add746523c6f28"}
+{"collectedfrom":[{"key":"10|openaire____::21f8a223b9925c2f87c404096080b046","value":"Registry of Research Data Repository"}],"dataInfo":{"deletedbyinference":false,"inferred":false,"invisible":false,"provenanceaction":{"classid":"sysimport:crosswalk:entityregistry","classname":"sysimport:crosswalk:entityregistry","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":"0.9"},"lastupdatetimestamp":1588608946167,"relClass":"provides","relType":"datasourceOrganization","source":"20|re3data_____::0f697c2543a43bc0da793bf78ecd4996","subRelType":"provision","target":"10|re3data_____::770ef1f8eb03f174c0add746523c6f28"}

dhp-workflows/dhp-dedup-openaire/src/test/resources/eu/dnetlib/dhp/dedup/test/relation_2.json

@@ -0,0 +1,10 @@
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681628"}
+{"collectedfrom":null,"dataInfo":{"deletedbyinference":false,"inferenceprovenance":"decisiontree-dedup-test","inferred":true,"invisible":false,"provenanceaction":{"classid":"sysimport:dedup","classname":"sysimport:dedup","schemeid":"dnet:provenanceActions","schemename":"dnet:provenanceActions"},"trust":null},"lastupdatetimestamp":null,"relClass":"isMergedIn","relType":"resultResult","source":"50|dedup_wf_001::498c4e6cfff198831b488a6c62221241","subRelType":"dedup","target":"50|doiboost____::8e5e14d80d0f2ebe6a6a55d972681629"}

dhp-workflows/dhp-dedup-scholexplorer/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-distcp/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-mapper/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/OafToOafMapper.java

@@ -5,10 +5,8 @@ import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.createOpenaireId
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.field;
 import static eu.dnetlib.dhp.schema.common.ModelConstants.*;
 
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Arrays;
+import java.util.stream.Collectors;
-import java.util.List;
-import java.util.Map;
 
 import org.apache.commons.lang3.StringUtils;
 import org.dom4j.Document;
@@ -115,12 +113,17 @@ public class OafToOafMapper extends AbstractMdRecordToOafMapper {
 			.setProcessingchargecurrency(
 				field(doc.valueOf("//oaf:processingchargeamount/@currency"), info));
 
-		for (final Object o : doc.selectNodes("//dc:identifier")) {
+		List<Node> nodes = Lists.newArrayList(doc.selectNodes("//dc:identifier"));
-			final String url = ((Node) o).getText().trim();
+		instance
-			if (url.startsWith("http")) {
+			.setUrl(
-				instance.setUrl(Arrays.asList(url));
+				nodes
-			}
+					.stream()
-		}
+					.filter(n -> StringUtils.isNotBlank(n.getText()))
+					.map(n -> n.getText().trim())
+					.filter(u -> u.startsWith("http"))
+					.distinct()
+					.collect(Collectors.toCollection(ArrayList::new)));
+
 		return Lists.newArrayList(instance);
 	}
 

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/OdfToOafMapper.java

@@ -6,10 +6,7 @@ import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.field;
 import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.structuredProperty;
 import static eu.dnetlib.dhp.schema.common.ModelConstants.*;
 
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
 
 import org.apache.commons.lang3.StringUtils;
 import org.dom4j.Document;
@@ -80,6 +77,7 @@ public class OdfToOafMapper extends AbstractMdRecordToOafMapper {
 		final KeyValue hostedby) {
 
 		final Instance instance = new Instance();
+		final Set<String> url = new HashSet<>();
 		instance.setUrl(new ArrayList<>());
 		instance
 			.setInstancetype(
@@ -100,17 +98,18 @@ public class OdfToOafMapper extends AbstractMdRecordToOafMapper {
 				field(doc.valueOf("//oaf:processingchargeamount/@currency"), info));
 
 		for (final Object o : doc.selectNodes("//datacite:alternateIdentifier[@alternateIdentifierType='URL']")) {
-			instance.getUrl().add(((Node) o).getText().trim());
+			url.add(((Node) o).getText().trim());
 		}
 		for (final Object o : doc.selectNodes("//datacite:identifier[@identifierType='URL']")) {
-			instance.getUrl().add(((Node) o).getText().trim());
+			url.add(((Node) o).getText().trim());
 		}
 		for (final Object o : doc.selectNodes("//datacite:alternateIdentifier[@alternateIdentifierType='DOI']")) {
-			instance.getUrl().add(HTTP_DX_DOI_PREIFX + ((Node) o).getText().trim());
+			url.add(HTTP_DX_DOI_PREIFX + ((Node) o).getText().trim());
 		}
 		for (final Object o : doc.selectNodes("//datacite:identifier[@identifierType='DOI']")) {
-			instance.getUrl().add(HTTP_DX_DOI_PREIFX + ((Node) o).getText().trim());
+			url.add(HTTP_DX_DOI_PREIFX + ((Node) o).getText().trim());
 		}
+		instance.getUrl().addAll(url);
 		return Arrays.asList(instance);
 	}
 

dhp-workflows/dhp-graph-provision-scholexplorer/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-provision/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/dhp-graph-provision/src/main/resources/eu/dnetlib/dhp/oa/provision/oozie_app/workflow.xml

@@ -405,6 +405,9 @@
                 --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
                 --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
                 --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.speculation=false
+                --conf spark.hadoop.mapreduce.map.speculative=false
+                --conf spark.hadoop.mapreduce.reduce.speculative=false
             </spark-opts>
             <arg>--inputPath</arg><arg>${workingDir}/xml</arg>
             <arg>--isLookupUrl</arg> <arg>${isLookupUrl}</arg>

dhp-workflows/dhp-stats-update/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>dhp-stats-update</artifactId>

dhp-workflows/dhp-stats-update/src/main/resources/eu/dnetlib/dhp/oa/graph/stats/oozie_app/scripts/step9_6.sql

@@ -1,2 +1,2 @@
 DROP TABLE IF EXISTS ${stats_db_name}.datasource_languages;
-CREATE TABLE ${stats_db_name}.datasource_languages AS SELECT substr(d.id, 4) as id, langs.languages as language from openaire.datasource d LATERAL VIEW explode(d.odlanguages.value) langs as languages;
+CREATE TABLE ${stats_db_name}.datasource_languages AS SELECT substr(d.id, 4) as id, langs.languages as language from ${openaire_db_name}.datasource d LATERAL VIEW explode(d.odlanguages.value) langs as languages;

dhp-workflows/dhp-worfklow-profiles/pom.xml

@@ -1,11 +1,9 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
+<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/xsd/maven-4.0.0.xsd">
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <artifactId>dhp-workflows</artifactId>
         <groupId>eu.dnetlib.dhp</groupId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

dhp-workflows/pom.xml

@@ -6,7 +6,7 @@
     <parent>
         <groupId>eu.dnetlib.dhp</groupId>
         <artifactId>dhp</artifactId>
-        <version>1.1.7-SNAPSHOT</version>
+        <version>1.1.8-SNAPSHOT</version>
         <relativePath>../</relativePath>
     </parent>
 

pom.xml

@@ -3,7 +3,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>eu.dnetlib.dhp</groupId>
 	<artifactId>dhp</artifactId>
-	<version>1.1.7-SNAPSHOT</version>
+	<version>1.1.8-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<licenses>