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

added verification tha one result linked to more than on project is saved just onece

This commit is contained in:
Miriam Baglioni 2021-05-20 10:40:09 +02:00
parent b8f25fbf39
commit 85bd6ab5d0
1 changed files with 8 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/dump/funderresults/SparkResultLinkedToProject.java
View File

@ -98,7 +98,13 @@ public class SparkResultLinkedToProject implements Serializable {
"join project p " + "join project p " +
"on rel.target = p.id " + "on rel.target = p.id " +
"") "")
.as(Encoders.bean(inputClazz)); .as(Encoders.bean(inputClazz))
;
tmp.groupByKey(
(MapFunction< R, String>) value -> value
.getId(),
Encoders.STRING())
.mapGroups((MapGroupsFunction<String, R, R>) (k, it) -> it.next(), Encoders.bean(inputClazz))
// //
// relations // relations
@ -113,7 +119,7 @@ public class SparkResultLinkedToProject implements Serializable {
// .mapGroups((MapGroupsFunction<String, Tuple2<Relation, R>, R>) (k, it) -> { // .mapGroups((MapGroupsFunction<String, Tuple2<Relation, R>, R>) (k, it) -> {
// return it.next()._2(); // return it.next()._2();
// }, Encoders.bean(inputClazz)) // }, Encoders.bean(inputClazz))
tmp //tmp
.write() .write()
.mode(SaveMode.Overwrite) .mode(SaveMode.Overwrite)
.option("compression", "gzip") .option("compression", "gzip")