yarn
diff --git a/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java b/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
index 03e6674e4..ae5bf9252 100644
--- a/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
+++ b/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
@@ -12,8 +12,6 @@ import org.apache.spark.sql.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Joiner;
-
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.schema.common.ModelSupport;
import eu.dnetlib.dhp.schema.oaf.DataInfo;
diff --git a/dhp-workflows/dhp-graph-mapper/pom.xml b/dhp-workflows/dhp-graph-mapper/pom.xml
index 0439c2ba3..5ddcda3fa 100644
--- a/dhp-workflows/dhp-graph-mapper/pom.xml
+++ b/dhp-workflows/dhp-graph-mapper/pom.xml
@@ -9,6 +9,37 @@
dhp-graph-mapper
+
+
+
+ net.alchim31.maven
+ scala-maven-plugin
+ 4.0.1
+
+
+ scala-compile-first
+ initialize
+
+ add-source
+ compile
+
+
+
+ scala-test-compile
+ process-test-resources
+
+ testCompile
+
+
+
+
+ ${scala.version}
+
+
+
+
+
+
@@ -61,6 +92,13 @@
org.postgresql
postgresql
+
+ org.json4s
+ json4s-jackson_2.11
+ 3.5.3
+
+
+
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/merge/MergeGraphSparkJob.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/merge/MergeGraphSparkJob.java
new file mode 100644
index 000000000..b723de955
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/merge/MergeGraphSparkJob.java
@@ -0,0 +1,162 @@
+
+package eu.dnetlib.dhp.oa.graph.merge;
+
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.FilterFunction;
+import org.apache.spark.api.java.function.MapFunction;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.SaveMode;
+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.common.HdfsSupport;
+import eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob;
+import eu.dnetlib.dhp.schema.common.ModelSupport;
+import eu.dnetlib.dhp.schema.oaf.*;
+import scala.Tuple2;
+
+/**
+ * Combines the content from two aggregator graph tables of the same type, entities (or relationships) with the same ids
+ * are picked preferring those from the BETA aggregator rather then from PROD. The identity of a relationship is defined
+ * by eu.dnetlib.dhp.schema.common.ModelSupport#idFn()
+ */
+public class MergeGraphSparkJob {
+
+ private static final Logger log = LoggerFactory.getLogger(CleanGraphSparkJob.class);
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private static final String PRIORITY_DEFAULT = "BETA"; // BETA | PROD
+
+ public static void main(String[] args) throws Exception {
+
+ String jsonConfiguration = IOUtils
+ .toString(
+ CleanGraphSparkJob.class
+ .getResourceAsStream(
+ "/eu/dnetlib/dhp/oa/graph/merge_graphs_parameters.json"));
+ final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+ parser.parseArgument(args);
+
+ String priority = Optional
+ .ofNullable(parser.get("priority"))
+ .orElse(PRIORITY_DEFAULT);
+ log.info("priority: {}", priority);
+
+ Boolean isSparkSessionManaged = Optional
+ .ofNullable(parser.get("isSparkSessionManaged"))
+ .map(Boolean::valueOf)
+ .orElse(Boolean.TRUE);
+ log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+ String betaInputPath = parser.get("betaInputPath");
+ log.info("betaInputPath: {}", betaInputPath);
+
+ String prodInputPath = parser.get("prodInputPath");
+ log.info("prodInputPath: {}", prodInputPath);
+
+ String outputPath = parser.get("outputPath");
+ log.info("outputPath: {}", outputPath);
+
+ String graphTableClassName = parser.get("graphTableClassName");
+ log.info("graphTableClassName: {}", graphTableClassName);
+
+ Class entityClazz = (Class) Class.forName(graphTableClassName);
+
+ SparkConf conf = new SparkConf();
+ conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
+ conf.registerKryoClasses(ModelSupport.getOafModelClasses());
+
+ runWithSparkSession(
+ conf,
+ isSparkSessionManaged,
+ spark -> {
+ removeOutputDir(spark, outputPath);
+ mergeGraphTable(spark, priority, betaInputPath, prodInputPath, entityClazz, entityClazz, outputPath);
+ });
+ }
+
+ private static void mergeGraphTable(
+ SparkSession spark,
+ String priority,
+ String betaInputPath,
+ String prodInputPath,
+ Class
p_clazz,
+ Class b_clazz,
+ String outputPath) {
+
+ Dataset> beta = readTableFromPath(spark, betaInputPath, b_clazz);
+ Dataset> prod = readTableFromPath(spark, prodInputPath, p_clazz);
+
+ prod
+ .joinWith(beta, prod.col("_1").equalTo(beta.col("_1")), "full_outer")
+ .map((MapFunction, Tuple2>, P>) value -> {
+ Optional p = Optional.ofNullable(value._1()).map(Tuple2::_2);
+ Optional b = Optional.ofNullable(value._2()).map(Tuple2::_2);
+ switch (priority) {
+ default:
+ case "BETA":
+ return mergeWithPriorityToBETA(p, b);
+ case "PROD":
+ return mergeWithPriorityToPROD(p, b);
+ }
+ }, Encoders.bean(p_clazz))
+ .filter((FilterFunction
) Objects::nonNull)
+ .write()
+ .mode(SaveMode.Overwrite)
+ .option("compression", "gzip")
+ .json(outputPath);
+ }
+
+ private static
P mergeWithPriorityToPROD(Optional
p, Optional b) {
+ if (b.isPresent() & !p.isPresent()) {
+ return (P) b.get();
+ }
+ if (p.isPresent()) {
+ return p.get();
+ }
+ return null;
+ }
+
+ private static
P mergeWithPriorityToBETA(Optional
p, Optional b) {
+ if (p.isPresent() & !b.isPresent()) {
+ return p.get();
+ }
+ if (b.isPresent()) {
+ return (P) b.get();
+ }
+ return null;
+ }
+
+ private static Dataset> readTableFromPath(
+ SparkSession spark, String inputEntityPath, Class clazz) {
+
+ log.info("Reading Graph table from: {}", inputEntityPath);
+ return spark
+ .read()
+ .textFile(inputEntityPath)
+ .map(
+ (MapFunction>) value -> {
+ final T t = OBJECT_MAPPER.readValue(value, clazz);
+ final String id = ModelSupport.idFn().apply(t);
+ return new Tuple2<>(id, t);
+ },
+ Encoders.tuple(Encoders.STRING(), Encoders.kryo(clazz)));
+ }
+
+ private static void removeOutputDir(SparkSession spark, String path) {
+ HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
+ }
+
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/AbstractMdRecordToOafMapper.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/AbstractMdRecordToOafMapper.java
index c43ee29fe..94da4c152 100644
--- a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/AbstractMdRecordToOafMapper.java
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/AbstractMdRecordToOafMapper.java
@@ -1,36 +1,10 @@
package eu.dnetlib.dhp.oa.graph.raw;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.createOpenaireId;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.dataInfo;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.field;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.journal;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.keyValue;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.listFields;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.oaiIProvenance;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.qualifier;
-import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.structuredProperty;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DATASET_DEFAULT_RESULTTYPE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_ACCESS_MODES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_PID_TYPES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.IS_PRODUCED_BY;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.NOT_AVAILABLE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.ORP_DEFAULT_RESULTTYPE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.OUTCOME;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.PRODUCES;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.PUBLICATION_DEFAULT_RESULTTYPE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.REPOSITORY_PROVENANCE_ACTIONS;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.RESULT_PROJECT;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.SOFTWARE_DEFAULT_RESULTTYPE;
-import static eu.dnetlib.dhp.schema.common.ModelConstants.UNKNOWN;
+import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.*;
+import static eu.dnetlib.dhp.schema.common.ModelConstants.*;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
import org.apache.commons.lang3.StringUtils;
import org.dom4j.Document;
@@ -40,24 +14,8 @@ import org.dom4j.Node;
import eu.dnetlib.dhp.oa.graph.raw.common.VocabularyGroup;
import eu.dnetlib.dhp.schema.common.LicenseComparator;
-import eu.dnetlib.dhp.schema.oaf.Author;
-import eu.dnetlib.dhp.schema.oaf.Context;
-import eu.dnetlib.dhp.schema.oaf.DataInfo;
-import eu.dnetlib.dhp.schema.oaf.Dataset;
-import eu.dnetlib.dhp.schema.oaf.Field;
-import eu.dnetlib.dhp.schema.oaf.GeoLocation;
-import eu.dnetlib.dhp.schema.oaf.Instance;
-import eu.dnetlib.dhp.schema.oaf.Journal;
-import eu.dnetlib.dhp.schema.oaf.KeyValue;
-import eu.dnetlib.dhp.schema.oaf.OAIProvenance;
-import eu.dnetlib.dhp.schema.oaf.Oaf;
-import eu.dnetlib.dhp.schema.oaf.OtherResearchProduct;
-import eu.dnetlib.dhp.schema.oaf.Publication;
-import eu.dnetlib.dhp.schema.oaf.Qualifier;
-import eu.dnetlib.dhp.schema.oaf.Relation;
-import eu.dnetlib.dhp.schema.oaf.Result;
-import eu.dnetlib.dhp.schema.oaf.Software;
-import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+import eu.dnetlib.dhp.schema.common.ModelConstants;
+import eu.dnetlib.dhp.schema.oaf.*;
public abstract class AbstractMdRecordToOafMapper {
@@ -99,7 +57,6 @@ public abstract class AbstractMdRecordToOafMapper {
final Document doc = DocumentHelper
.parseText(xml.replaceAll(DATACITE_SCHEMA_KERNEL_4, DATACITE_SCHEMA_KERNEL_3));
- final String type = doc.valueOf("//dr:CobjCategory/@type");
final KeyValue collectedFrom = getProvenanceDatasource(
doc, "//oaf:collectedFrom/@id", "//oaf:collectedFrom/@name");
@@ -118,12 +75,32 @@ public abstract class AbstractMdRecordToOafMapper {
final DataInfo info = prepareDataInfo(doc, invisible);
final long lastUpdateTimestamp = new Date().getTime();
- return createOafs(doc, type, collectedFrom, hostedBy, info, lastUpdateTimestamp);
+ final List instances = prepareInstances(doc, info, collectedFrom, hostedBy);
+
+ final String type = getResultType(doc, instances);
+
+ return createOafs(doc, type, instances, collectedFrom, info, lastUpdateTimestamp);
} catch (final Exception e) {
throw new RuntimeException(e);
}
}
+ protected String getResultType(final Document doc, final List instances) {
+ String type = doc.valueOf("//dr:CobjCategory/@type");
+
+ if (StringUtils.isBlank(type) & vocs.vocabularyExists(ModelConstants.DNET_RESULT_TYPOLOGIES)) {
+ String instanceType = instances
+ .stream()
+ .map(i -> i.getInstancetype().getClassid())
+ .findFirst()
+ .orElse("0000"); // Unknown
+ Qualifier resultType = vocs.getSynonymAsQualifier(ModelConstants.DNET_RESULT_TYPOLOGIES, instanceType);
+ return resultType.getClassid();
+ }
+
+ return type;
+ }
+
private KeyValue getProvenanceDatasource(final Document doc, final String xpathId, final String xpathName) {
final String dsId = doc.valueOf(xpathId);
final String dsName = doc.valueOf(xpathName);
@@ -138,8 +115,8 @@ public abstract class AbstractMdRecordToOafMapper {
protected List createOafs(
final Document doc,
final String type,
+ final List instances,
final KeyValue collectedFrom,
- final KeyValue hostedBy,
final DataInfo info,
final long lastUpdateTimestamp) {
@@ -148,14 +125,14 @@ public abstract class AbstractMdRecordToOafMapper {
switch (type.toLowerCase()) {
case "publication":
final Publication p = new Publication();
- populateResultFields(p, doc, collectedFrom, hostedBy, info, lastUpdateTimestamp);
+ populateResultFields(p, doc, instances, collectedFrom, info, lastUpdateTimestamp);
p.setResulttype(PUBLICATION_DEFAULT_RESULTTYPE);
p.setJournal(prepareJournal(doc, info));
oafs.add(p);
break;
case "dataset":
final Dataset d = new Dataset();
- populateResultFields(d, doc, collectedFrom, hostedBy, info, lastUpdateTimestamp);
+ populateResultFields(d, doc, instances, collectedFrom, info, lastUpdateTimestamp);
d.setResulttype(DATASET_DEFAULT_RESULTTYPE);
d.setStoragedate(prepareDatasetStorageDate(doc, info));
d.setDevice(prepareDatasetDevice(doc, info));
@@ -168,7 +145,7 @@ public abstract class AbstractMdRecordToOafMapper {
break;
case "software":
final Software s = new Software();
- populateResultFields(s, doc, collectedFrom, hostedBy, info, lastUpdateTimestamp);
+ populateResultFields(s, doc, instances, collectedFrom, info, lastUpdateTimestamp);
s.setResulttype(SOFTWARE_DEFAULT_RESULTTYPE);
s.setDocumentationUrl(prepareSoftwareDocumentationUrls(doc, info));
s.setLicense(prepareSoftwareLicenses(doc, info));
@@ -180,7 +157,7 @@ public abstract class AbstractMdRecordToOafMapper {
case "otherresearchproducts":
default:
final OtherResearchProduct o = new OtherResearchProduct();
- populateResultFields(o, doc, collectedFrom, hostedBy, info, lastUpdateTimestamp);
+ populateResultFields(o, doc, instances, collectedFrom, info, lastUpdateTimestamp);
o.setResulttype(ORP_DEFAULT_RESULTTYPE);
o.setContactperson(prepareOtherResearchProductContactPersons(doc, info));
o.setContactgroup(prepareOtherResearchProductContactGroups(doc, info));
@@ -259,14 +236,16 @@ public abstract class AbstractMdRecordToOafMapper {
private void populateResultFields(
final Result r,
final Document doc,
+ final List instances,
final KeyValue collectedFrom,
- final KeyValue hostedBy,
final DataInfo info,
final long lastUpdateTimestamp) {
r.setDataInfo(info);
r.setLastupdatetimestamp(lastUpdateTimestamp);
r.setId(createOpenaireId(50, doc.valueOf("//dri:objIdentifier"), false));
- r.setOriginalId(Arrays.asList(doc.valueOf("//dri:objIdentifier")));
+
+ r.setOriginalId(Arrays.asList(findOriginalId(doc)));
+
r.setCollectedfrom(Arrays.asList(collectedFrom));
r.setPid(prepareResultPids(doc, info));
r.setDateofcollection(doc.valueOf("//dr:dateOfCollection"));
@@ -291,7 +270,7 @@ public abstract class AbstractMdRecordToOafMapper {
r.setCoverage(prepareCoverages(doc, info));
r.setContext(prepareContexts(doc, info));
r.setExternalReference(new ArrayList<>()); // NOT PRESENT IN MDSTORES
- final List instances = prepareInstances(doc, info, collectedFrom, hostedBy);
+
r.setInstance(instances);
r.setBestaccessright(getBestAccessRights(instances));
}
@@ -429,6 +408,18 @@ public abstract class AbstractMdRecordToOafMapper {
return null;
}
+ private String findOriginalId(final Document doc) {
+ final Node n = doc.selectSingleNode("//*[local-name()='provenance']/*[local-name()='originDescription']");
+ if (n != null) {
+ final String id = n.valueOf("./*[local-name()='identifier']");
+ if (StringUtils.isNotBlank(id)) {
+ return id;
+ }
+ }
+ return doc.valueOf("//*[local-name()='header']/*[local-name()='identifier']");
+
+ }
+
protected Qualifier prepareQualifier(final Node node, final String xpath, final String schemeId) {
return prepareQualifier(node.valueOf(xpath).trim(), schemeId);
}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/OafMapperUtils.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/OafMapperUtils.java
index 8ede40773..63db13b8f 100644
--- a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/OafMapperUtils.java
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/common/OafMapperUtils.java
@@ -4,7 +4,11 @@ package eu.dnetlib.dhp.oa.graph.raw.common;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
@@ -57,6 +61,7 @@ public class OafMapperUtils {
.stream(values)
.map(v -> field(v, info))
.filter(Objects::nonNull)
+ .filter(distinctByKey(f -> f.getValue()))
.collect(Collectors.toList());
}
@@ -65,6 +70,7 @@ public class OafMapperUtils {
.stream()
.map(v -> field(v, info))
.filter(Objects::nonNull)
+ .filter(distinctByKey(f -> f.getValue()))
.collect(Collectors.toList());
}
@@ -237,4 +243,10 @@ public class OafMapperUtils {
public static String asString(final Object o) {
return o == null ? "" : o.toString();
}
+
+ public static Predicate distinctByKey(
+ final Function keyExtractor) {
+ final Map seen = new ConcurrentHashMap<>();
+ return t -> seen.putIfAbsent(keyExtractor.apply(t), Boolean.TRUE) == null;
+ }
}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/EBIAggregator.scala b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/EBIAggregator.scala
new file mode 100644
index 000000000..41fcd2636
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/EBIAggregator.scala
@@ -0,0 +1,89 @@
+package eu.dnetlib.dhp.sx.ebi
+import eu.dnetlib.dhp.schema.oaf.{Publication, Relation, Dataset => OafDataset}
+import org.apache.spark.sql.{Encoder, Encoders}
+import org.apache.spark.sql.expressions.Aggregator
+
+
+
+object EBIAggregator {
+
+ def getDatasetAggregator(): Aggregator[(String, OafDataset), OafDataset, OafDataset] = new Aggregator[(String, OafDataset), OafDataset, OafDataset]{
+
+ override def zero: OafDataset = new OafDataset()
+
+ override def reduce(b: OafDataset, a: (String, OafDataset)): OafDataset = {
+ b.mergeFrom(a._2)
+ if (b.getId == null)
+ b.setId(a._2.getId)
+ b
+ }
+
+
+ override def merge(wx: OafDataset, wy: OafDataset): OafDataset = {
+ wx.mergeFrom(wy)
+ if(wx.getId == null && wy.getId.nonEmpty)
+ wx.setId(wy.getId)
+ wx
+ }
+ override def finish(reduction: OafDataset): OafDataset = reduction
+
+ override def bufferEncoder: Encoder[OafDataset] =
+ Encoders.kryo(classOf[OafDataset])
+
+ override def outputEncoder: Encoder[OafDataset] =
+ Encoders.kryo(classOf[OafDataset])
+ }
+
+
+ def getPublicationAggregator(): Aggregator[(String, Publication), Publication, Publication] = new Aggregator[(String, Publication), Publication, Publication]{
+
+ override def zero: Publication = new Publication()
+
+ override def reduce(b: Publication, a: (String, Publication)): Publication = {
+ b.mergeFrom(a._2)
+ if (b.getId == null)
+ b.setId(a._2.getId)
+ b
+ }
+
+
+ override def merge(wx: Publication, wy: Publication): Publication = {
+ wx.mergeFrom(wy)
+ if(wx.getId == null && wy.getId.nonEmpty)
+ wx.setId(wy.getId)
+ wx
+ }
+ override def finish(reduction: Publication): Publication = reduction
+
+ override def bufferEncoder: Encoder[Publication] =
+ Encoders.kryo(classOf[Publication])
+
+ override def outputEncoder: Encoder[Publication] =
+ Encoders.kryo(classOf[Publication])
+ }
+
+
+ def getRelationAggregator(): Aggregator[(String, Relation), Relation, Relation] = new Aggregator[(String, Relation), Relation, Relation]{
+
+ override def zero: Relation = new Relation()
+
+ override def reduce(b: Relation, a: (String, Relation)): Relation = {
+ a._2
+ }
+
+
+ override def merge(a: Relation, b: Relation): Relation = {
+ if(b!= null) b else a
+ }
+ override def finish(reduction: Relation): Relation = reduction
+
+ override def bufferEncoder: Encoder[Relation] =
+ Encoders.kryo(classOf[Relation])
+
+ override def outputEncoder: Encoder[Relation] =
+ Encoders.kryo(classOf[Relation])
+ }
+
+
+
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkAddLinkUpdates.scala b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkAddLinkUpdates.scala
new file mode 100644
index 000000000..897bbd540
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkAddLinkUpdates.scala
@@ -0,0 +1,138 @@
+package eu.dnetlib.dhp.sx.ebi
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import eu.dnetlib.dhp.schema.oaf.{Instance, KeyValue, Oaf}
+import eu.dnetlib.dhp.schema.scholexplorer.OafUtils.createQualifier
+import eu.dnetlib.dhp.schema.scholexplorer.{DLIDataset, DLIRelation, OafUtils, ProvenaceInfo}
+import eu.dnetlib.dhp.utils.DHPUtils
+import eu.dnetlib.scholexplorer.relation.RelationMapper
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.sql._
+import org.json4s
+import org.json4s.DefaultFormats
+import org.json4s.JsonAST.{JField, JObject, JString}
+import org.json4s.jackson.JsonMethods.parse
+
+import scala.collection.JavaConverters._
+
+object SparkAddLinkUpdates {
+
+ val relationMapper = RelationMapper.load
+
+
+case class EBILinks(relation:String, pubdate:String, tpid:String, tpidType:String, turl:String, title:String, publisher:String) {}
+
+
+ def generatePubmedDLICollectedFrom(): KeyValue = {
+ OafUtils.generateKeyValue("dli_________::europe_pmc__", "Europe PMC")
+ }
+
+
+ def ebiLinksToOaf(input:(String, String)):List[Oaf] = {
+ val pmid :String = input._1
+ val input_json :String = input._2
+ implicit lazy val formats: DefaultFormats.type = org.json4s.DefaultFormats
+ lazy val json: json4s.JValue = parse(input_json)
+
+
+ val targets:List[EBILinks] = for {
+ JObject(link) <- json \\ "Category" \\ "Link"
+ JField("PublicationDate", JString(pubdate)) <- link
+ JField("RelationshipType", JObject(relationshipType)) <- link
+ JField("Name", JString(relname)) <- relationshipType
+ JField("Target", JObject(target)) <- link
+ JField("Identifier", JObject(identifier)) <- target
+ JField("ID", JString(tpid)) <- identifier
+ JField("IDScheme", JString(tpidtype)) <- identifier
+ JField("IDURL", JString(turl)) <- identifier
+ JField("Title", JString(title)) <- target
+ JField("Publisher", JObject(pub)) <- target
+ JField("Name", JString(publisher)) <- pub
+ } yield EBILinks(relname, pubdate, tpid, tpidtype, turl,title, publisher)
+
+
+
+ val dnetPublicationId = s"50|${DHPUtils.md5(s"$pmid::pmid")}"
+
+ targets.flatMap(l => {
+ val relation = new DLIRelation
+ val inverseRelation = new DLIRelation
+ val targetDnetId = s"50|${DHPUtils.md5(s"${l.tpid.toLowerCase.trim}::${l.tpidType.toLowerCase.trim}")}"
+ val relInfo = relationMapper.get(l.relation.toLowerCase)
+ val relationSemantic = relInfo.getOriginal
+ val inverseRelationSemantic = relInfo.getInverse
+
+ relation.setSource(dnetPublicationId)
+ relation.setTarget(targetDnetId)
+ relation.setRelClass("datacite")
+ relation.setRelType(relationSemantic)
+ relation.setCollectedfrom(List(generatePubmedDLICollectedFrom()).asJava)
+
+ inverseRelation.setSource(targetDnetId)
+ inverseRelation.setTarget(dnetPublicationId)
+ inverseRelation.setRelClass("datacite")
+ inverseRelation.setRelType(inverseRelationSemantic)
+ inverseRelation.setCollectedfrom(List(generatePubmedDLICollectedFrom()).asJava)
+
+
+
+ val d = new DLIDataset
+ d.setId(targetDnetId)
+ d.setDataInfo(OafUtils.generateDataInfo())
+ d.setPid(List(OafUtils.createSP(l.tpid.toLowerCase.trim, l.tpidType.toLowerCase.trim, "dnet:pid_types")).asJava)
+ d.setCompletionStatus("complete")
+ val pi = new ProvenaceInfo
+ pi.setId("dli_________::europe_pmc__")
+ pi.setName( "Europe PMC")
+ pi.setCompletionStatus("complete")
+ pi.setCollectionMode("collected")
+ d.setDlicollectedfrom(List(pi).asJava)
+ d.setCollectedfrom(List(generatePubmedDLICollectedFrom()).asJava)
+ d.setPublisher(OafUtils.asField(l.publisher))
+ d.setTitle(List(OafUtils.createSP(l.title, "main title", "dnet:dataCite_title")).asJava)
+ d.setDateofacceptance(OafUtils.asField(l.pubdate))
+ val i = new Instance
+ i.setCollectedfrom(generatePubmedDLICollectedFrom())
+ i.setDateofacceptance(d.getDateofacceptance)
+ i.setUrl(List(l.turl).asJava)
+ i.setInstancetype(createQualifier("0021", "Dataset", "dnet:publication_resource", "dnet:publication_resource"))
+ d.setInstance(List(i).asJava)
+ List(relation, inverseRelation, d)
+ })
+ }
+
+
+ def main(args: Array[String]): Unit = {
+ val conf: SparkConf = new SparkConf()
+ val parser = new ArgumentApplicationParser(IOUtils.toString(SparkCreateEBIDataFrame.getClass.getResourceAsStream("/eu/dnetlib/dhp/sx/ebi/ebi_to_df_params.json")))
+ parser.parseArgument(args)
+ val spark: SparkSession =
+ SparkSession
+ .builder()
+ .config(conf)
+ .appName(SparkCreateEBIDataFrame.getClass.getSimpleName)
+ .master(parser.get("master")).getOrCreate()
+
+
+ val workingPath = parser.get("workingPath")
+ implicit val oafEncoder: Encoder[Oaf] = Encoders.kryo(classOf[Oaf])
+ implicit val relEncoder: Encoder[DLIRelation] = Encoders.kryo(classOf[DLIRelation])
+ implicit val datEncoder: Encoder[DLIDataset] = Encoders.kryo(classOf[DLIDataset])
+
+ val ds:Dataset[(String,String)] = spark.read.load(s"$workingPath/baseline_links_updates").as[(String,String)](Encoders.tuple(Encoders.STRING, Encoders.STRING))
+
+ ds.flatMap(l =>ebiLinksToOaf(l)).write.mode(SaveMode.Overwrite).save(s"$workingPath/baseline_links_updates_oaf")
+
+ ds.filter(s => s.isInstanceOf)
+
+
+
+ val oDataset:Dataset[Oaf] = spark.read.load(s"$workingPath/baseline_links_updates_oaf").as[Oaf]
+
+ oDataset.filter(p =>p.isInstanceOf[DLIRelation]).map(p => p.asInstanceOf[DLIRelation]).write.mode(SaveMode.Overwrite).save(s"$workingPath/baseline_links_updates_relation")
+ oDataset.filter(p =>p.isInstanceOf[DLIDataset]).map(p => p.asInstanceOf[DLIDataset]).write.mode(SaveMode.Overwrite).save(s"$workingPath/baseline_links_updates_dataset")
+
+
+
+ }
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkCreateBaselineDataFrame.scala b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkCreateBaselineDataFrame.scala
new file mode 100644
index 000000000..77e03c9b3
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkCreateBaselineDataFrame.scala
@@ -0,0 +1,49 @@
+package eu.dnetlib.dhp.sx.ebi
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
+import eu.dnetlib.dhp.sx.ebi.model.{PMArticle, PMAuthor, PMJournal, PMParser}
+
+
+import scala.io.Source
+import scala.xml.pull.XMLEventReader
+
+object SparkCreateBaselineDataFrame {
+
+
+ def main(args: Array[String]): Unit = {
+ val conf: SparkConf = new SparkConf()
+ val parser = new ArgumentApplicationParser(IOUtils.toString(SparkCreateEBIDataFrame.getClass.getResourceAsStream("/eu/dnetlib/dhp/sx/ebi/ebi_to_df_params.json")))
+ parser.parseArgument(args)
+ val spark: SparkSession =
+ SparkSession
+ .builder()
+ .config(conf)
+ .appName(SparkCreateEBIDataFrame.getClass.getSimpleName)
+ .master(parser.get("master")).getOrCreate()
+
+ val sc = spark.sparkContext
+
+ val workingPath = parser.get("workingPath")
+
+ implicit val PMEncoder: Encoder[PMArticle] = Encoders.kryo(classOf[PMArticle])
+ implicit val PMJEncoder: Encoder[PMJournal] = Encoders.kryo(classOf[PMJournal])
+ implicit val PMAEncoder: Encoder[PMAuthor] = Encoders.kryo(classOf[PMAuthor])
+ val k: RDD[(String, String)] = sc.wholeTextFiles(s"$workingPath/baseline",2000)
+
+ val ds:Dataset[PMArticle] = spark.createDataset(k.filter(i => i._1.endsWith(".gz")).flatMap(i =>{
+ val xml = new XMLEventReader(Source.fromBytes(i._2.getBytes()))
+ new PMParser(xml)
+
+ } ))
+
+ ds.write.mode(SaveMode.Overwrite).save(s"$workingPath/baseline_dataset")
+
+
+
+
+ }
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkCreateEBIDataFrame.scala b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkCreateEBIDataFrame.scala
new file mode 100644
index 000000000..60857f0fc
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/SparkCreateEBIDataFrame.scala
@@ -0,0 +1,87 @@
+package eu.dnetlib.dhp.sx.ebi
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import eu.dnetlib.dhp.schema.oaf.{Oaf, Publication, Relation, Dataset => OafDataset}
+import eu.dnetlib.dhp.sx.graph.parser.{DatasetScholexplorerParser, PublicationScholexplorerParser}
+import eu.dnetlib.scholexplorer.relation.RelationMapper
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
+import org.codehaus.jackson.map.{ObjectMapper, SerializationConfig}
+import org.slf4j.{Logger, LoggerFactory}
+import scala.collection.JavaConverters._
+
+object SparkCreateEBIDataFrame {
+
+
+ def main(args: Array[String]): Unit = {
+ val logger: Logger = LoggerFactory.getLogger(SparkCreateEBIDataFrame.getClass)
+ val conf: SparkConf = new SparkConf()
+ val parser = new ArgumentApplicationParser(IOUtils.toString(SparkCreateEBIDataFrame.getClass.getResourceAsStream("/eu/dnetlib/dhp/sx/ebi/ebi_to_df_params.json")))
+ parser.parseArgument(args)
+ val spark: SparkSession =
+ SparkSession
+ .builder()
+ .config(conf)
+ .appName(SparkCreateEBIDataFrame.getClass.getSimpleName)
+ .master(parser.get("master")).getOrCreate()
+
+ val sc = spark.sparkContext
+
+
+ val workingPath = parser.get("workingPath")
+ val relationMapper = RelationMapper.load
+
+ implicit val oafEncoder: Encoder[Oaf] = Encoders.kryo(classOf[Oaf])
+ implicit val datasetEncoder: Encoder[OafDataset] = Encoders.kryo(classOf[OafDataset])
+ implicit val pubEncoder: Encoder[Publication] = Encoders.kryo(classOf[Publication])
+ implicit val relEncoder: Encoder[Relation] = Encoders.kryo(classOf[Relation])
+
+ logger.info("Extract Publication and relation from publication_xml")
+ val oafPubsRDD:RDD[Oaf] = sc.textFile(s"$workingPath/publication_xml").map(s =>
+ {
+ new ObjectMapper().readValue(s, classOf[String])
+ }).flatMap(s => {
+ val d = new PublicationScholexplorerParser
+ d.parseObject(s, relationMapper).asScala.iterator})
+
+ val mapper = new ObjectMapper()
+ mapper.getSerializationConfig.enable(SerializationConfig.Feature.INDENT_OUTPUT)
+ spark.createDataset(oafPubsRDD).write.mode(SaveMode.Overwrite).save(s"$workingPath/oaf")
+
+ logger.info("Extract Publication and relation from dataset_xml")
+ val oafDatsRDD:RDD[Oaf] = sc.textFile(s"$workingPath/dataset_xml").map(s =>
+ {
+ new ObjectMapper().readValue(s, classOf[String])
+ }).flatMap(s => {
+ val d = new DatasetScholexplorerParser
+ d.parseObject(s, relationMapper).asScala.iterator})
+
+ spark.createDataset(oafDatsRDD).write.mode(SaveMode.Append).save(s"$workingPath/oaf")
+ val dataset: Dataset[OafDataset] = spark.read.load(s"$workingPath/oaf").as[Oaf].filter(o => o.isInstanceOf[OafDataset]).map(d => d.asInstanceOf[OafDataset])
+ val publication: Dataset[Publication] = spark.read.load(s"$workingPath/oaf").as[Oaf].filter(o => o.isInstanceOf[Publication]).map(d => d.asInstanceOf[Publication])
+ val relations: Dataset[Relation] = spark.read.load(s"$workingPath/oaf").as[Oaf].filter(o => o.isInstanceOf[Relation]).map(d => d.asInstanceOf[Relation])
+ publication.map(d => (d.getId, d))(Encoders.tuple(Encoders.STRING, pubEncoder))
+ .groupByKey(_._1)(Encoders.STRING)
+ .agg(EBIAggregator.getPublicationAggregator().toColumn)
+ .map(p => p._2)
+ .write.mode(SaveMode.Overwrite).save(s"$workingPath/publication")
+
+ dataset.map(d => (d.getId, d))(Encoders.tuple(Encoders.STRING, datasetEncoder))
+ .groupByKey(_._1)(Encoders.STRING)
+ .agg(EBIAggregator.getDatasetAggregator().toColumn)
+ .map(p => p._2)
+ .write.mode(SaveMode.Overwrite).save(s"$workingPath/dataset")
+
+ relations.map(d => (s"${d.getSource}::${d.getRelType}::${d.getTarget}", d))(Encoders.tuple(Encoders.STRING, relEncoder))
+ .groupByKey(_._1)(Encoders.STRING)
+ .agg(EBIAggregator.getRelationAggregator().toColumn)
+ .map(p => p._2)
+ .write.mode(SaveMode.Overwrite).save(s"$workingPath/relation")
+
+
+
+ relations.map(r => (r.getSource, r.getTarget))(Encoders.tuple(Encoders.STRING,Encoders.STRING))
+ }
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMArticle.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMArticle.java
new file mode 100644
index 000000000..75d4628e6
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMArticle.java
@@ -0,0 +1,64 @@
+
+package eu.dnetlib.dhp.sx.ebi.model;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+public class PMArticle implements Serializable {
+
+ private String pmid;
+ private String date;
+ private PMJournal journal;
+ private String title;
+ private String description;
+ private List authors = new ArrayList<>();
+
+ public String getPmid() {
+ return pmid;
+ }
+
+ public void setPmid(String pmid) {
+ this.pmid = pmid;
+ }
+
+ public String getDate() {
+ return date;
+ }
+
+ public void setDate(String date) {
+ this.date = date;
+ }
+
+ public PMJournal getJournal() {
+ return journal;
+ }
+
+ public void setJournal(PMJournal journal) {
+ this.journal = journal;
+ }
+
+ public String getTitle() {
+ return title;
+ }
+
+ public void setTitle(String title) {
+ this.title = title;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public void setDescription(String description) {
+ this.description = description;
+ }
+
+ public List getAuthors() {
+ return authors;
+ }
+
+ public void setAuthors(List authors) {
+ this.authors = authors;
+ }
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMAuthor.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMAuthor.java
new file mode 100644
index 000000000..4a2198542
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMAuthor.java
@@ -0,0 +1,31 @@
+
+package eu.dnetlib.dhp.sx.ebi.model;
+
+import java.io.Serializable;
+
+public class PMAuthor implements Serializable {
+
+ private String lastName;
+ private String foreName;
+
+ public String getLastName() {
+ return lastName;
+ }
+
+ public void setLastName(String lastName) {
+ this.lastName = lastName;
+ }
+
+ public String getForeName() {
+ return foreName;
+ }
+
+ public void setForeName(String foreName) {
+ this.foreName = foreName;
+ }
+
+ public String getFullName() {
+ return String.format("%s, %s", this.foreName, this.lastName);
+ }
+
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMJournal.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMJournal.java
new file mode 100644
index 000000000..d4ff5a158
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMJournal.java
@@ -0,0 +1,53 @@
+
+package eu.dnetlib.dhp.sx.ebi.model;
+
+import java.io.Serializable;
+
+public class PMJournal implements Serializable {
+
+ private String issn;
+ private String volume;
+ private String issue;
+ private String date;
+ private String title;
+
+ public String getIssn() {
+ return issn;
+ }
+
+ public void setIssn(String issn) {
+ this.issn = issn;
+ }
+
+ public String getVolume() {
+ return volume;
+ }
+
+ public void setVolume(String volume) {
+ this.volume = volume;
+ }
+
+ public String getIssue() {
+ return issue;
+ }
+
+ public void setIssue(String issue) {
+ this.issue = issue;
+ }
+
+ public String getDate() {
+ return date;
+ }
+
+ public void setDate(String date) {
+ this.date = date;
+ }
+
+ public String getTitle() {
+ return title;
+ }
+
+ public void setTitle(String title) {
+ this.title = title;
+ }
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMParser.scala b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMParser.scala
new file mode 100644
index 000000000..903eba134
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/ebi/model/PMParser.scala
@@ -0,0 +1,92 @@
+package eu.dnetlib.dhp.sx.ebi.model
+import scala.xml.pull.{EvElemEnd, EvElemStart, EvText, XMLEventReader}
+class PMParser(xml:XMLEventReader) extends Iterator[PMArticle] {
+
+ var currentArticle:PMArticle = generateNextArticle()
+
+ override def hasNext: Boolean = currentArticle!= null
+
+ override def next(): PMArticle = {
+ val tmp = currentArticle
+ currentArticle = generateNextArticle()
+ tmp
+ }
+
+
+ def generateNextArticle():PMArticle = {
+
+ var currentAuthor: PMAuthor = null
+ var currentJournal: PMJournal = null
+ var currNode: String = null
+ var currentYear = "0"
+ var currentMonth = "01"
+ var currentDay = "01"
+
+ while (xml.hasNext) {
+ xml.next match {
+ case EvElemStart(_, label, _, _) =>
+ currNode = label
+ label match {
+ case "PubmedArticle" => currentArticle = new PMArticle
+ case "Author" => currentAuthor = new PMAuthor
+ case "Journal" => currentJournal = new PMJournal
+ case _ =>
+ }
+ case EvElemEnd(_, label) =>
+ label match {
+ case "PubmedArticle" => return currentArticle
+ case "Author" => currentArticle.getAuthors.add(currentAuthor)
+ case "Journal" => currentArticle.setJournal(currentJournal)
+ case "DateCompleted" => currentArticle.setDate(s"$currentYear-$currentMonth-$currentDay")
+ case "PubDate" => currentJournal.setDate(s"$currentYear-$currentMonth-$currentDay")
+ case _ =>
+ }
+ case EvText(text) =>
+ if (currNode!= null && text.trim.nonEmpty)
+ currNode match {
+ case "ArticleTitle" => {
+ if (currentArticle.getTitle==null)
+ currentArticle.setTitle(text.trim)
+ else
+ currentArticle.setTitle(currentArticle.getTitle + text.trim)
+ }
+ case "AbstractText" => {
+ if (currentArticle.getDescription==null)
+ currentArticle.setDescription(text.trim)
+ else
+ currentArticle.setDescription(currentArticle.getDescription + text.trim)
+ }
+ case "PMID" => currentArticle.setPmid(text.trim)
+ case "ISSN" => currentJournal.setIssn(text.trim)
+ case "Year" => currentYear = text.trim
+ case "Month" => currentMonth = text.trim
+ case "Day" => currentDay = text.trim
+ case "Volume" => currentJournal.setVolume( text.trim)
+ case "Issue" => currentJournal.setIssue (text.trim)
+ case "LastName" => {
+ if (currentAuthor != null)
+ currentAuthor.setLastName(text.trim)
+
+ }
+ case "ForeName" => if (currentAuthor != null)
+ currentAuthor.setForeName(text.trim)
+ case "Title" =>
+ if (currentJournal.getTitle==null)
+ currentJournal.setTitle(text.trim)
+ else
+ currentJournal.setTitle(currentJournal.getTitle + text.trim)
+ case _ =>
+
+ }
+ case _ =>
+ }
+
+ }
+ null
+ }
+}
+
+
+
+
+
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/AbstractScholexplorerParser.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/AbstractScholexplorerParser.java
index 0db2b2688..75f28c129 100644
--- a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/AbstractScholexplorerParser.java
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/AbstractScholexplorerParser.java
@@ -150,6 +150,17 @@ public abstract class AbstractScholexplorerParser {
return uk;
}
+ protected Qualifier generateQualifier(final String classId, final String className, final String schemeId,
+ final String schemeName) {
+ final Qualifier q = new Qualifier();
+ q.setClassid(classId);
+ q.setClassid(className);
+ q.setSchemeid(schemeId);
+ q.setSchemename(schemeName);
+ return q;
+
+ }
+
protected void generateRelations(
RelationMapper relationMapper,
Result parsedObject,
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/DatasetScholexplorerParser.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/DatasetScholexplorerParser.java
index afba57bb8..60371fa53 100644
--- a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/DatasetScholexplorerParser.java
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/DatasetScholexplorerParser.java
@@ -64,7 +64,6 @@ public class DatasetScholexplorerParser extends AbstractScholexplorerParser {
currentDate.setQualifier(dateQualifier);
parsedObject.setRelevantdate(Collections.singletonList(currentDate));
}
-
final String completionStatus = VtdUtilityParser
.getSingleValue(ap, vn, "//*[local-name()='completionStatus']");
final String provisionMode = VtdUtilityParser.getSingleValue(ap, vn, "//*[local-name()='provisionMode']");
@@ -149,6 +148,37 @@ public class DatasetScholexplorerParser extends AbstractScholexplorerParser {
inferPid(currentPid);
parsedObject.setPid(Collections.singletonList(currentPid));
+ String resolvedURL = null;
+
+ switch (currentPid.getQualifier().getClassname().toLowerCase()) {
+ case "uniprot":
+ resolvedURL = "https://www.uniprot.org/uniprot/" + currentPid.getValue();
+ break;
+ case "ena":
+ if (StringUtils.isNotBlank(currentPid.getValue()) && currentPid.getValue().length() > 7)
+ resolvedURL = "https://www.ebi.ac.uk/ena/data/view/" + currentPid.getValue().substring(0, 8);
+ break;
+ case "chembl":
+ resolvedURL = "https://www.ebi.ac.uk/chembl/compound_report_card/" + currentPid.getValue();
+ break;
+
+ case "ncbi-n":
+ resolvedURL = "https://www.ncbi.nlm.nih.gov/nuccore/" + currentPid.getValue();
+ break;
+ case "ncbi-p":
+ resolvedURL = "https://www.ncbi.nlm.nih.gov/nuccore/" + currentPid.getValue();
+ break;
+ case "genbank":
+ resolvedURL = "https://www.ncbi.nlm.nih.gov/nuccore/" + currentPid.getValue();
+ break;
+ case "pdb":
+ resolvedURL = "https://www.ncbi.nlm.nih.gov/nuccore/" + currentPid.getValue();
+ break;
+ case "url":
+ resolvedURL = currentPid.getValue();
+ break;
+ }
+
final String sourceId = generateId(
currentPid.getValue(), currentPid.getQualifier().getClassid(), "dataset");
parsedObject.setId(sourceId);
@@ -251,6 +281,11 @@ public class DatasetScholexplorerParser extends AbstractScholexplorerParser {
t -> {
final StructuredProperty st = new StructuredProperty();
st.setValue(t);
+ st
+ .setQualifier(
+ generateQualifier(
+ "main title", "main title", "dnet:dataCite_title",
+ "dnet:dataCite_title"));
return st;
})
.collect(Collectors.toList()));
@@ -282,6 +317,13 @@ public class DatasetScholexplorerParser extends AbstractScholexplorerParser {
.collect(Collectors.toList()));
}
+ if (StringUtils.isNotBlank(resolvedURL)) {
+ Instance i = new Instance();
+ i.setCollectedfrom(parsedObject.getCollectedfrom().get(0));
+ i.setUrl(Collections.singletonList(resolvedURL));
+ parsedObject.setInstance(Collections.singletonList(i));
+ }
+
result.add(parsedObject);
return result;
} catch (Throwable e) {
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/PublicationScholexplorerParser.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/PublicationScholexplorerParser.java
index bf59a6f0e..8d76004dc 100644
--- a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/PublicationScholexplorerParser.java
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/sx/graph/parser/PublicationScholexplorerParser.java
@@ -202,6 +202,11 @@ public class PublicationScholexplorerParser extends AbstractScholexplorerParser
t -> {
final StructuredProperty st = new StructuredProperty();
st.setValue(t);
+ st
+ .setQualifier(
+ generateQualifier(
+ "main title", "main title", "dnet:dataCite_title",
+ "dnet:dataCite_title"));
return st;
})
.collect(Collectors.toList()));
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge/oozie_app/config-default.xml b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge/oozie_app/config-default.xml
new file mode 100644
index 000000000..2e0ed9aee
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge/oozie_app/config-default.xml
@@ -0,0 +1,18 @@
+
+
+ jobTracker
+ yarnRM
+
+
+ nameNode
+ hdfs://nameservice1
+
+
+ oozie.use.system.libpath
+ true
+
+
+ oozie.action.sharelib.for.spark
+ spark2
+
+
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge/oozie_app/workflow.xml b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge/oozie_app/workflow.xml
new file mode 100644
index 000000000..07a125fb6
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge/oozie_app/workflow.xml
@@ -0,0 +1,293 @@
+
+
+
+
+ betaInputGgraphPath
+ the beta graph root path
+
+
+ prodInputGgraphPath
+ the production graph root path
+
+
+ graphOutputPath
+ the output merged graph root path
+
+
+ priority
+ decides from which infrastructure the content must win in case of ID clash
+
+
+
+ sparkDriverMemory
+ memory for driver process
+
+
+ sparkExecutorMemory
+ memory for individual executor
+
+
+ sparkExecutorCores
+ number of cores used by single executor
+
+
+ oozieActionShareLibForSpark2
+ oozie action sharelib for spark 2.*
+
+
+ spark2ExtraListeners
+ com.cloudera.spark.lineage.NavigatorAppListener
+ spark 2.* extra listeners classname
+
+
+ spark2SqlQueryExecutionListeners
+ com.cloudera.spark.lineage.NavigatorQueryListener
+ spark 2.* sql query execution listeners classname
+
+
+ spark2YarnHistoryServerAddress
+ spark 2.* yarn history server address
+
+
+ spark2EventLogDir
+ spark 2.* event log dir location
+
+
+
+
+
+
+ Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge publications
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/publication
+ --prodInputPath${prodInputGgraphPath}/publication
+ --outputPath${graphOutputPath}/publication
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Publication
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge datasets
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/dataset
+ --prodInputPath${prodInputGgraphPath}/dataset
+ --outputPath${graphOutputPath}/dataset
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Dataset
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge otherresearchproducts
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/otherresearchproduct
+ --prodInputPath${prodInputGgraphPath}/otherresearchproduct
+ --outputPath${graphOutputPath}/otherresearchproduct
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.OtherResearchProduct
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge softwares
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/software
+ --prodInputPath${prodInputGgraphPath}/software
+ --outputPath${graphOutputPath}/software
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Software
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge datasources
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/datasource
+ --prodInputPath${prodInputGgraphPath}/datasource
+ --outputPath${graphOutputPath}/datasource
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Datasource
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge organizations
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/organization
+ --prodInputPath${prodInputGgraphPath}/organization
+ --outputPath${graphOutputPath}/organization
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Organization
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge projects
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/project
+ --prodInputPath${prodInputGgraphPath}/project
+ --outputPath${graphOutputPath}/project
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Project
+ --priority${priority}
+
+
+
+
+
+
+
+ yarn
+ cluster
+ Merge relations
+ eu.dnetlib.dhp.oa.graph.merge.MergeGraphSparkJob
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-cores=${sparkExecutorCores}
+ --executor-memory=${sparkExecutorMemory}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.extraListeners=${spark2ExtraListeners}
+ --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+ --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+ --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+ --conf spark.sql.shuffle.partitions=7680
+
+ --betaInputPath${betaInputGgraphPath}/relation
+ --prodInputPath${prodInputGgraphPath}/relation
+ --outputPath${graphOutputPath}/relation
+ --graphTableClassNameeu.dnetlib.dhp.schema.oaf.Relation
+ --priority${priority}
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge_graphs_parameters.json b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge_graphs_parameters.json
new file mode 100644
index 000000000..1a612807b
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/merge_graphs_parameters.json
@@ -0,0 +1,38 @@
+[
+ {
+ "paramName": "issm",
+ "paramLongName": "isSparkSessionManaged",
+ "paramDescription": "when true will stop SparkSession after job execution",
+ "paramRequired": false
+ },
+ {
+ "paramName": "bin",
+ "paramLongName": "betaInputPath",
+ "paramDescription": "the beta graph root path",
+ "paramRequired": true
+ },
+ {
+ "paramName": "pin",
+ "paramLongName": "prodInputPath",
+ "paramDescription": "the production graph root path",
+ "paramRequired": true
+ },
+ {
+ "paramName": "out",
+ "paramLongName": "outputPath",
+ "paramDescription": "the output merged graph root path",
+ "paramRequired": true
+ },
+ {
+ "paramName": "class",
+ "paramLongName": "graphTableClassName",
+ "paramDescription": "class name moelling the graph table",
+ "paramRequired": true
+ },
+ {
+ "paramName": "pr",
+ "paramLongName": "priority",
+ "paramDescription": "decides from which infrastructure the content must win in case of ID clash",
+ "paramRequired": false
+ }
+]
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/ebi_to_df_params.json b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/ebi_to_df_params.json
new file mode 100644
index 000000000..366f1426e
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/ebi_to_df_params.json
@@ -0,0 +1,4 @@
+[
+ {"paramName":"mt", "paramLongName":"master", "paramDescription": "should be local or yarn", "paramRequired": true},
+ {"paramName":"w", "paramLongName":"workingPath", "paramDescription": "the path of the sequencial file to read", "paramRequired": true}
+]
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/oozie_app/config-default.xml b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/oozie_app/config-default.xml
new file mode 100644
index 000000000..cac3cc2bb
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/oozie_app/config-default.xml
@@ -0,0 +1,68 @@
+
+
+
+
+
+
+
+
+ jobTracker
+ yarn
+
+
+ nameNode
+ hdfs://hadoop-rm1.garr-pa1.d4science.org:8020
+
+
+ hive_metastore_uris
+ thrift://hadoop-edge3.garr-pa1.d4science.org:9083
+
+
+ spark2YarnHistoryServerAddress
+ http://hadoop-rm2.garr-pa1.d4science.org:19888
+
+
+
+
+ oozie.launcher.mapreduce.user.classpath.first
+ true
+
+
+
+ oozie.use.system.libpath
+ true
+
+
+ oozie.action.sharelib.for.spark
+ spark2
+
+
+ spark2EventLogDir
+ /user/spark/spark2ApplicationHistory
+
+
+ spark2ExtraListeners
+ "com.cloudera.spark.lineage.NavigatorAppListener"
+
+
+ spark2SqlQueryExecutionListeners
+ "com.cloudera.spark.lineage.NavigatorQueryListener"
+
+
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/oozie_app/workflow.xml b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/oozie_app/workflow.xml
new file mode 100644
index 000000000..a5035c56c
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/sx/ebi/oozie_app/workflow.xml
@@ -0,0 +1,97 @@
+
+
+
+ workingPath
+ the Working Path
+
+
+ sparkDriverMemory
+ memory for driver process
+
+
+ sparkExecutorMemory
+ memory for individual executor
+
+
+ sparkExecutorCores
+ number of cores used by single executor
+
+
+
+
+
+
+
+ Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]
+
+
+
+
+
+ yarn-cluster
+ cluster
+ Create Baselnie DataSet
+
+ eu.dnetlib.dhp.sx.ebi.SparkCreateBaselineDataFrame
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-memory=${sparkExecutorMemory}
+ --executor-cores=1
+ --driver-memory=${sparkDriverMemory}
+ --executor-cores=${sparkExecutorCores}
+ ${sparkExtraOPT}
+
+ --workingPath${workingPath}
+ --masteryarn
+
+
+
+
+
+
+
+ yarn-cluster
+ cluster
+ Create Baselnie DataSet
+
+ eu.dnetlib.dhp.sx.ebi.SparkAddLinkUpdates
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-memory=${sparkExecutorMemory}
+ --executor-cores=1
+ --driver-memory=${sparkDriverMemory}
+ --executor-cores=${sparkExecutorCores}
+ ${sparkExtraOPT}
+
+ --workingPath${workingPath}
+ --masteryarn
+
+
+
+
+
+
+
+
+ yarn-cluster
+ cluster
+ Create EBI DataSet
+
+ eu.dnetlib.dhp.sx.ebi.SparkCreateEBIDataFrame
+ dhp-graph-mapper-${projectVersion}.jar
+
+ --executor-memory=${sparkExecutorMemory}
+ --executor-cores=${sparkExecutorCores}
+ --driver-memory=${sparkDriverMemory}
+ --conf spark.sql.shuffle.partitions=1000
+ ${sparkExtraOPT}
+
+ --workingPath${workingPath}
+ --masteryarn
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/raw/MappersTest.java b/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/raw/MappersTest.java
index b1f0ecf0d..b157f7ec6 100644
--- a/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/raw/MappersTest.java
+++ b/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/raw/MappersTest.java
@@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.when;
import java.io.IOException;
@@ -20,6 +21,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import eu.dnetlib.dhp.oa.graph.clean.CleaningFunctionTest;
import eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils;
import eu.dnetlib.dhp.oa.graph.raw.common.VocabularyGroup;
import eu.dnetlib.dhp.schema.common.ModelConstants;
@@ -31,24 +33,25 @@ import eu.dnetlib.dhp.schema.oaf.Publication;
import eu.dnetlib.dhp.schema.oaf.Relation;
import eu.dnetlib.dhp.schema.oaf.Software;
import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
+import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
@ExtendWith(MockitoExtension.class)
public class MappersTest {
+ @Mock
+ private ISLookUpService isLookUpService;
+
@Mock
private VocabularyGroup vocs;
@BeforeEach
public void setUp() throws Exception {
- when(vocs.getTermAsQualifier(anyString(), anyString()))
- .thenAnswer(
- invocation -> OafMapperUtils
- .qualifier(
- invocation.getArgument(1), invocation.getArgument(1), invocation.getArgument(0),
- invocation.getArgument(0)));
-
- when(vocs.termExists(anyString(), anyString())).thenReturn(true);
+ lenient().when(isLookUpService.quickSearchProfile(VocabularyGroup.VOCABULARIES_XQUERY)).thenReturn(vocs());
+ lenient()
+ .when(isLookUpService.quickSearchProfile(VocabularyGroup.VOCABULARY_SYNONYMS_XQUERY))
+ .thenReturn(synonyms());
+ vocs = VocabularyGroup.loadVocsFromIS(isLookUpService);
}
@Test
@@ -68,9 +71,14 @@ public class MappersTest {
final Relation r2 = (Relation) list.get(2);
assertValidId(p.getId());
+
+ assertTrue(p.getOriginalId().size() == 1);
+ assertEquals("10.3897/oneeco.2.e13718", p.getOriginalId().get(0));
+
assertValidId(p.getCollectedfrom().get(0).getKey());
assertTrue(StringUtils.isNotBlank(p.getTitle().get(0).getValue()));
assertFalse(p.getDataInfo().getInvisible());
+ assertTrue(p.getSource().size() == 1);
assertTrue(p.getAuthor().size() > 0);
final Optional author = p
@@ -79,6 +87,7 @@ public class MappersTest {
.filter(a -> a.getPid() != null && !a.getPid().isEmpty())
.findFirst();
assertTrue(author.isPresent());
+
final StructuredProperty pid = author
.get()
.getPid()
@@ -169,6 +178,8 @@ public class MappersTest {
final Relation r2 = (Relation) list.get(2);
assertValidId(d.getId());
+ assertTrue(d.getOriginalId().size() == 1);
+ assertEquals("oai:zenodo.org:3234526", d.getOriginalId().get(0));
assertValidId(d.getCollectedfrom().get(0).getKey());
assertTrue(StringUtils.isNotBlank(d.getTitle().get(0).getValue()));
assertTrue(d.getAuthor().size() > 0);
@@ -261,4 +272,15 @@ public class MappersTest {
assertEquals(':', id.charAt(15));
assertEquals(':', id.charAt(16));
}
+
+ private List vocs() throws IOException {
+ return IOUtils
+ .readLines(CleaningFunctionTest.class.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/clean/terms.txt"));
+ }
+
+ private List synonyms() throws IOException {
+ return IOUtils
+ .readLines(CleaningFunctionTest.class.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/clean/synonyms.txt"));
+ }
+
}
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/sx/ebi/TestEBI.scala b/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/sx/ebi/TestEBI.scala
new file mode 100644
index 000000000..fa390a21b
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/sx/ebi/TestEBI.scala
@@ -0,0 +1,20 @@
+package eu.dnetlib.dhp.sx.ebi
+
+import org.junit.jupiter.api.Test
+
+class TestEBI {
+
+
+
+ @Test
+ def testEBIData() = {
+ SparkAddLinkUpdates.main("-mt local[*] -w /home/sandro/Downloads".split(" "))
+
+
+
+
+
+
+ }
+
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/oaf_record.xml b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/oaf_record.xml
index ead22aa96..3b2658bcf 100644
--- a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/oaf_record.xml
+++ b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/oaf_record.xml
@@ -34,6 +34,8 @@
info:eu-repo/semantics/altIdentifier/eissn/2367-8194
info:eu-repo/grantAgreement/EC/FP7/226852
One Ecosystem 2: e13718
+ One Ecosystem 2: e13718
+ One Ecosystem 2: e13718
Ecosystem Services hotspots
Natura 2000
Quiet Protected Areas
@@ -47,7 +49,8 @@
regulating services
supporting services
Research Article
- 0001
+
+ 0001
2017-01-01
corda_______::226852
OPEN
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_dataset.xml b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_dataset.xml
index 5525a2753..6648d1804 100644
--- a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_dataset.xml
+++ b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_dataset.xml
@@ -82,7 +82,8 @@
All files are in MATLAB .mat format.
- 0021
+
+ 0021
2019-01-01
OPEN
und
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_software.xml b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_software.xml
index fd3fdd473..6a9170ce1 100644
--- a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_software.xml
+++ b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/raw/odf_software.xml
@@ -52,7 +52,8 @@
subjectScheme="EDAM Ontology" valueURI="http://edamontology.org/topic_3534">Protein binding sites
- 0029
+
+ 0029
2018-06-06
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/sx/ebi/rel1.json b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/sx/ebi/rel1.json
new file mode 100644
index 000000000..038b84a49
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/sx/ebi/rel1.json
@@ -0,0 +1,55 @@
+{
+ "Category": [
+ {
+ "Section": [
+ {
+ "Linklist": {
+ "Link": [
+ {
+ "LinkProvider": {
+ "Name": "Europe PMC"
+ },
+ "Target": {
+ "Publisher": {
+ "Name": "Altmetric"
+ },
+ "ImageURL": "https://api.altmetric.com/v1/donut/58578459_64.png",
+ "Identifier": {
+ "ID": "https://www.altmetric.com/details/58578459",
+ "IDScheme": "URL",
+ "IDURL": "https://www.altmetric.com/details/58578459"
+ },
+ "Type": {
+ "Name": "dataset"
+ },
+ "Title": "Optical clumped isotope thermometry of carbon dioxide"
+ },
+ "Source": {
+ "Identifier": {
+ "ID": "30886173",
+ "IDScheme": "PMID"
+ },
+ "Type": {
+ "Name": "literature"
+ }
+ },
+ "PublicationDate": "06-04-2019",
+ "RelationshipType": {
+ "Name": "IsReferencedBy"
+ },
+ "ObtainedBy": "ext_links"
+ }
+ ]
+ },
+ "ObtainedBy": "ext_links",
+ "SectionLinkCount": 1,
+ "Tags": [
+ "altmetrics"
+ ]
+ }
+ ],
+ "CategoryLinkCount": 1,
+ "Name": "Altmetric"
+ }
+ ]
+}
diff --git a/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/sx/ebi/rel_multiple.json b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/sx/ebi/rel_multiple.json
new file mode 100644
index 000000000..2ad55861e
--- /dev/null
+++ b/dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/sx/ebi/rel_multiple.json
@@ -0,0 +1,191 @@
+{
+ "version": "6.3",
+ "hitCount": 4,
+ "request": {
+ "id": "28818901",
+ "source": "MED"
+ },
+ "dataLinkList": {
+ "Category": [
+ {
+ "Name": "Nucleotide Sequences",
+ "CategoryLinkCount": 3,
+ "Section": [
+ {
+ "ObtainedBy": "tm_accession",
+ "Tags": [
+ "supporting_data"
+ ],
+ "SectionLinkCount": 1,
+ "Linklist": {
+ "Link": [
+ {
+ "ObtainedBy": "tm_accession",
+ "PublicationDate": "27-02-2020",
+ "LinkProvider": {
+ "Name": "Europe PMC"
+ },
+ "RelationshipType": {
+ "Name": "References"
+ },
+ "Source": {
+ "Type": {
+ "Name": "literature"
+ },
+ "Identifier": {
+ "ID": "28818901",
+ "IDScheme": "MED"
+ }
+ },
+ "Target": {
+ "Type": {
+ "Name": "dataset"
+ },
+ "Identifier": {
+ "ID": "AP008937",
+ "IDScheme": "ENA",
+ "IDURL": "http://identifiers.org/ena.embl/AP008937"
+ },
+ "Title": "AP008937",
+ "Publisher": {
+ "Name": "Europe PMC"
+ }
+ },
+ "Frequency": 1
+ }
+ ]
+ }
+ },
+ {
+ "ObtainedBy": "submission",
+ "Tags": [
+ "related_data"
+ ],
+ "SectionLinkCount": 2,
+ "CollectionURL": "http://www.ebi.ac.uk/ena/data/search?query=28818901",
+ "Linklist": {
+ "Link": [
+ {
+ "ObtainedBy": "submission",
+ "PublicationDate": "25-06-2018",
+ "LinkProvider": {
+ "Name": "Europe PMC"
+ },
+ "RelationshipType": {
+ "Name": "IsReferencedBy"
+ },
+ "Source": {
+ "Type": {
+ "Name": "literature"
+ },
+ "Identifier": {
+ "ID": "28818901",
+ "IDScheme": "PMID"
+ }
+ },
+ "Target": {
+ "Type": {
+ "Name": "dataset"
+ },
+ "Identifier": {
+ "ID": "NIWV01000000",
+ "IDScheme": "ENA",
+ "IDURL": "http://www.ebi.ac.uk/ena/data/view/NIWV01000000"
+ },
+ "Title": "Nucleotide sequences",
+ "Publisher": {
+ "Name": "ENA"
+ }
+ }
+ },
+ {
+ "ObtainedBy": "submission",
+ "PublicationDate": "25-06-2018",
+ "LinkProvider": {
+ "Name": "Europe PMC"
+ },
+ "RelationshipType": {
+ "Name": "IsReferencedBy"
+ },
+ "Source": {
+ "Type": {
+ "Name": "literature"
+ },
+ "Identifier": {
+ "ID": "28818901",
+ "IDScheme": "PMID"
+ }
+ },
+ "Target": {
+ "Type": {
+ "Name": "dataset"
+ },
+ "Identifier": {
+ "ID": "PRJNA390617",
+ "IDScheme": "ENA",
+ "IDURL": "http://www.ebi.ac.uk/ena/data/view/PRJNA390617"
+ },
+ "Title": "Lactobacillus fermentum strain:BFE 6620",
+ "Publisher": {
+ "Name": "ENA"
+ }
+ }
+ }
+ ]
+ }
+ }
+ ]
+ },
+ {
+ "Name": "BioStudies: supplemental material and supporting data",
+ "CategoryLinkCount": 1,
+ "Section": [
+ {
+ "ObtainedBy": "ext_links",
+ "Tags": [
+ "supporting_data"
+ ],
+ "SectionLinkCount": 1,
+ "Linklist": {
+ "Link": [
+ {
+ "ObtainedBy": "ext_links",
+ "PublicationDate": "24-07-2018",
+ "LinkProvider": {
+ "Name": "Europe PMC"
+ },
+ "RelationshipType": {
+ "Name": "IsReferencedBy"
+ },
+ "Source": {
+ "Type": {
+ "Name": "literature"
+ },
+ "Identifier": {
+ "ID": "28818901",
+ "IDScheme": "PMID"
+ }
+ },
+ "Target": {
+ "Type": {
+ "Name": "dataset"
+ },
+ "Identifier": {
+ "ID": "http://www.ebi.ac.uk/biostudies/studies/S-EPMC5604774?xr=true",
+ "IDScheme": "URL",
+ "IDURL": "http://www.ebi.ac.uk/biostudies/studies/S-EPMC5604774?xr=true"
+ },
+ "Title": "Draft Genome Sequence of Lactobacillus fermentum BFE 6620, a Potential Starter Culture for African Vegetable Foods, Isolated from Fermented Cassava.",
+ "Publisher": {
+ "Name": "BioStudies: supplemental material and supporting data"
+ }
+ }
+ }
+ ]
+ }
+ }
+ ]
+ }
+ ]
+ }
+}
\ No newline at end of file
diff --git a/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/DLIToOAF.scala b/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/DLIToOAF.scala
index 637362acf..86b68fbd2 100644
--- a/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/DLIToOAF.scala
+++ b/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/DLIToOAF.scala
@@ -5,11 +5,12 @@ import java.time.format.DateTimeFormatter
import eu.dnetlib.dhp.common.PacePerson
import eu.dnetlib.dhp.schema.action.AtomicAction
-import eu.dnetlib.dhp.schema.oaf.{Author, DataInfo, Dataset, ExternalReference, Field, Instance, KeyValue, Oaf, Publication, Qualifier, Relation, StructuredProperty}
+import eu.dnetlib.dhp.schema.oaf.{Author, Dataset, ExternalReference, Field, Instance, KeyValue, Oaf, Publication, Qualifier, Relation, Result, StructuredProperty}
import eu.dnetlib.dhp.schema.scholexplorer.{DLIDataset, DLIPublication, DLIRelation}
import eu.dnetlib.dhp.utils.DHPUtils
import org.apache.commons.lang3.StringUtils
import org.codehaus.jackson.map.ObjectMapper
+import eu.dnetlib.dhp.schema.scholexplorer.OafUtils._
import scala.collection.JavaConverters._
@@ -99,6 +100,20 @@ object DLIToOAF {
)
+ def fixInstance(r:Publication) :Publication = {
+ val collectedFrom = r.getCollectedfrom.asScala.head
+ r.getInstance().asScala.foreach(i => i.setCollectedfrom(collectedFrom))
+ r
+ }
+
+
+ def fixInstanceDataset(r:Dataset) :Dataset = {
+ val collectedFrom = r.getCollectedfrom.asScala.head
+ r.getInstance().asScala.foreach(i => i.setCollectedfrom(collectedFrom))
+ r
+ }
+
+
def toActionSet(item: Oaf): (String, String) = {
val mapper = new ObjectMapper()
@@ -412,46 +427,6 @@ object DLIToOAF {
}
- def generateKeyValue(key: String, value: String): KeyValue = {
- val kv: KeyValue = new KeyValue()
- kv.setKey(key)
- kv.setValue(value)
- kv.setDataInfo(generateDataInfo("0.9"))
- kv
- }
- def generateDataInfo(trust: String = "0.9", invisibile: Boolean = false): DataInfo = {
- val di = new DataInfo
- di.setDeletedbyinference(false)
- di.setInferred(false)
- di.setInvisible(false)
- di.setTrust(trust)
- di.setProvenanceaction(createQualifier("sysimport:actionset", "dnet:provenanceActions"))
- di
- }
-
- def createQualifier(cls: String, sch: String): Qualifier = {
- createQualifier(cls, cls, sch, sch)
- }
-
-
- def createQualifier(classId: String, className: String, schemeId: String, schemeName: String): Qualifier = {
- val q: Qualifier = new Qualifier
- q.setClassid(classId)
- q.setClassname(className)
- q.setSchemeid(schemeId)
- q.setSchemename(schemeName)
- q
- }
-
-
- def asField[T](value: T): Field[T] = {
- val tmp = new Field[T]
- tmp.setValue(value)
- tmp
-
-
- }
-
}
diff --git a/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/SparkExportContentForOpenAire.scala b/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/SparkExportContentForOpenAire.scala
index edf951df4..fd8f2d136 100644
--- a/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/SparkExportContentForOpenAire.scala
+++ b/dhp-workflows/dhp-graph-provision-scholexplorer/src/main/java/eu/dnetlib/dhp/export/SparkExportContentForOpenAire.scala
@@ -1,7 +1,7 @@
package eu.dnetlib.dhp.`export`
import eu.dnetlib.dhp.application.ArgumentApplicationParser
-import eu.dnetlib.dhp.schema.oaf.{Publication, Relation, Dataset => OafDataset}
+import eu.dnetlib.dhp.schema.oaf.{Instance, Publication, Relation, Dataset => OafDataset}
import eu.dnetlib.dhp.schema.scholexplorer.{DLIDataset, DLIPublication, DLIRelation}
import org.apache.commons.io.IOUtils
import org.apache.hadoop.io.Text
@@ -166,10 +166,13 @@ object SparkExportContentForOpenAire {
}).write.mode(SaveMode.Overwrite).save(s"$workingPath/relationAS")
- val fRels:Dataset[(String,String)] = spark.read.load(s"$workingPath/relationAS").as[Relation].map(DLIToOAF.toActionSet)
- val fpubs:Dataset[(String,String)] = spark.read.load(s"$workingPath/publicationAS").as[Publication].map(DLIToOAF.toActionSet)
- val fdats:Dataset[(String,String)] = spark.read.load(s"$workingPath/datasetAS").as[OafDataset].map(DLIToOAF.toActionSet)
+ spark.read.load(s"$workingPath/publicationAS").as[Publication].map(DLIToOAF.fixInstance).write.mode(SaveMode.Overwrite).save(s"$workingPath/publicationAS_fixed")
+ spark.read.load(s"$workingPath/datasetAS").as[OafDataset].map(DLIToOAF.fixInstanceDataset).write.mode(SaveMode.Overwrite).save(s"$workingPath/datasetAS_fixed")
+
+ val fRels:Dataset[(String,String)] = spark.read.load(s"$workingPath/relationAS").as[Relation].map(DLIToOAF.toActionSet)
+ val fpubs:Dataset[(String,String)] = spark.read.load(s"$workingPath/publicationAS_fixed").as[Publication].map(DLIToOAF.toActionSet)
+ val fdats:Dataset[(String,String)] = spark.read.load(s"$workingPath/datasetAS_fixed").as[OafDataset].map(DLIToOAF.toActionSet)
fRels.union(fpubs).union(fdats).rdd.map(s => (new Text(s._1), new Text(s._2))).saveAsHadoopFile(s"$workingPath/rawset", classOf[Text], classOf[Text], classOf[SequenceFileOutputFormat[Text,Text]], classOf[GzipCodec])
}
diff --git a/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_beta_construction.xml b/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_beta_construction.xml
new file mode 100644
index 000000000..08ed24cd0
--- /dev/null
+++ b/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_beta_construction.xml
@@ -0,0 +1,779 @@
+
+
+
+ Graph Construction [HYBRID]
+ Data Provision
+ 30
+
+
+
+ reuse cached content from the PROD aggregation system
+
+ reuseProdContent
+ true
+
+
+
+
+
+
+
+ set the PROD aggregator content path
+
+ prodContentPath
+ /tmp/core_aggregator
+
+
+
+
+
+
+
+ Set the path containing the PROD AGGREGATOR graph
+
+ prodAggregatorGraphPath
+ /tmp/core_provision/graph/00_prod_graph_aggregator
+
+
+
+
+
+
+
+ reuse cached content from the BETA aggregation system
+
+ reuseBetaContent
+ true
+
+
+
+
+
+
+
+ set the BETA aggregator content path
+
+ betaContentPath
+ /tmp/beta_aggregator
+
+
+
+
+
+
+
+ Set the path containing the BETA AGGREGATOR graph
+
+ betaAggregatorGraphPath
+ /tmp/core_provision/graph/00_beta_graph_aggregator
+
+
+
+
+
+
+
+ Set the IS lookup service address
+
+ isLookUpUrl
+ http://services.openaire.eu:8280/is/services/isLookUp?wsdl
+
+
+
+
+
+
+
+ Set the target path to store the MERGED graph
+
+ mergedGraphPath
+ /tmp/core_provision/graph/01_graph_merged
+
+
+
+
+
+
+
+ Set the target path to store the RAW graph
+
+ rawGraphPath
+ /tmp/core_provision/graph/02_graph_raw
+
+
+
+
+
+
+
+ Set the target path to store the DEDUPED graph
+
+ dedupGraphPath
+ /tmp/core_provision/graph/03_graph_dedup
+
+
+
+
+
+
+
+ Set the target path to store the INFERRED graph
+
+ inferredGraphPath
+ /tmp/core_provision/graph/04_graph_inferred
+
+
+
+
+
+
+
+ Set the target path to store the CONSISTENCY graph
+
+ consistentGraphPath
+ /tmp/core_provision/graph/05_graph_consistent
+
+
+
+
+
+
+
+ Set the target path to store the ORCID enriched graph
+
+ orcidGraphPath
+ /tmp/core_provision/graph/06_graph_orcid
+
+
+
+
+
+
+
+ Set the target path to store the BULK TAGGED graph
+
+ bulkTaggingGraphPath
+ /tmp/core_provision/graph/07_graph_bulktagging
+
+
+
+
+
+
+
+ Set the target path to store the AFFILIATION from INSTITUTIONAL REPOS graph
+
+ affiliationGraphPath
+ /tmp/core_provision/graph/08_graph_affiliation
+
+
+
+
+
+
+
+ Set the target path to store the COMMUNITY from SELECTED SOURCES graph
+
+ communityOrganizationGraphPath
+ /tmp/core_provision/graph/09_graph_comunity_organization
+
+
+
+
+
+
+
+ Set the target path to store the FUNDING from SEMANTIC RELATION graph
+
+ fundingGraphPath
+ /tmp/core_provision/graph/10_graph_funding
+
+
+
+
+
+
+
+ Set the target path to store the COMMUNITY from SEMANTIC RELATION graph
+
+ communitySemRelGraphPath
+ /tmp/core_provision/graph/11_graph_comunity_sem_rel
+
+
+
+
+
+
+
+ Set the target path to store the COUNTRY enriched graph
+
+ countryGraphPath
+ /tmp/core_provision/graph/12_graph_country
+
+
+
+
+
+
+
+ Set the target path to store the CLEANED graph
+
+ cleanedGraphPath
+ /tmp/core_provision/graph/13_graph_cleaned
+
+
+
+
+
+
+
+ Set the target path to store the blacklisted graph
+
+ blacklistedGraphPath
+ /tmp/core_provision/graph/14_graph_blacklisted
+
+
+
+
+
+
+
+ Set the map of paths for the Bulk Tagging
+
+ bulkTaggingPathMap
+ {"author" : "$['author'][*]['fullname']", "title" : "$['title'][*]['value']", "orcid" : "$['author'][*]['pid'][*][?(@['key']=='ORCID')]['value']", "contributor" : "$['contributor'][*]['value']", "description" : "$['description'][*]['value']"}
+
+
+
+
+
+
+
+ Set the map of associations organization, community list for the propagation of community to result through organization
+
+ propagationOrganizationCommunityMap
+ {"20|corda__h2020::3fb05a9524c3f790391261347852f638":["mes","euromarine"], "20|corda__h2020::e8dbe14cca9bf6fce09d468872f813f8":["mes","euromarine"], "20|snsf________::9b253f265e3bef5cae6d881fdf61aceb":["mes","euromarine"],"20|rcuk________::e054eea0a47665af8c3656b5785ccf76":["mes","euromarine"],"20|corda__h2020::edc18d67c9b11fb616ca9f6e1db1b151":["mes","euromarine"],"20|rcuk________::d5736d9da90521ddcdc7828a05a85e9a":["mes","euromarine"],"20|corda__h2020::f5d418d3aa1cf817ddefcc3fdc039f27":["mes","euromarine"],"20|snsf________::8fa091f8f25a846779acb4ea97b50aef":["mes","euromarine"],"20|corda__h2020::81e020977211c2c40fae2e1a50bffd71":["mes","euromarine"],"20|corda_______::81e020977211c2c40fae2e1a50bffd71":["mes","euromarine"],"20|snsf________::31d0a100e54e3cdb3c6f52d91e638c78":["mes","euromarine"],"20|corda__h2020::ea379ef91b8cc86f9ac5edc4169292db":["mes","euromarine"],"20|corda__h2020::f75ee2ee48e5cb0ec8c8d30aaa8fef70":["mes","euromarine"],"20|rcuk________::e16010089551a1a9182a94604fc0ea59":["mes","euromarine"],"20|corda__h2020::38531a2cce7c5c347ffc439b07c1f43b":["mes","euromarine"],"20|corda_______::38531a2cce7c5c347ffc439b07c1f43b":["mes","euromarine"],"20|grid________::b2cbbf5eadbbf87d534b022bad3191d7":["mes","euromarine"],"20|snsf________::74730ef1439d7f7636a8be58a6b471b8":["mes","euromarine"],"20|nsf_________::ad72e19043a5a467e35f9b444d11563e":["mes","euromarine"],"20|rcuk________::0fc3e92500290902a2d38ec2445e74c3":["mes","euromarine"],"20|grid________::ad2c29905da0eb3c06b3fa80cacd89ea":["mes","euromarine"],"20|corda__h2020::30b53e4d63d3724f00acb9cbaca40860":["mes","euromarine"],"20|corda__h2020::f60f84bee14ad93f0db0e49af1d5c317":["mes","euromarine"], "20|corda__h2020::7bf251ac3765b5e89d82270a1763d09f":["mes","euromarine"], "20|corda__h2020::65531bd11be9935948c7f2f4db1c1832":["mes","euromarine"], "20|corda__h2020::e0e98f86bbc76638bbb72a8fe2302946":["mes","euromarine"], "20|snsf________::3eb43582ac27601459a8d8b3e195724b":["mes","euromarine"], "20|corda__h2020::af2481dab65d06c8ea0ae02b5517b9b6":["mes","euromarine"], "20|corda__h2020::c19d05cfde69a50d3ebc89bd0ee49929":["mes","euromarine"], "20|corda__h2020::af0bfd9fc09f80d9488f56d71a9832f0":["mes","euromarine"], "20|rcuk________::f33c02afb0dc66c49d0ed97ca5dd5cb0":["beopen"],
+ "20|grid________::a867f78acdc5041b34acfe4f9a349157":["beopen"], "20|grid________::7bb116a1a9f95ab812bf9d2dea2be1ff":["beopen"], "20|corda__h2020::6ab0e0739dbe625b99a2ae45842164ad":["beopen"], "20|corda__h2020::8ba50792bc5f4d51d79fca47d860c602":["beopen"], "20|corda_______::8ba50792bc5f4d51d79fca47d860c602":["beopen"], "20|corda__h2020::e70e9114979e963eef24666657b807c3":["beopen"], "20|corda_______::e70e9114979e963eef24666657b807c3":["beopen"], "20|corda_______::15911e01e9744d57205825d77c218737":["beopen"], "20|opendoar____::056a41e24e2a9a67215e87bbee6a80ab":["beopen"], "20|opendoar____::7f67f2e6c6fbb0628f8160fcd3d92ae3":["beopen"], "20|grid________::a8ecfd7c084e561168bcbe6bf0daf3e3":["beopen"], "20|corda_______::7bbe6cc5d8ec1864739a04b0d020c9e9":["beopen"], "20|corda_______::3ff558e30c2e434d688539548300b050":["beopen"], "20|corda__h2020::5ffee5b3b83b33a8cf0e046877bd3a39":["beopen"], "20|corda__h2020::5187217e2e806a6df3579c46f82401bc":["beopen"], "20|grid________::5fa7e2709bcd945e26bfa18689adeec1":["beopen"], "20|corda_______::d8696683c53027438031a96ad27c3c07":["beopen"], "20|corda__h2020::d8696683c53027438031a96ad27c3c07":["beopen"], "20|rcuk________::23a79ebdfa59790864e4a485881568c1":["beopen"], "20|corda__h2020::b76cf8fe49590a966953c37e18608af9":["beopen"], "20|grid________::d2f0204126ee709244a488a4cd3b91c2":["beopen"], "20|corda__h2020::05aba9d2ed17533d15221e5655ac11e6":["beopen"], "20|grid________::802401579481dc32062bdee69f5e6a34":["beopen"], "20|corda__h2020::3f6d9d54cac975a517ba6b252c81582d":["beopen"]}
+
+
+
+
+
+
+
+
+ Set the dedup orchestrator name
+
+ dedupConfig
+ decisiontree-dedup-test
+
+
+
+
+
+
+
+ declares the ActionSet ids to promote in the RAW graph
+
+ actionSetIdsRawGraph
+ scholexplorer-dump,gridac-dump,doiboost-organizations,doiboost,orcidworks-no-doi,iis-wos-entities,iis-entities-software,iis-entities-patent
+
+
+
+
+
+
+
+ declares the ActionSet ids to promote in the INFERRED graph
+
+ actionSetIdsIISGraph
+ iis-researchinitiative,iis-document-citations,iis-document-affiliation,iis-document-classes,iis-document-similarities,iis-referenced-datasets-main,iis-referenced-datasets-preprocessing,iis-referenced-projects-main,iis-referenced-projects-preprocessing,iis-referenceextraction-pdb,document_software_url,iis-extracted-metadata,iis-communities,iis-referenced-patents,iis-covid-19
+
+
+
+
+
+
+
+ wait configurations
+
+
+
+
+
+
+
+
+
+ create the AGGREGATOR graph
+
+ executeOozieJob
+ IIS
+
+ {
+ 'graphOutputPath' : 'betaAggregatorGraphPath',
+ 'isLookupUrl' : 'isLookUpUrl',
+ 'reuseContent' : 'reuseBetaContent',
+ 'contentPath' : 'betaContentPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/graph/raw_all/oozie_app',
+ 'mongoURL' : 'mongodb://beta.services.openaire.eu',
+ 'mongoDb' : 'mdstore',
+ 'postgresURL' : 'jdbc:postgresql://beta.services.openaire.eu:5432/dnet_openaireplus',
+ 'postgresUser' : 'dnet',
+ 'postgresPassword' : '',
+ 'workingDir' : '/tmp/core_provision/working_dir/beta_aggregator'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ create the AGGREGATOR graph
+
+ executeOozieJob
+ IIS
+
+ {
+ 'graphOutputPath' : 'prodAggregatorGraphPath',
+ 'isLookupUrl' : 'isLookUpUrl',
+ 'reuseContent' : 'reuseProdContent',
+ 'contentPath' : 'prodContentPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/graph/raw_all/oozie_app',
+ 'mongoURL' : 'mongodb://services.openaire.eu',
+ 'mongoDb' : 'mdstore',
+ 'postgresURL' : 'jdbc:postgresql://postgresql.services.openaire.eu:5432/dnet_openaireplus',
+ 'postgresUser' : 'dnet',
+ 'postgresPassword' : '',
+ 'workingDir' : '/tmp/core_provision/working_dir/prod_aggregator'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ wait configurations
+
+
+
+
+
+
+
+ create the AGGREGATOR graph
+
+ executeOozieJob
+ IIS
+
+ {
+ 'betaInputGgraphPath' : 'betaAggregatorGraphPath',
+ 'prodInputGgraphPath' : 'prodAggregatorGraphPath',
+ 'graphOutputPath' : 'mergedGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/graph/merge/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/merge_graph'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ create the RAW graph
+
+ executeOozieJob
+ IIS
+
+ {
+ 'inputActionSetIds' : 'actionSetIdsRawGraph',
+ 'inputGraphRootPath' : 'mergedGraphPath',
+ 'outputGraphRootPath' : 'rawGraphPath',
+ 'isLookupUrl' : 'isLookUpUrl'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/actionmanager/wf/main/oozie_app',
+ 'sparkExecutorCores' : '3',
+ 'sparkExecutorMemory' : '10G',
+ 'activePromoteDatasetActionPayload' : 'true',
+ 'activePromoteDatasourceActionPayload' : 'true',
+ 'activePromoteOrganizationActionPayload' : 'true',
+ 'activePromoteOtherResearchProductActionPayload' : 'true',
+ 'activePromoteProjectActionPayload' : 'true',
+ 'activePromotePublicationActionPayload' : 'true',
+ 'activePromoteRelationActionPayload' : 'true',
+ 'activePromoteResultActionPayload' : 'true',
+ 'activePromoteSoftwareActionPayload' : 'true',
+ 'mergeAndGetStrategy' : 'MERGE_FROM_AND_GET',
+ 'workingDir' : '/tmp/core_provision/working_dir/promoteActionsRaw'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ search for duplicates in the raw graph
+
+ executeOozieJob
+ IIS
+
+ {
+ 'actionSetId' : 'dedupConfig',
+ 'graphBasePath' : 'rawGraphPath',
+ 'dedupGraphPath': 'dedupGraphPath',
+ 'isLookUpUrl' : 'isLookUpUrl'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/dedup/scan/oozie_app',
+ 'workingPath' : '/tmp/core_provision/working_dir/dedup'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ create the INFERRED graph
+
+ executeOozieJob
+ IIS
+
+ {
+ 'inputActionSetIds' : 'actionSetIdsIISGraph',
+ 'inputGraphRootPath' : 'dedupGraphPath',
+ 'outputGraphRootPath' : 'inferredGraphPath',
+ 'isLookupUrl' : 'isLookUpUrl'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/actionmanager/wf/main/oozie_app',
+ 'sparkExecutorCores' : '3',
+ 'sparkExecutorMemory' : '10G',
+ 'activePromoteDatasetActionPayload' : 'true',
+ 'activePromoteDatasourceActionPayload' : 'true',
+ 'activePromoteOrganizationActionPayload' : 'true',
+ 'activePromoteOtherResearchProductActionPayload' : 'true',
+ 'activePromoteProjectActionPayload' : 'true',
+ 'activePromotePublicationActionPayload' : 'true',
+ 'activePromoteRelationActionPayload' : 'true',
+ 'activePromoteResultActionPayload' : 'true',
+ 'activePromoteSoftwareActionPayload' : 'true',
+ 'mergeAndGetStrategy' : 'MERGE_FROM_AND_GET',
+ 'workingDir' : '/tmp/core_provision/working_dir/promoteActionsIIS'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ mark duplicates as deleted and redistribute the relationships
+
+ executeOozieJob
+ IIS
+
+ {
+ 'graphBasePath' : 'inferredGraphPath',
+ 'dedupGraphPath': 'consistentGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/dedup/consistency/oozie_app',
+ 'workingPath' : '/tmp/core_provision/working_dir/dedup'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ propagates ORCID among results linked by allowedsemrels semantic relationships
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'consistentGraphPath',
+ 'outputPath': 'orcidGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/orcidtoresultfromsemrel/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/orcid',
+ 'allowedsemrels' : 'isSupplementedBy;isSupplementTo',
+ 'saveGraph' : 'true'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ mark results respecting some rules as belonging to communities
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'orcidGraphPath',
+ 'outputPath': 'bulkTaggingGraphPath',
+ 'isLookUpUrl' : 'isLookUpUrl',
+ 'pathMap' : 'bulkTaggingPathMap'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/bulktag/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/bulktag'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ creates relashionships between results and organizations when the organizations are associated to institutional repositories
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'bulkTaggingGraphPath',
+ 'outputPath': 'affiliationGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/affiliation/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/affiliation',
+ 'saveGraph' : 'true'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ marks as belonging to communities the result collected from datasources related to the organizations specified in the organizationCommunityMap
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'affiliationGraphPath',
+ 'outputPath': 'communityOrganizationGraphPath',
+ 'organizationtoresultcommunitymap': 'propagationOrganizationCommunityMap'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/community_organization/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/community_organization',
+ 'saveGraph' : 'true'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ created relation between projects and results linked to other results trough allowedsemrel semantic relations linked to projects
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'communityOrganizationGraphPath',
+ 'outputPath': 'fundingGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/funding/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/funding',
+ 'allowedsemrels' : 'isSupplementedBy;isSupplementTo',
+ 'saveGraph' : 'true'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ tag as belonging to communitites result in in allowedsemrels relation with other result already linked to communities
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'fundingGraphPath',
+ 'outputPath': 'communitySemRelGraphPath',
+ 'isLookUpUrl' : 'isLookUpUrl'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/community_semrel/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/community_semrel',
+ 'allowedsemrels' : 'isSupplementedBy;isSupplementTo',
+ 'saveGraph' : 'true'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ associated to results colleced from allowedtypes and those in the whithelist the country of the organization(s) handling the datasource it is collected from
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'communitySemRelGraphPath',
+ 'outputPath': 'countryGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/country/oozie_app',
+ 'sparkExecutorCores' : '3',
+ 'sparkExecutorMemory' : '10G',
+ 'workingDir' : '/tmp/core_provision/working_dir/country',
+ 'allowedtypes' : 'pubsrepository::institutional',
+ 'whitelist' : '10|opendoar____::300891a62162b960cf02ce3827bb363c',
+ 'saveGraph' : 'true'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ clean the properties in the graph typed as Qualifier according to the vocabulary indicated in schemeid
+
+ executeOozieJob
+ IIS
+
+ {
+ 'graphInputPath' : 'countryGraphPath',
+ 'graphOutputPath': 'cleanedGraphPath',
+ 'isLookupUrl': 'isLookUpUrl'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/graph/clean/oozie_app',
+ 'workingPath' : '/tmp/core_provision/working_dir/clean'
+ }
+
+ build-report
+
+
+
+
+
+
+
+ removes blacklisted relations
+
+ executeOozieJob
+ IIS
+
+ {
+ 'sourcePath' : 'cleanedGraphPath',
+ 'outputPath': 'blacklistedGraphPath'
+ }
+
+
+ {
+ 'oozie.wf.application.path' : '/lib/dnet/oa/enrichment/blacklist/oozie_app',
+ 'workingDir' : '/tmp/core_provision/working_dir/blacklist',
+ 'postgresURL' : 'jdbc:postgresql://beta.services.openaire.eu:5432/dnet_openaireplus',
+ 'postgresUser' : 'dnet',
+ 'postgresPassword' : ''
+ }
+
+ build-report
+
+
+
+
+
+
+
+
+ wf_20200615_163630_609
+ 2020-06-15T17:08:00+00:00
+ SUCCESS
+
+
+
+
\ No newline at end of file
diff --git a/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_construction.xml b/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_prod_construction.xml
similarity index 99%
rename from dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_construction.xml
rename to dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_prod_construction.xml
index 4d77883b4..047433320 100644
--- a/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_construction.xml
+++ b/dhp-workflows/dhp-worfklow-profiles/src/main/resources/eu/dnetlib/dhp/wf/profiles/graph_prod_construction.xml
@@ -7,7 +7,7 @@
- Graph Construction [OCEAN]
+ Graph Construction [PROD]
Data Provision
30
@@ -413,7 +413,7 @@
build-report
-
+
diff --git a/pom.xml b/pom.xml
index 5ece9aaf6..cec3dd75a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -315,7 +315,7 @@
eu.dnetlib
dnet-pace-core
- 4.0.2
+ 4.0.4
eu.dnetlib