D-Net
/
dnet-hadoop
14
5
Fork 11
Code Issues 14 Pull Requests 12 Projects 2 Releases 2 Wiki Activity

Merge pull request 'Support for JDK 17 and s3 default file system (kubernetes environment)' (#495) from kubernetes into beta 

Reviewed-on: #495
This commit is contained in:
Giambattista Bloisi 2025-01-13 14:16:51 +01:00
parent 62bb82a8ef 357ec7414c
commit 8f0d1c2c9e
47 changed files with 1401 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -27,5 +27,6 @@ spark-warehouse
/**/.factorypath
/**/.scalafmt.conf
/.java-version
/dhp-shade-package/dependency-reduced-pom.xml
/**/dependency-reduced-pom.xml
/**/job.properties

37
dhp-common/pom.xml
View File

@ -62,6 +62,12 @@
</build>
<dependencies>
<dependency>
<groupId>eu.dnetlib</groupId>
<artifactId>dhp-rmi-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>edu.cmu</groupId>
<artifactId>secondstring</artifactId>
@ -125,17 +131,6 @@
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-transports-http</artifactId>
</dependency>
<dependency>
<groupId>eu.dnetlib</groupId>
<artifactId>cnr-rmi-api</artifactId>
<exclusions>
<exclusion>
<artifactId>log4j</artifactId>
<groupId>log4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.ximpleware</groupId>
<artifactId>vtd-xml</artifactId>
@ -174,24 +169,4 @@
<artifactId>opencsv</artifactId>
</dependency>
</dependencies>
<!-- dependencies required on JDK9+ because J2EE has been removed -->
<profiles>
<profile>
<id>spark-34</id>
<dependencies>
<dependency>
<groupId>javax.xml.bind</groupId>
<artifactId>jaxb-api</artifactId>
<version>2.2.11</version>
</dependency>
<dependency>
<groupId>com.sun.xml.ws</groupId>
<artifactId>jaxws-ri</artifactId>
<version>2.3.3</version>
<type>pom</type>
</dependency>
</dependencies>
</profile>
</profiles>
</project>

108
dhp-common/src/main/java/eu/dnetlib/dhp/oa/merge/CopyEntitiesSparkJob.java Normal file
View File

@ -0,0 +1,108 @@
package eu.dnetlib.dhp.oa.merge;
import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
import java.util.Arrays;
import java.util.Optional;
import org.apache.commons.io.IOUtils;
import org.apache.spark.SparkConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.schema.common.ModelSupport;
import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
/**
* Copy specified entities from a graph snapshot to another
*/
public class CopyEntitiesSparkJob {
private static final Logger log = LoggerFactory.getLogger(CopyEntitiesSparkJob.class);
private ArgumentApplicationParser parser;
public CopyEntitiesSparkJob(ArgumentApplicationParser parser) {
this.parser = parser;
}
public static void main(String[] args) throws Exception {
String jsonConfiguration = IOUtils
.toString(
CopyEntitiesSparkJob.class
.getResourceAsStream(
"/eu/dnetlib/dhp/oa/merge/copy_graph_entities_parameters.json"));
final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
parser.parseArgument(args);
Boolean isSparkSessionManaged = Optional
.ofNullable(parser.get("isSparkSessionManaged"))
.map(Boolean::valueOf)
.orElse(Boolean.TRUE);
log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
new CopyEntitiesSparkJob(parser).run(isSparkSessionManaged);
}
public void run(Boolean isSparkSessionManaged)
throws ISLookUpException {
String graphInputPath = parser.get("graphInputPath");
log.info("graphInputPath: {}", graphInputPath);
String outputPath = parser.get("outputPath");
log.info("outputPath: {}", outputPath);
String entities = parser.get("entities");
log.info("entities: {}", entities);
String format = parser.get("format");
log.info("format: {}", format);
SparkConf conf = new SparkConf();
runWithSparkSession(
conf,
isSparkSessionManaged,
spark -> {
Arrays
.stream(entities.split(","))
.map(x -> x.trim().toLowerCase())
.filter(ModelSupport.oafTypes::containsKey)
.forEachOrdered(
entity -> {
switch (format.toLowerCase()) {
case "text":
spark
.read()
.text(graphInputPath + "/" + entity)
.write()
.option("compression", "gzip")
.mode("overwrite")
.text(outputPath + "/" + entity);
break;
case "json":
spark
.read()
.json(graphInputPath + "/" + entity)
.write()
.option("compression", "gzip")
.mode("overwrite")
.json(outputPath + "/" + entity);
break;
case "parquet":
spark
.read()
.parquet(graphInputPath + "/" + entity)
.write()
.option("compression", "gzip")
.mode("overwrite")
.parquet(outputPath + "/" + entity);
break;
}
});
});
}
}

2
dhp-common/src/main/java/eu/dnetlib/dhp/oa/merge/GroupEntitiesSparkJob.java
View File

@ -85,8 +85,6 @@ public class GroupEntitiesSparkJob {
log.info("filterInvisible: {}", filterInvisible);
SparkConf conf = new SparkConf();
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
conf.registerKryoClasses(ModelSupport.getOafModelClasses());
final VocabularyGroup vocs = VocabularyGroup.loadVocsFromIS(isLookUpService);

32
dhp-common/src/main/resources/eu/dnetlib/dhp/oa/merge/copy_graph_entities_parameters.json Normal file
View File

@ -0,0 +1,32 @@
[
{
"paramName": "issm",
"paramLongName": "isSparkSessionManaged",
"paramDescription": "when true will stop SparkSession after job execution",
"paramRequired": false
},
{
"paramName": "gin",
"paramLongName": "graphInputPath",
"paramDescription": "the input graph root path",
"paramRequired": true
},
{
"paramName": "out",
"paramLongName": "outputPath",
"paramDescription": "the output graph root path",
"paramRequired": true
},
{
"paramName": "ent",
"paramLongName": "entities",
"paramDescription": "the output graph root path",
"paramRequired": true
},
{
"paramName": "fmt",
"paramLongName": "format",
"paramDescription": "the output graph root path",
"paramRequired": true
}
]

3
dhp-common/src/main/scala/eu/dnetlib/dhp/application/SparkScalaApplication.scala
View File

@ -2,6 +2,7 @@ package eu.dnetlib.dhp.application
import eu.dnetlib.dhp.common.Constants
import eu.dnetlib.dhp.utils.DHPUtils.writeHdfsFile
import org.apache.commons.lang3.StringUtils
import scala.io.Source
@ -69,7 +70,7 @@ abstract class AbstractScalaApplication(
.builder()
.config(conf)
.appName(getClass.getSimpleName)
if (master != null)
if (StringUtils.isNotBlank(master))
b.master(master)
b.getOrCreate()
}

34
dhp-rmi-api/pom.xml Normal file
View File

@ -0,0 +1,34 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>eu.dnetlib.dhp</groupId>
<artifactId>dhp</artifactId>
<version>1.2.5-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>eu.dnetlib</groupId>
<artifactId>dhp-rmi-api</artifactId>
<packaging>jar</packaging>
<version>1.2.5-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-core</artifactId>
<version>${cxf.version}</version>
</dependency>
<dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-frontend-jaxws</artifactId>
<version>${cxf.version}</version>
</dependency>
<dependency>
<groupId>org.glassfish.metro</groupId>
<artifactId>webservices-rt</artifactId>
<version>${metro.version}</version>
</dependency>
</dependencies>
</project>

11
dhp-rmi-api/src/main/java/eu/dnetlib/common/rmi/APIDeprecatedException.java Normal file
View File

@ -0,0 +1,11 @@
package eu.dnetlib.common.rmi;
public class APIDeprecatedException extends RuntimeException {
/**
*
*/
private static final long serialVersionUID = -5606373588445519515L;
}

35
dhp-rmi-api/src/main/java/eu/dnetlib/common/rmi/BaseService.java Normal file
View File

@ -0,0 +1,35 @@
package eu.dnetlib.common.rmi;
import jakarta.jws.WebMethod;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface BaseService {
/**
* All DRIVER services must implement method notify() in order to communicate with the IS_SN
*
* @param subsrciptionId
* @param topic
* @param isId
* @param message
*/
@WebMethod(operationName = "notify")
void notify(@WebParam(name = "subscrId") String subscriptionId,
@WebParam(name = "topic") String topic,
@WebParam(name = "is_id") String isId,
@WebParam(name = "message") String message);
/**
* Identifies the service's version. Version syntax: ${NAME}-${MAJOR}.${MINOR}.${MICRO}[-${LABEL}]
*
* @return the service's version
*/
@WebMethod(operationName = "identify")
String identify();
void start();
}

19
dhp-rmi-api/src/main/java/eu/dnetlib/common/rmi/DNetRestDocumentation.java Normal file
View File

@ -0,0 +1,19 @@
package eu.dnetlib.common.rmi;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* Created by claudio on 30/11/2016.
* to be used in REST controllers, and autodiscovered to build and publish their documentation
*/
@Target({
ElementType.TYPE
})
@Retention(RetentionPolicy.RUNTIME)
public @interface DNetRestDocumentation {
}

27
dhp-rmi-api/src/main/java/eu/dnetlib/common/rmi/RMIException.java Normal file
View File

@ -0,0 +1,27 @@
package eu.dnetlib.common.rmi;
/**
* All RMI exception thrown from the service remote method invocation interfaces inherit this class
*
* @author marko
*/
abstract public class RMIException extends Exception { // NOPMD
/**
*
*/
private static final long serialVersionUID = 428841258652765265L;
public RMIException(final Throwable exception) {
super(exception);
}
public RMIException(final String string) {
super(string);
}
public RMIException(final String string, final Throwable exception) {
super(string, exception);
}
}

11
dhp-rmi-api/src/main/java/eu/dnetlib/common/rmi/UnimplementedException.java Normal file
View File

@ -0,0 +1,11 @@
package eu.dnetlib.common.rmi;
public class UnimplementedException extends RuntimeException {
/**
*
*/
private static final long serialVersionUID = 6040968020696349497L;
}

31
dhp-rmi-api/src/main/java/eu/dnetlib/data/information/collectionservice/rmi/CollectionService.java Normal file
View File

@ -0,0 +1,31 @@
package eu.dnetlib.data.information.collectionservice.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
/**
* The Collection Service is used to ...
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface CollectionService extends BaseService {
String getCollection(@WebParam(name = "collId")
final String collId) throws CollectionServiceException;
List<String> getCollections(@WebParam(name = "collIds")
final List<String> collIds) throws CollectionServiceException;
void updateCollection(@WebParam(name = "coll")
final String coll) throws CollectionServiceException;
void deleteCollection(@WebParam(name = "collId")
final String collId) throws CollectionServiceException;
String createCollection(@WebParam(name = "coll")
final String coll) throws CollectionServiceException;
}

27
dhp-rmi-api/src/main/java/eu/dnetlib/data/information/collectionservice/rmi/CollectionServiceException.java Normal file
View File

@ -0,0 +1,27 @@
package eu.dnetlib.data.information.collectionservice.rmi;
import eu.dnetlib.common.rmi.RMIException;
import jakarta.xml.ws.WebFault;
@WebFault
public class CollectionServiceException extends RMIException {
/**
*
*/
private static final long serialVersionUID = 8094008463553904905L;
public CollectionServiceException(Throwable e) {
super(e);
}
public CollectionServiceException(String message, Throwable e) {
super(message, e);
}
public CollectionServiceException(String message) {
super(message);
}
}

57
dhp-rmi-api/src/main/java/eu/dnetlib/data/information/publisher/rmi/PublisherService.java Normal file
View File

@ -0,0 +1,57 @@
package eu.dnetlib.data.information.publisher.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebMethod;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
import jakarta.xml.ws.wsaddressing.W3CEndpointReference;
/**
* Publisher service. Provides access to metadata records and objects.
*
* @author marko
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface PublisherService extends BaseService {
/**
* Get a (metadata) resource by ID.
*
* @param id
* @param format
* @param layout
* @param interpretation
* @return
*/
@WebMethod
String getResourceById(@WebParam(name = "id")
final String id,
@WebParam(name = "format")
final String format,
@WebParam(name = "layout")
final String layout,
@WebParam(name = "interpretation")
final String interpretation);
/**
* Get (metadata) resources by IDs.
*
* @param ids
* @param format
* @param layout
* @param interpretation
* @return
*/
@WebMethod
W3CEndpointReference getResourcesByIds(@WebParam(name = "ids")
final List<String> ids,
@WebParam(name = "format")
final String format,
@WebParam(name = "layout")
final String layout,
@WebParam(name = "interpretation")
final String interpretation);
}

30
dhp-rmi-api/src/main/java/eu/dnetlib/data/mdstore/DocumentNotFoundException.java Normal file
View File

@ -0,0 +1,30 @@
package eu.dnetlib.data.mdstore;
/**
* Signals that a metadata record cannot be found in a given MDStore.
*/
public class DocumentNotFoundException extends MDStoreServiceException {
/**
*
*/
private static final long serialVersionUID = 5188036989114250548L;
public DocumentNotFoundException(final String s, final Throwable e) {
super(s, e);
}
public DocumentNotFoundException(final String s) {
super(s);
}
public DocumentNotFoundException(final Throwable e) {
super(e);
}
public DocumentNotFoundException() {
super();
}
}

118
dhp-rmi-api/src/main/java/eu/dnetlib/data/mdstore/MDStoreService.java Normal file
View File

@ -0,0 +1,118 @@
package eu.dnetlib.data.mdstore;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebMethod;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
import jakarta.xml.ws.wsaddressing.W3CEndpointReference;
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface MDStoreService extends BaseService {
/**
* Identifies service and version.
*
* @return
*/
@Override
String identify();
/**
* Returns ResultSet EPR for delivered mdstore records.
*
* @param mdId
* @param from
* @param until
* @param recordFilter REGEX on the metadata record
* @return ResultSet EPR
* @throws MDStoreServiceException
*/
W3CEndpointReference deliverMDRecords(@WebParam(name = "mdId")
final String mdId,
@WebParam(name = "from")
final String from,
@WebParam(name = "until")
final String until,
@WebParam(name = "recordsFilter")
final String recordFilter) throws MDStoreServiceException;
/**
* Deliver single record from selected mdstore.
*
* @param mdId
* @param recordId
* @return record
* @throws MDStoreServiceException
*/
String deliverRecord(@WebParam(name = "mdId")
final String mdId, @WebParam(name = "recordId")
final String recordId) throws MDStoreServiceException;
/**
* Returns list of all stored indices.
*
* @return list of all stored indices
*/
List<String> getListOfMDStores() throws MDStoreServiceException;
List<String> listMDStores(@WebParam(name = "format")
final String format,
@WebParam(name = "layout")
final String layout,
@WebParam(name = "interpretation")
final String interpretation) throws MDStoreServiceException;
W3CEndpointReference bulkDeliverMDRecords(@WebParam(name = "format")
final String format,
@WebParam(name = "layout")
final String layout,
@WebParam(name = "interpretation")
final String interpretation) throws MDStoreServiceException;
/**
* Store md records from a result set
*
* @param mdId
* @param rsId
* @param storingType
* @return returns true immediately.
* @throws MDStoreServiceException
*/
@Deprecated
boolean storeMDRecordsFromRS(@WebParam(name = "mdId")
final String mdId,
@WebParam(name = "rsId")
final String rsId,
@WebParam(name = "storingType")
final String storingType) throws MDStoreServiceException;
/**
* Gets the size of the mdstore with the given identifier.
*
* @param mdId identifier of an mdstore
* @return the number of records in the store
*/
@WebMethod(operationName = "size")
int size(@WebParam(name = "mdId")
final String mdId) throws MDStoreServiceException;
/**
* Gets the sum of records stored in all mdstore with the given format, layout , interpretation
*
* @param format format
* @param layout layout
* @param interpretation interpretation
* @return the total number of records in the mdstores of the given type
*/
@WebMethod(operationName = "sizeByFormat")
int size(@WebParam(name = "format")
final String format,
@WebParam(name = "layout")
final String layout,
@WebParam(name = "interpretation")
final String interpretation) throws MDStoreServiceException;
}

33
dhp-rmi-api/src/main/java/eu/dnetlib/data/mdstore/MDStoreServiceException.java Normal file
View File

@ -0,0 +1,33 @@
package eu.dnetlib.data.mdstore;
/**
* General mdstore service exception.
*
* @author claudio atzori
* @version 1.0.0
*/
public class MDStoreServiceException extends Exception {
/**
*
*/
private static final long serialVersionUID = -6772977735282310658L;
public MDStoreServiceException(String s, Throwable e) {
super(s, e);
}
public MDStoreServiceException(String s) {
super(s);
}
public MDStoreServiceException(Throwable e) {
super(e);
}
public MDStoreServiceException() {
super();
}
}

28
dhp-rmi-api/src/main/java/eu/dnetlib/data/utility/objectpackaging/rmi/ObjectPackagingException.java Normal file
View File

@ -0,0 +1,28 @@
package eu.dnetlib.data.utility.objectpackaging.rmi;
import eu.dnetlib.common.rmi.RMIException;
import jakarta.xml.ws.WebFault;
@WebFault
public class ObjectPackagingException extends RMIException {
private static final long serialVersionUID = 3468254939586031822L;
/**
*
*/
public ObjectPackagingException(Throwable e) {
super(e);
}
public ObjectPackagingException(String message, Throwable e) {
super(message, e);
}
public ObjectPackagingException(String message) {
super(message);
}
}

37
dhp-rmi-api/src/main/java/eu/dnetlib/data/utility/objectpackaging/rmi/ObjectPackagingService.java Normal file
View File

@ -0,0 +1,37 @@
package eu.dnetlib.data.utility.objectpackaging.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
import jakarta.xml.ws.wsaddressing.W3CEndpointReference;
/**
* The Object Packaging Service is used to combine the records spread
* into one information package, namely an Object Record.
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface ObjectPackagingService extends BaseService {
/**
* Return the EPR of the resultSet containing the generated packages
*
* @param eprs A list of EPRs used to access the input resultSets. ResultSets MUST be ordered using an order key identified by xpath_ID
* @param xpath_ID A valid xpath, used to access the ordered ID of the elements of the input resultSets.
* @return EPR of the generated resultset
*/
W3CEndpointReference generatePackages(@WebParam(name = "eprs") List<W3CEndpointReference> eprs,
@WebParam(name = "xpath_ID") String xpath_ID) throws ObjectPackagingException;
/**
* Return the EPR of the resultSet containing the unpackaged element
*
* @param epr The epr used to access the resultset that contains input packages, packages are xml record in this format: <objectRecord><elem>REC1</elem><elem>REC2</elem><elem>REC3</elem></objectRecord>
* @return EPR of the generated resultset
*/
W3CEndpointReference splitPackages(@WebParam(name = "epr") W3CEndpointReference epr)
throws ObjectPackagingException;
}

17
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/dlm/rmi/DlmService.java Normal file
View File

@ -0,0 +1,17 @@
package eu.dnetlib.enabling.dlm.rmi;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebService;
/**
* Distributed lock manager. Currently is used mostly to start the underlying lock manager (e.g. zookeeper) and let
* client interface directly with it.
*
* <p>The DLM service profile contains the entry point of the underlying locking service.</p>
*
* @author marko
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface DlmService extends BaseService {
}

19
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/hcm/rmi/HostingContextManagerService.java Normal file
View File

@ -0,0 +1,19 @@
package eu.dnetlib.enabling.hcm.rmi;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebService;
/**
* Like a HostingNodeManager, but any webapp (web context) can have its own.
* <p>
* useful for dispatching notifications shared by all the services local to a single context.
* </p>
*
* @author marko
* @author antonis
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface HostingContextManagerService extends BaseService {
}

15
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/hnm/rmi/HostingNodeManagerService.java Normal file
View File

@ -0,0 +1,15 @@
package eu.dnetlib.enabling.hnm.rmi;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
/**
* The HostingNodeManager Service is used to ...
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface HostingNodeManagerService extends BaseService {
String echo(@WebParam(name = "s") String s);
}

47
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/lookup/rmi/ISLookUpDocumentNotFoundException.java Normal file
View File

@ -0,0 +1,47 @@
package eu.dnetlib.enabling.is.lookup.rmi;
import jakarta.xml.ws.WebFault;
/**
* Thrown when a given document is not found.
*
* @author marko
*/
@WebFault
public class ISLookUpDocumentNotFoundException extends ISLookUpException {
/**
* exception chain + message.
*
* @param message message
* @param e
*/
public ISLookUpDocumentNotFoundException(String message, Throwable e) {
super(message, e);
}
/**
* exception chain constructor.
*
* @param e
*/
public ISLookUpDocumentNotFoundException(Throwable e) {
super(e);
}
/**
* exception message.
*
* @param message
*/
public ISLookUpDocumentNotFoundException(String message) {
super(message);
}
/**
*
*/
private static final long serialVersionUID = 2295995755165801937L;
}

27
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/lookup/rmi/ISLookUpException.java Normal file
View File

@ -0,0 +1,27 @@
package eu.dnetlib.enabling.is.lookup.rmi;
import eu.dnetlib.common.rmi.RMIException;
import jakarta.xml.ws.WebFault;
@WebFault
public class ISLookUpException extends RMIException {
/**
*
*/
private static final long serialVersionUID = -5626136963653382533L;
public ISLookUpException(Throwable e) {
super(e);
}
public ISLookUpException(String message, Throwable e) {
super(message, e);
}
public ISLookUpException(String message) {
super(message);
}
}

57
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/lookup/rmi/ISLookUpService.java Normal file
View File

@ -0,0 +1,57 @@
package eu.dnetlib.enabling.is.lookup.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface ISLookUpService extends BaseService {
Boolean flushCachedResultSets();
@Deprecated
String getCollection(@WebParam(name = "profId") String profId, @WebParam(name = "format") String format)
throws ISLookUpException;
String retrieveCollection(@WebParam(name = "profId") String profId) throws ISLookUpException;
String getResourceProfile(@WebParam(name = "profId") String profId)
throws ISLookUpException;
String getResourceProfileByQuery(@WebParam(name = "XQuery") String XQuery)
throws ISLookUpException;
String getResourceQoSParams(@WebParam(name = "id") String id) throws ISLookUpException;
String getResourceTypeSchema(@WebParam(name = "resourceType") String resourceType)
throws ISLookUpException;
List<String> listCollections(
@WebParam(name = "format") String format,
@WebParam(name = "idfather") String idfather,
@WebParam(name = "owner") String owner) throws ISLookUpException;
@Deprecated
List<String> listDHNIDs() throws ISLookUpException;
List<String> listResourceTypes() throws ISLookUpException;
@Deprecated
List<String> listServiceIDs(@WebParam(name = "serviceType") String serviceType) throws ISLookUpException;
@Deprecated
List<String> listServiceTypes() throws ISLookUpException;
/**
* Like searchProfile(), but bypassing the resultset. Useful for short xquery results.
*
* @param xquery xquery to be executed
* @return list of strings (never null)
* @throws ISLookUpException could happen
*/
List<String> quickSearchProfile(@WebParam(name = "XQuery") String xquery) throws ISLookUpException;
}

28
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/registry/ISRegistryDocumentNotFoundException.java Normal file
View File

@ -0,0 +1,28 @@
package eu.dnetlib.enabling.is.registry;
import eu.dnetlib.enabling.is.registry.rmi.ISRegistryException;
public class ISRegistryDocumentNotFoundException extends ISRegistryException {
/**
*
*/
private static final long serialVersionUID = -1304948213334188538L;
public ISRegistryDocumentNotFoundException(String string, Throwable e) {
super(string, e);
// TODO Auto-generated constructor stub
}
public ISRegistryDocumentNotFoundException(String string) {
super(string);
// TODO Auto-generated constructor stub
}
public ISRegistryDocumentNotFoundException(Throwable e) {
super(e);
// TODO Auto-generated constructor stub
}
}

25
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/registry/rmi/ISRegistryException.java Normal file
View File

@ -0,0 +1,25 @@
package eu.dnetlib.enabling.is.registry.rmi;
import eu.dnetlib.common.rmi.RMIException;
public class ISRegistryException extends RMIException {
/**
*
*/
private static final long serialVersionUID = -3347405941287624771L;
public ISRegistryException(Throwable e) {
super(e);
}
public ISRegistryException(String string) {
super(string);
}
public ISRegistryException(String string, Throwable e) {
super(string, e);
}
}

72
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/registry/rmi/ISRegistryService.java Normal file
View File

@ -0,0 +1,72 @@
package eu.dnetlib.enabling.is.registry.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebService;
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface ISRegistryService extends BaseService {
boolean addOrUpdateResourceType(String resourceType, String resourceSchema) throws ISRegistryException;
boolean addResourceType(String resourceType, String resourceSchema) throws ISRegistryException;
boolean deleteProfile(String profId) throws ISRegistryException;
@Deprecated
boolean deleteProfiles(List<String> arrayprofId) throws ISRegistryException;
/**
* @param resourceType
* @param hierarchical remove subscription topics
* @return
* @throws ISRegistryException
*/
boolean deleteResourceType(String resourceType, Boolean hierarchical) throws ISRegistryException;
boolean executeXUpdate(String XQuery) throws ISRegistryException;
String insertProfileForValidation(String resourceType, String resourceProfile) throws ISRegistryException;
String invalidateProfile(String profId) throws ISRegistryException;
boolean refreshProfile(String profId, String resourceType) throws ISRegistryException;
/**
* register a XML Profile.
*
* @param resourceProfile xml profile
* @return profile id
* @throws ISRegistryException
*/
String registerProfile(String resourceProfile) throws ISRegistryException;
String registerSecureProfile(String resourceProfId, String secureProfId) throws ISRegistryException;
boolean updateProfile(String profId, String resourceProfile, String resourceType) throws ISRegistryException;
@Deprecated
String updateProfileDHN(String resourceProfile) throws ISRegistryException;
boolean addProfileNode(String profId, String xpath, String node) throws ISRegistryException;
boolean updateProfileNode(String profId, String xpath, String node) throws ISRegistryException;
boolean removeProfileNode(String profId, String nodeId) throws ISRegistryException;
@Deprecated
boolean updateRegionDescription(String profId, String resourceProfile) throws ISRegistryException;
String validateProfile(String profId) throws ISRegistryException;
@Deprecated
List<String> validateProfiles(List<String> profIds) throws ISRegistryException;
void addBlackBoardMessage(String profId, String messageId, String message) throws ISRegistryException;
void replyBlackBoardMessage(String profId, String message) throws ISRegistryException;
void deleteBlackBoardMessage(String profId, String messageId) throws ISRegistryException;
}

25
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/sn/rmi/ISSNException.java Normal file
View File

@ -0,0 +1,25 @@
package eu.dnetlib.enabling.is.sn.rmi;
import eu.dnetlib.common.rmi.RMIException;
public class ISSNException extends RMIException {
/**
*
*/
private static final long serialVersionUID = -7384073901457430004L;
public ISSNException(final Throwable e) {
super(e);
}
public ISSNException(final String message) {
super(message);
}
public ISSNException(final String message, final Throwable e) {
super(message, e);
}
}

121
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/sn/rmi/ISSNService.java Normal file
View File

@ -0,0 +1,121 @@
package eu.dnetlib.enabling.is.sn.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
import jakarta.xml.ws.wsaddressing.W3CEndpointReference;
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface ISSNService extends BaseService {
/**
* fossil.
*
* @param topic
* @return
* @throws ISSNException
*/
String getCurrentMessage(@WebParam(name = "topic") String topic) throws ISSNException;
/**
* puts a subcription in a paused state. paused subscription are not notified even when triggered.
*
* @param subscrId subscription identifier
* @return returns false if the subscription is already paused.
* @throws ISSNException may happen
*/
boolean pauseSubscription(@WebParam(name = "subscrId") String subscrId) throws ISSNException;
/**
* Used to renew the subscription before it expires.
*
* <p>
* In practice it resets the ttl to another value, so it can be used to reset a infinte ttl subscription to a finite
* value.
* </p>
*
* @param subscrId subscription id
* @param terminationTime new ttl (from now), or 0 (infinite)
* @return true if successful
* @throws ISSNException may happen
*/
boolean renew(@WebParam(name = "subscrId") String subscrId, @WebParam(name = "terminationTime") int terminationTime)
throws ISSNException;
/**
* resumes a paused subscription.
*
* @param subscrId subscription id
* @return true if resumed. false if it was not paused.
* @throws ISSNException may happen
*/
boolean resumeSubscription(@WebParam(name = "subscrId") String subscrId) throws ISSNException;
/**
* @param consumerReference epr to be called when the notification is triggered
* @param topicExpression topic expression to register
* @param initialTerminationTime ttl in seconds (0 = infinite)
* @return subscription id
* @throws ISSNException may happen
*/
String subscribe(
@WebParam(name = "consumerReference") W3CEndpointReference consumerReference,
@WebParam(name = "topicExpression") String topicExpression,
@WebParam(name = "initialTerminationTime") int initialTerminationTime)
throws ISSNException;
boolean unsubscribe(@WebParam(name = "subscrId") String subscrId) throws ISSNException;
/**
* fossil.
*
* @param resourceType
* @param profileId
* @param profile
* @return
* @throws ISSNException
*/
boolean actionCreatePerformed(
@WebParam(name = "resourceType") String resourceType,
@WebParam(name = "profileId") String profileId,
@WebParam(name = "profile") String profile) throws ISSNException;
/**
* fossil.
*
* @param resourceType
* @param profileId
* @param profileBefore
* @param profileAfter
* @return
* @throws ISSNException
*/
boolean actionUpdatePerformed(
@WebParam(name = "resourceType") String resourceType,
@WebParam(name = "profileId") String profileId,
@WebParam(name = "profileBefore") String profileBefore,
@WebParam(name = "profileAfter") String profileAfter) throws ISSNException;
/**
* fossil.
*
* @param resourceType
* @param profileId
* @return
* @throws ISSNException
*/
boolean actionDeletePerformed(@WebParam(name = "resourceType") String resourceType,
@WebParam(name = "profileId") String profileId)
throws ISSNException;
/**
* list all subscriptions. Mostly for debug reasons.
*
* @return list of subscription ids.
*/
List<String> listSubscriptions();
}

21
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/sn/rmi/SubscriptionRequestRejectedException.java Normal file
View File

@ -0,0 +1,21 @@
package eu.dnetlib.enabling.is.sn.rmi;
/**
* Thrown when a subscription request is rejected.
*
* @author claudio
*/
public class SubscriptionRequestRejectedException extends ISSNException {
/**
*
*/
private static final long serialVersionUID = 263095606953662098L;
public SubscriptionRequestRejectedException(String message) {
super(message);
// TODO Auto-generated constructor stub
}
}

25
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/store/rmi/ISStoreException.java Normal file
View File

@ -0,0 +1,25 @@
package eu.dnetlib.enabling.is.store.rmi;
import eu.dnetlib.common.rmi.RMIException;
public class ISStoreException extends RMIException {
/**
*
*/
private static final long serialVersionUID = 8683126829156096420L;
public ISStoreException(Throwable e) {
super(e);
}
public ISStoreException(String message, Throwable e) {
super(message, e);
}
public ISStoreException(String message) {
super(message);
}
}

47
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/is/store/rmi/ISStoreService.java Normal file
View File

@ -0,0 +1,47 @@
package eu.dnetlib.enabling.is.store.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface ISStoreService extends BaseService {
boolean createFileColl(@WebParam(name = "fileColl") String fileColl) throws ISStoreException;
boolean deleteFileColl(@WebParam(name = "fileColl") String fileColl) throws ISStoreException;
boolean deleteXML(@WebParam(name = "fileName") String fileName, @WebParam(name = "fileColl") String fileColl)
throws ISStoreException;
boolean executeXUpdate(@WebParam(name = "query") String query) throws ISStoreException;
List<String> getFileColls() throws ISStoreException;
List<String> getFileNames(@WebParam(name = "fileColl") String fileColl) throws ISStoreException;
String getXML(@WebParam(name = "fileName") String fileName, @WebParam(name = "fileColl") String fileColl)
throws ISStoreException;
String getXMLbyQuery(@WebParam(name = "query") String query) throws ISStoreException;
boolean insertXML(@WebParam(name = "fileName") String fileName, @WebParam(name = "fileColl") String fileColl,
@WebParam(name = "file") String file)
throws ISStoreException;
boolean reindex();
List<String> quickSearchXML(@WebParam(name = "query") String query) throws ISStoreException;
boolean sync();
boolean updateXML(@WebParam(name = "fileName") String fileName, @WebParam(name = "fileColl") String fileColl,
@WebParam(name = "file") String file)
throws ISStoreException;
String backup() throws ISStoreException;
}

22
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/resultset/rmi/ResultSetException.java Normal file
View File

@ -0,0 +1,22 @@
package eu.dnetlib.enabling.resultset.rmi;
import eu.dnetlib.common.rmi.RMIException;
public class ResultSetException extends RMIException {
/**
*
*/
private static final long serialVersionUID = -7130554407601059627L;
public ResultSetException(Throwable e) {
super(e);
// TODO Auto-generated constructor stub
}
public ResultSetException(String string) {
super(string);
}
}

135
dhp-rmi-api/src/main/java/eu/dnetlib/enabling/resultset/rmi/ResultSetService.java Normal file
View File

@ -0,0 +1,135 @@
package eu.dnetlib.enabling.resultset.rmi;
import java.util.List;
import eu.dnetlib.common.rmi.BaseService;
import jakarta.jws.WebMethod;
import jakarta.jws.WebParam;
import jakarta.jws.WebService;
import jakarta.xml.ws.wsaddressing.W3CEndpointReference;
/**
* ResultSet service interface.
* <p>
* TODO: implement other compatibility methods as needed.
*
* @author marko
*/
@WebService(targetNamespace = "http://services.dnetlib.eu/")
public interface ResultSetService extends BaseService {
/**
* create a new pull rs.
*
* @param bdId bulk data identifier
* @param initialPageSize page size for the polling on the server side.
* @param expiryTime RS expiry time
* @return
*/
W3CEndpointReference createPullRSEPR(
@WebParam(name = "dataProviderServiceAddress") W3CEndpointReference dataProviderEPR,
@WebParam(name = "bdId") String bdId,
@WebParam(name = "initialPageSize") int initialPageSize,
@WebParam(name = "expiryTime") int expiryTime,
@WebParam(name = "styleSheet") String styleSheet,
@WebParam(name = "keepAliveTime") Integer keepAliveTime,
@WebParam(name = "total") Integer total);
/**
* create a new pull rs.
* <p>
* compatibility version
*
* @param bdId bulk data identifier
* @param initialPageSize page size for the polling on the server side.
* @param expiryTime RS expiry time
* @return
*/
W3CEndpointReference createPullRS(
@WebParam(name = "dataProviderServiceAddress") String dataProviderServiceAddress,
@WebParam(name = "bdId") String bdId,
@WebParam(name = "initialPageSize") int initialPageSize,
@WebParam(name = "expiryTime") int expiryTime,
@WebParam(name = "styleSheet") String styleSheet,
@WebParam(name = "keepAliveTime") Integer keepAliveTime,
@WebParam(name = "total") Integer total);
/**
* close a result set. A closed resultset is guaranteed not to grow.
*
* @param rsId
*/
void closeRS(@WebParam(name = "rsId") String rsId);
/**
* get one 'page' of results.
* <p>
* TODO: define how results are returned when the range is not present in the result set.
*
* @param fromPosition counting from 1
* @param toPosition included
* @param requestMode
* @return a page of data
*/
List<String> getResult(
@WebParam(name = "rsId") String rsId,
@WebParam(name = "fromPosition") int fromPosition,
@WebParam(name = "toPosition") int toPosition,
@WebParam(name = "requestMode") String requestMode) throws ResultSetException;
/**
* get the number of result elements present in the resultset.
*
* @param rsId result set identifier
* @return number of results available in the resultset
* @throws ResultSetException
*/
int getNumberOfElements(@WebParam(name = "rsId") String rsId) throws ResultSetException;
/**
* create a new push resultset.
*
* @param expiryTime RS expiry time
* @param keepAliveTime keep alive time
* @return epr of new resultset
* @throws ResultSetException
*/
W3CEndpointReference createPushRS(@WebParam(name = "expiryTime") int expiryTime,
@WebParam(name = "keepAliveTime") int keepAliveTime)
throws ResultSetException;
/**
* add new data to a push resultset.
*
* @param rsId resultset id
* @param elements list of elements to be addded
* @return dummy value
* @throws ResultSetException
*/
String populateRS(@WebParam(name = "rsId") String rsId, @WebParam(name = "elements") List<String> elements)
throws ResultSetException;
/**
* return current status of a resultset.
*
* @param rsId resultset id
* @return status
* @throws ResultSetException
*/
String getRSStatus(@WebParam(name = "rsId") String rsId) throws ResultSetException;
/**
* read a resultset property.
*
* @param rsId resultset id
* @param name property value
* @return property value
* @throws ResultSetException
*/
String getProperty(@WebParam(name = "rsId") String rsId, @WebParam(name = "name") String name)
throws ResultSetException;
@WebMethod(operationName = "identify")
String identify();
}

113
dhp-shade-package/dependency-reduced-pom.xml
View File

@ -1,113 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<artifactId>dhp</artifactId>
<groupId>eu.dnetlib.dhp</groupId>
<version>1.2.5-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>dhp-shade-package</artifactId>
<description>This module create a jar of all module dependencies</description>
<build>
<plugins>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer>
<mainClass>eu.dnetlib.dhp.oa.dedup.SparkCreateSimRels</mainClass>
</transformer>
<transformer />
<transformer>
<resource>META-INF/cxf/bus-extensions.txt</resource>
</transformer>
</transformers>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/maven/**</exclude>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<relocations>
<relocation>
<pattern>com</pattern>
<shadedPattern>repackaged.com.google.common</shadedPattern>
<includes>
<include>com.google.common.**</include>
</includes>
</relocation>
</relocations>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<dependencies>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.28</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
<version>5.6.1</version>
<scope>test</scope>
<exclusions>
<exclusion>
<artifactId>junit-jupiter-api</artifactId>
<groupId>org.junit.jupiter</groupId>
</exclusion>
<exclusion>
<artifactId>junit-jupiter-params</artifactId>
<groupId>org.junit.jupiter</groupId>
</exclusion>
<exclusion>
<artifactId>junit-jupiter-engine</artifactId>
<groupId>org.junit.jupiter</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>3.3.3</version>
<scope>test</scope>
<exclusions>
<exclusion>
<artifactId>byte-buddy</artifactId>
<groupId>net.bytebuddy</groupId>
</exclusion>
<exclusion>
<artifactId>byte-buddy-agent</artifactId>
<groupId>net.bytebuddy</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-junit-jupiter</artifactId>
<version>3.3.3</version>
<scope>test</scope>
</dependency>
</dependencies>
<distributionManagement>
<site>
<id>DHPSite</id>
<url>${dhp.site.stage.path}/dhp-common</url>
</site>
</distributionManagement>
</project>

21
dhp-shade-package/pom.xml
View File

@ -25,7 +25,6 @@
<dependencies>
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-actionmanager</artifactId>-->
@ -46,21 +45,21 @@
<!-- <artifactId>dhp-broker-events</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-dedup-openaire</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<dependency>
<groupId>eu.dnetlib.dhp</groupId>
<artifactId>dhp-dedup-openaire</artifactId>
<version>${project.version}</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-enrichment</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-graph-mapper</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<dependency>
<groupId>eu.dnetlib.dhp</groupId>
<artifactId>dhp-graph-mapper</artifactId>
<version>${project.version}</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-graph-provision</artifactId>-->

9
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/AbstractSparkAction.java
View File

@ -128,13 +128,12 @@ abstract class AbstractSparkAction implements Serializable {
.collect(Collectors.joining(SP_SEPARATOR));
}
protected static MapFunction<String, Relation> patchRelFn() {
protected static MapFunction<Relation, Relation> patchRelFn() {
return value -> {
final Relation rel = OBJECT_MAPPER.readValue(value, Relation.class);
if (rel.getDataInfo() == null) {
rel.setDataInfo(new DataInfo());
if (value.getDataInfo() == null) {
value.setDataInfo(new DataInfo());
}
return rel;
return value;
};
}

16
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCopyOpenorgsMergeRels.java
View File

@ -68,22 +68,20 @@ public class SparkCopyOpenorgsMergeRels extends AbstractSparkAction {
final String relationPath = DedupUtility.createEntityPath(graphBasePath, "relation");
// collect organization merge relations from openorgs database
JavaRDD<Relation> mergeRelsRDD = spark
Dataset<Relation> relations = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(this::isOpenorgs) // take only openorgs relations
.filter(this::isMergeRel); // take merges and isMergedIn relations
log.info("Number of Openorgs Merge Relations collected: {}", mergeRelsRDD.count());
final Dataset<Relation> relations = spark
.createDataset(
mergeRelsRDD.rdd(),
Encoders.bean(Relation.class));
relations.cache();
log.info("Number of Openorgs Merge Relations collected: {}", relations.count());
saveParquet(relations, outputPath, SaveMode.Append);
relations.unpersist();
}
private boolean isMergeRel(Relation rel) {

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCopyOpenorgsSimRels.java
View File

@ -69,7 +69,9 @@ public class SparkCopyOpenorgsSimRels extends AbstractSparkAction {
Dataset<Relation> rawRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.filter(this::filterOpenorgsRels);

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCopyRelationsNoOpenorgs.java
View File

@ -58,7 +58,9 @@ public class SparkCopyRelationsNoOpenorgs extends AbstractSparkAction {
JavaRDD<Relation> simRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(x -> !isOpenorgsDedupRel(x));

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPrepareNewOrgs.java
View File

@ -111,7 +111,9 @@ public class SparkPrepareNewOrgs extends AbstractSparkAction {
// collect diffrels from the raw graph relations: <other id, "diffRel">
JavaPairRDD<String, String> diffRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(r -> filterRels(r, ModelSupport.getMainType(EntityType.organization)))

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPrepareOrgRels.java
View File

@ -133,7 +133,9 @@ public class SparkPrepareOrgRels extends AbstractSparkAction {
// collect diffrels from the raw graph relations: <<best id, other id>, "diffRel">
JavaRDD<Tuple2<Tuple2<String, String>, String>> diffRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(r -> filterRels(r, "organization"))

3
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkUpdateEntity.java
View File

@ -2,6 +2,7 @@
package eu.dnetlib.dhp.oa.dedup;
import java.io.IOException;
import java.net.URI;
import java.util.Map;
import org.apache.commons.io.IOUtils;
@ -132,7 +133,7 @@ public class SparkUpdateEntity extends AbstractSparkAction {
boolean result = false;
FileSystem fileSystem = FileSystem.get(new Configuration());
FileSystem fileSystem = FileSystem.get(URI.create(basePath), new Configuration());
FileStatus[] fileStatuses = fileSystem.listStatus(new Path(basePath));
for (FileStatus fs : fileStatuses) {

7
dhp-workflows/pom.xml
View File

@ -40,7 +40,6 @@
<module>dhp-usage-stats-build</module>
<module>dhp-usage-raw-data-update</module>
<module>dhp-broker-events</module>
<module>dhp-doiboost</module>
<module>dhp-impact-indicators</module>
<module>dhp-swh</module>
<module>dhp-incremental-graph</module>
@ -95,6 +94,12 @@
</dependency>
</dependencies>
<profiles>
<profile>
<id>spark-24</id>
<modules>
<module>dhp-doiboost</module>
</modules>
</profile>
<profile>
<!-- This profile sets properties that are required for test oozie workflows To be used only with 'oozie-package' profile -->
<id>attach-test-resources</id>

17
pom.xml
View File

@ -19,6 +19,7 @@
</licenses>
<modules>
<module>dhp-rmi-api</module>
<module>dhp-build</module>
<module>dhp-pace-core</module>
<module>dhp-common</module>
@ -440,12 +441,6 @@
<scope>provided</scope>
</dependency>
<dependency>
<groupId>eu.dnetlib</groupId>
<artifactId>cnr-rmi-api</artifactId>
<version>${cnr-rmi-api.version}</version>
</dependency>
<dependency>
<groupId>eu.dnetlib.dhp</groupId>
<artifactId>dnet-openaire-broker-common</artifactId>
@ -926,7 +921,6 @@
<!-- dependency versions -->
<apache.poi.version>4.1.2</apache.poi.version>
<cnr-rmi-api.version>[2.6.1]</cnr-rmi-api.version>
<common.compress.version>1.20</common.compress.version>
<common.csv.version>1.8</common.csv.version>
<common.text.version>1.8</common.text.version>
@ -936,6 +930,7 @@
<commons-io.version>2.4</commons-io.version>
<commons.logging.version>1.1.3</commons.logging.version>
<commons-validator.version>1.7</commons-validator.version>
<cxf.version>3.5.9</cxf.version>
<dateparser.version>1.0.7</dateparser.version>
<dhp-schemas.version>[10.0.0]</dhp-schemas.version>
<dhp.cdh.version>cdh5.9.2</dhp.cdh.version>
@ -953,6 +948,7 @@
<json4s.version>3.5.3</json4s.version>
<jsonschemagenerator.version>4.13.0</jsonschemagenerator.version>
<junit-jupiter.version>5.6.1</junit-jupiter.version>
<metro.version>3.0.3</metro.version>
<mockito-core.version>3.3.3</mockito-core.version>
<mongodb.driver.version>3.4.2</mongodb.driver.version>
<okhttp.version>4.7.2</okhttp.version>
@ -1036,6 +1032,7 @@
<net.alchim31.maven.version>4.8.1</net.alchim31.maven.version>
<!-- dependencies -->
<dhp.hadoop.version>3.3.4</dhp.hadoop.version>
<common.compress.version>1.23.0</common.compress.version>
<common.csv.version>1.8</common.csv.version>
<common.text.version>1.10.0</common.text.version>
@ -1049,7 +1046,7 @@
<dhp.guava.version>14.0.1</dhp.guava.version>
<solr.version>8.11.0</solr.version>
<sparksolr.version>4.0.4</sparksolr.version>
<dhp.spark.version>3.5.1.openaire-SNAPSHOT</dhp.spark.version>
<dhp.spark.version>3.5.3</dhp.spark.version>
<dhp.jackson.version>2.15.2</dhp.jackson.version>
<dhp.commons.lang.version>3.12.0</dhp.commons.lang.version>
<log4j.version>2.20.0</log4j.version>
@ -1067,6 +1064,10 @@
<activation>
<jdk>[11</jdk>
</activation>
<properties>
<cxf.version>4.0.5</cxf.version>
<metro.version>4.0.4</metro.version>
</properties>
<build>
<pluginManagement>
<plugins>