From 770adc26e9d509837710de1d15bdeab88ed675a0 Mon Sep 17 00:00:00 2001
From: Claudio Atzori <claudio.atzori@isti.cnr.it>
Date: Fri, 10 Jul 2020 19:35:10 +0200
Subject: [PATCH] WIP aggregator to make relationships unique

---
 .../dnetlib/dhp/oa/dedup/SparkPropagateRelation.java   | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java b/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
index e65eb7ab5..13d2e4cd7 100644
--- a/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
+++ b/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
@@ -96,14 +96,14 @@ public class SparkPropagateRelation extends AbstractSparkAction {
 			FieldType.TARGET,
 			getDeletedFn());
 
-		save(distinctRelations(newRels.union(updated).union(mergeRels)), outputRelationPath, SaveMode.Overwrite);
+		save(newRels.union(updated).union(mergeRels).distinct(), outputRelationPath, SaveMode.Overwrite);
 	}
 
 	private Dataset<Relation> distinctRelations(Dataset<Relation> rels) {
 		return rels
-				.groupByKey((MapFunction<Relation, String>) r -> ModelSupport.idFn().apply(r), Encoders.STRING())
-				.agg(new RelationAggregator().toColumn())
-				.map((MapFunction<Tuple2<String, Relation>, Relation>) t -> t._2(), Encoders.bean(Relation.class));
+			.groupByKey((MapFunction<Relation, String>) r -> ModelSupport.idFn().apply(r), Encoders.STRING())
+			.agg(new RelationAggregator().toColumn())
+			.map((MapFunction<Tuple2<String, Relation>, Relation>) t -> t._2(), Encoders.bean(Relation.class));
 	}
 
 	class RelationAggregator extends Aggregator<Relation, Relation, Relation> {
@@ -115,13 +115,11 @@ public class SparkPropagateRelation extends AbstractSparkAction {
 
 		@Override
 		public Relation reduce(Relation b, Relation a) {
-			b.mergeFrom(a);
 			return b;
 		}
 
 		@Override
 		public Relation merge(Relation b, Relation a) {
-			b.mergeFrom(a);
 			return b;
 		}