From c8284bab06c13f47d135cbd7d8a8fe009f090b6b Mon Sep 17 00:00:00 2001
From: Claudio Atzori <claudio.atzori@isti.cnr.it>
Date: Mon, 13 Jul 2020 15:54:51 +0200
Subject: [PATCH] WIP SparkCreateMergeRels distinct relations

---
 .../main/java/eu/dnetlib/dhp/oa/dedup/RelationAggregator.java | 4 ++--
 .../java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java  | 4 ++--
 .../src/test/java/eu/dnetlib/dhp/oa/dedup/SparkDedupTest.java | 4 +++-
 3 files changed, 7 insertions(+), 5 deletions(-)

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
index 0a29aa51b2..7935fe1ca2 100644
--- 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
@@ -36,11 +36,11 @@ public class RelationAggregator extends Aggregator<Relation, Relation, Relation>
 
 	@Override
 	public Encoder<Relation> bufferEncoder() {
-		return Encoders.bean(Relation.class);
+		return Encoders.kryo(Relation.class);
 	}
 
 	@Override
 	public Encoder<Relation> outputEncoder() {
-		return Encoders.bean(Relation.class);
+		return Encoders.kryo(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 88fe5b26d1..baba3bc875 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
@@ -98,10 +98,10 @@ public class SparkPropagateRelation extends AbstractSparkAction {
 			getDeletedFn());
 
 		save(
-			newRels
+			distinctRelations(newRels
 				.union(updated)
 				.union(mergeRels)
-				.map((MapFunction<Relation, Relation>) r -> r, Encoders.kryo(Relation.class)),
+				.map((MapFunction<Relation, Relation>) r -> r, Encoders.kryo(Relation.class))),
 			outputRelationPath, SaveMode.Overwrite);
 	}
 
diff --git a/dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/SparkDedupTest.java b/dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/SparkDedupTest.java
index e106551260..82c2d82b72 100644
--- a/dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/SparkDedupTest.java
+++ b/dhp-workflows/dhp-dedup-openaire/src/test/java/eu/dnetlib/dhp/oa/dedup/SparkDedupTest.java
@@ -77,11 +77,13 @@ public class SparkDedupTest implements Serializable {
 		FileUtils.deleteDirectory(new File(testOutputBasePath));
 		FileUtils.deleteDirectory(new File(testDedupGraphBasePath));
 
+		final SparkConf conf = new SparkConf();
+		conf.set("spark.sql.shuffle.partitions", "200");
 		spark = SparkSession
 			.builder()
 			.appName(SparkDedupTest.class.getSimpleName())
 			.master("local[*]")
-			.config(new SparkConf())
+			.config(conf)
 			.getOrCreate();
 
 		jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());