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 = Utils .readPath(spark, inputPath + "/relation", Relation.class) - .filter("dataInfo.deletedbyinference = false and relClass = 'produces'"); + .filter("dataInfo.deletedbyinference = false and lower(relClass) = 'isproducedby'"); Dataset 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, String>) value -> value ._2() - .getTarget(), + .getSource(), Encoders.STRING()) .mapGroups( (MapGroupsFunction, ResultProject>) (s, @@ -93,7 +93,7 @@ public class SparkPrepareResultProject implements Serializable { Set projectSet = new HashSet<>(); Tuple2 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);