From 7a3fd9f54cc7cf803e799fc430b0eb898e33f3c6 Mon Sep 17 00:00:00 2001
From: Claudio Atzori <claudio.atzori@isti.cnr.it>
Date: Mon, 13 Jul 2020 10:11:36 +0200
Subject: [PATCH] dedup relation aggregator moved into dedicated class

---
 .../dhp/oa/dedup/RelationAggregator.java      | 46 +++++++++++++++++++
 .../dhp/oa/dedup/SparkPropagateRelation.java  | 36 +--------------
 2 files changed, 47 insertions(+), 35 deletions(-)
 create mode 100644 dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/RelationAggregator.java

diff --git a/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/RelationAggregator.java b/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/RelationAggregator.java
new file mode 100644
index 000000000..0a29aa51b
--- /dev/null
+++ b/dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/RelationAggregator.java
@@ -0,0 +1,46 @@
+
+package eu.dnetlib.dhp.oa.dedup;
+
+import java.util.Objects;
+
+import org.apache.spark.sql.Encoder;
+import org.apache.spark.sql.Encoders;
+import org.apache.spark.sql.expressions.Aggregator;
+
+import eu.dnetlib.dhp.schema.oaf.Relation;
+
+public class RelationAggregator extends Aggregator<Relation, Relation, Relation> {
+
+	private static Relation ZERO = new Relation();
+
+	@Override
+	public Relation zero() {
+		return ZERO;
+	}
+
+	@Override
+	public Relation reduce(Relation b, Relation a) {
+		return Objects.equals(a, ZERO) ? b : a;
+	}
+
+	@Override
+	public Relation merge(Relation b, Relation a) {
+		b.mergeFrom(a);
+		return b;
+	}
+
+	@Override
+	public Relation finish(Relation r) {
+		return r;
+	}
+
+	@Override
+	public Encoder<Relation> bufferEncoder() {
+		return Encoders.bean(Relation.class);
+	}
+
+	@Override
+	public Encoder<Relation> outputEncoder() {
+		return Encoders.bean(Relation.class);
+	}
+}
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 13d2e4cd7..c19769749 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
@@ -7,7 +7,6 @@ import org.apache.commons.io.IOUtils;
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.function.MapFunction;
 import org.apache.spark.sql.*;
-import org.apache.spark.sql.expressions.Aggregator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -96,7 +95,7 @@ public class SparkPropagateRelation extends AbstractSparkAction {
 			FieldType.TARGET,
 			getDeletedFn());
 
-		save(newRels.union(updated).union(mergeRels).distinct(), outputRelationPath, SaveMode.Overwrite);
+		save(distinctRelations(newRels.union(updated).union(mergeRels)), outputRelationPath, SaveMode.Overwrite);
 	}
 
 	private Dataset<Relation> distinctRelations(Dataset<Relation> rels) {
@@ -106,39 +105,6 @@ public class SparkPropagateRelation extends AbstractSparkAction {
 			.map((MapFunction<Tuple2<String, Relation>, Relation>) t -> t._2(), Encoders.bean(Relation.class));
 	}
 
-	class RelationAggregator extends Aggregator<Relation, Relation, Relation> {
-
-		@Override
-		public Relation zero() {
-			return new Relation();
-		}
-
-		@Override
-		public Relation reduce(Relation b, Relation a) {
-			return b;
-		}
-
-		@Override
-		public Relation merge(Relation b, Relation a) {
-			return b;
-		}
-
-		@Override
-		public Relation finish(Relation r) {
-			return r;
-		}
-
-		@Override
-		public Encoder<Relation> bufferEncoder() {
-			return Encoders.bean(Relation.class);
-		}
-
-		@Override
-		public Encoder<Relation> outputEncoder() {
-			return Encoders.bean(Relation.class);
-		}
-	}
-
 	private static Dataset<Relation> processDataset(
 		Dataset<Relation> rels,
 		Dataset<Tuple2<String, String>> mergedIds,