Merge branch 'master' of https://code-repo.d4science.org/D-Net/dnet-hadoop

dhp-workflows/dhp-broker-events/README.md

@@ -0,0 +1,3 @@
+# dhp-broker-events
+dhp-broker-events is a DNET module responsible 
+of the production of events for the OpenAIRE Broker Service.

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

@@ -0,0 +1,66 @@
+<?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/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>dhp-workflows</artifactId>
+        <groupId>eu.dnetlib.dhp</groupId>
+        <version>1.1.7-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>dhp-broker-events</artifactId>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_2.11</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_2.11</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-hive_2.11</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>eu.dnetlib.dhp</groupId>
+            <artifactId>dhp-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>eu.dnetlib.dhp</groupId>
+            <artifactId>dhp-schemas</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.jayway.jsonpath</groupId>
+            <artifactId>json-path</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>dom4j</groupId>
+            <artifactId>dom4j</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>jaxen</groupId>
+            <artifactId>jaxen</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>eu.dnetlib</groupId>
+            <artifactId>dnet-openaire-broker-common</artifactId>
+            <version>[1.0.0,2.0.0)</version>
+        </dependency>
+
+    </dependencies>
+
+
+</project>

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

@@ -62,8 +62,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 	public static void main(final String[] args) throws Exception {
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
 				IOUtils
-				.toString(
+						.toString(MigrateDbEntitiesApplication.class
-					MigrateDbEntitiesApplication.class
 								.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/migrate_db_entities_parameters.json")));
 
 		parser.parseArgument(args);
@@ -135,10 +134,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 			ds.setId(createOpenaireId(10, rs.getString("datasourceid"), true));
 			ds.setOriginalId(Arrays.asList(rs.getString("datasourceid")));
 			ds
-				.setCollectedfrom(
+					.setCollectedfrom(listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname")));
-					listKeyValues(
-						createOpenaireId(10, rs.getString("collectedfromid"), true),
-						rs.getString("collectedfromname")));
 			ds.setPid(new ArrayList<>());
 			ds.setDateofcollection(asString(rs.getDate("dateofcollection")));
 			ds.setDateoftransformation(null); // Value not returned by the SQL query
@@ -199,10 +195,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 			p.setId(createOpenaireId(40, rs.getString("projectid"), true));
 			p.setOriginalId(Arrays.asList(rs.getString("projectid")));
 			p
-				.setCollectedfrom(
+					.setCollectedfrom(listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname")));
-					listKeyValues(
-						createOpenaireId(10, rs.getString("collectedfromid"), true),
-						rs.getString("collectedfromname")));
 			p.setPid(new ArrayList<>());
 			p.setDateofcollection(asString(rs.getDate("dateofcollection")));
 			p.setDateoftransformation(asString(rs.getDate("dateoftransformation")));
@@ -256,10 +249,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 			o.setId(createOpenaireId(20, rs.getString("organizationid"), true));
 			o.setOriginalId(Arrays.asList(rs.getString("organizationid")));
 			o
-				.setCollectedfrom(
+					.setCollectedfrom(listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname")));
-					listKeyValues(
-						createOpenaireId(10, rs.getString("collectedfromid"), true),
-						rs.getString("collectedfromname")));
 			o.setPid(new ArrayList<>());
 			o.setDateofcollection(asString(rs.getDate("dateofcollection")));
 			o.setDateoftransformation(asString(rs.getDate("dateoftransformation")));
@@ -277,11 +267,9 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 					.setEcresearchorganization(field(Boolean.toString(rs.getBoolean("ecresearchorganization")), info));
 			o.setEchighereducation(field(Boolean.toString(rs.getBoolean("echighereducation")), info));
 			o
-				.setEcinternationalorganizationeurinterests(
+					.setEcinternationalorganizationeurinterests(field(Boolean.toString(rs.getBoolean("ecinternationalorganizationeurinterests")), info));
-					field(Boolean.toString(rs.getBoolean("ecinternationalorganizationeurinterests")), info));
 			o
-				.setEcinternationalorganization(
+					.setEcinternationalorganization(field(Boolean.toString(rs.getBoolean("ecinternationalorganization")), info));
-					field(Boolean.toString(rs.getBoolean("ecinternationalorganization")), info));
 			o.setEcenterprise(field(Boolean.toString(rs.getBoolean("ecenterprise")), info));
 			o.setEcsmevalidated(field(Boolean.toString(rs.getBoolean("ecsmevalidated")), info));
 			o.setEcnutscode(field(Boolean.toString(rs.getBoolean("ecnutscode")), info));
@@ -300,8 +288,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 			final DataInfo info = prepareDataInfo(rs);
 			final String orgId = createOpenaireId(20, rs.getString("organization"), true);
 			final String dsId = createOpenaireId(10, rs.getString("datasource"), true);
-			final List<KeyValue> collectedFrom = listKeyValues(
+			final List<KeyValue> collectedFrom = listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname"));
-				createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname"));
 
 			final Relation r1 = new Relation();
 			r1.setRelType("datasourceOrganization");
@@ -334,8 +321,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 			final DataInfo info = prepareDataInfo(rs);
 			final String orgId = createOpenaireId(20, rs.getString("resporganization"), true);
 			final String projectId = createOpenaireId(40, rs.getString("project"), true);
-			final List<KeyValue> collectedFrom = listKeyValues(
+			final List<KeyValue> collectedFrom = listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname"));
-				createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname"));
 
 			final Relation r1 = new Relation();
 			r1.setRelType("projectOrganization");
@@ -365,12 +351,10 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 
 	public List<Oaf> processClaims(final ResultSet rs) {
 
-		final DataInfo info = dataInfo(
+		final DataInfo info =
-			false, null, false, false,
+				dataInfo(false, null, false, false, qualifier("user:claim", "user:claim", "dnet:provenanceActions", "dnet:provenanceActions"), "0.9");
-			qualifier("user:claim", "user:claim", "dnet:provenanceActions", "dnet:provenanceActions"), "0.9");
 
-		final List<KeyValue> collectedFrom = listKeyValues(
+		final List<KeyValue> collectedFrom = listKeyValues(createOpenaireId(10, "infrastruct_::openaire", true), "OpenAIRE");
-			createOpenaireId(10, "infrastruct_::openaire", true), "OpenAIRE");
 
 		try {
 
@@ -456,15 +440,11 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 		final String inferenceprovenance = rs.getString("inferenceprovenance");
 		final Boolean inferred = rs.getBoolean("inferred");
 		final String trust = rs.getString("trust");
-		return dataInfo(
+		return dataInfo(deletedbyinference, inferenceprovenance, inferred, false, MigrationConstants.ENTITYREGISTRY_PROVENANCE_ACTION, trust);
-			deletedbyinference, inferenceprovenance, inferred, false,
-			MigrationConstants.ENTITYREGISTRY_PROVENANCE_ACTION, trust);
 	}
 
 	private Qualifier prepareQualifierSplitting(final String s) {
-		if (StringUtils.isBlank(s)) {
+		if (StringUtils.isBlank(s)) { return null; }
-			return null;
-		}
 		final String[] arr = s.split("@@@");
 		return arr.length == 4 ? qualifier(arr[0], arr[1], arr[2], arr[3]) : null;
 	}
@@ -478,16 +458,12 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 	}
 
 	private StructuredProperty prepareStructProp(final String s, final DataInfo dataInfo) {
-		if (StringUtils.isBlank(s)) {
+		if (StringUtils.isBlank(s)) { return null; }
-			return null;
-		}
 		final String[] parts = s.split("###");
 		if (parts.length == 2) {
 			final String value = parts[0];
 			final String[] arr = parts[1].split("@@@");
-			if (arr.length == 4) {
+			if (arr.length == 4) { return structuredProperty(value, arr[0], arr[1], arr[2], arr[3], dataInfo); }
-				return structuredProperty(value, arr[0], arr[1], arr[2], arr[3], dataInfo);
-			}
 		}
 		return null;
 	}
@@ -514,9 +490,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
 			if (arr.length == 3) {
 				final String issn = StringUtils.isNotBlank(arr[0]) ? arr[0].trim() : null;
 				final String eissn = StringUtils.isNotBlank(arr[1]) ? arr[1].trim() : null;
-				;
 				final String lissn = StringUtils.isNotBlank(arr[2]) ? arr[2].trim() : null;
-				;
+
 				if (issn != null || eissn != null || lissn != null) {
 					return journal(name, issn, eissn, eissn, null, null, null, null, null, null, null, info);
 				}

dhp-workflows/pom.xml

@@ -26,6 +26,7 @@
         <module>dhp-dedup-scholexplorer</module>
         <module>dhp-graph-provision-scholexplorer</module>
         <module>dhp-stats-update</module>
+        <module>dhp-broker-events</module>
     </modules>
 
     <pluginRepositories>