From 3319440c5348ec03399f7b4e77c7e165ee3de882 Mon Sep 17 00:00:00 2001
From: "miriam.baglioni" <miriam.baglioni@isti.cnr.it>
Date: Tue, 24 Nov 2020 14:41:09 +0100
Subject: [PATCH] changed the direction of the relation between projects and
 result considered to select the results linked to projects

---
 .../graph/dump/community/SparkPrepareResultProject.java   | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/dump/community/SparkPrepareResultProject.java b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/dump/community/SparkPrepareResultProject.java
index 612af6d16..2d43888b4 100644
--- a/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/dump/community/SparkPrepareResultProject.java
+++ b/dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/dump/community/SparkPrepareResultProject.java
@@ -76,16 +76,16 @@ public class SparkPrepareResultProject implements Serializable {
 	private static void prepareResultProjectList(SparkSession spark, String inputPath, String outputPath) {
 		Dataset<Relation> relation = Utils
 			.readPath(spark, inputPath + "/relation", Relation.class)
-			.filter("dataInfo.deletedbyinference = false and relClass = 'produces'");
+			.filter("dataInfo.deletedbyinference = false and lower(relClass) = 'isproducedby'");
 		Dataset<eu.dnetlib.dhp.schema.oaf.Project> projects = Utils
 			.readPath(spark, inputPath + "/project", eu.dnetlib.dhp.schema.oaf.Project.class);
 
 		projects
-			.joinWith(relation, projects.col("id").equalTo(relation.col("source")))
+			.joinWith(relation, projects.col("id").equalTo(relation.col("target")), "inner")
 			.groupByKey(
 				(MapFunction<Tuple2<eu.dnetlib.dhp.schema.oaf.Project, Relation>, String>) value -> value
 					._2()
-					.getTarget(),
+					.getSource(),
 				Encoders.STRING())
 			.mapGroups(
 				(MapGroupsFunction<String, Tuple2<eu.dnetlib.dhp.schema.oaf.Project, Relation>, ResultProject>) (s,
@@ -93,7 +93,7 @@ public class SparkPrepareResultProject implements Serializable {
 					Set<String> projectSet = new HashSet<>();
 					Tuple2<eu.dnetlib.dhp.schema.oaf.Project, Relation> first = it.next();
 					ResultProject rp = new ResultProject();
-					rp.setResultId(first._2().getTarget());
+					rp.setResultId(s);
 					eu.dnetlib.dhp.schema.oaf.Project p = first._1();
 					projectSet.add(p.getId());
 					Project ps = getProject(p);