andreas.czerniak
/
BrBETA_dnet-hadoop
forked from D-Net/dnet-hadoop
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

WIP SparkCreateMergeRels distinct relations

This commit is contained in:
Claudio Atzori 2020-07-13 16:13:36 +02:00
parent 8c67938ad0
commit 1143f426aa
2 changed files with 16 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/RelationAggregator.java
View File

@ -20,13 +20,12 @@ public class RelationAggregator extends Aggregator<Relation, Relation, Relation>
@Override
public Relation reduce(Relation b, Relation a) {
return Objects.equals(a, ZERO) ? b : a;
return mergeRel(b, a);
}
@Override
public Relation merge(Relation b, Relation a) {
b.mergeFrom(a);
return b;
return mergeRel(b, a);
}
@Override
@ -34,6 +33,18 @@ public class RelationAggregator extends Aggregator<Relation, Relation, Relation>
return r;
}
private Relation mergeRel(Relation b, Relation a) {
if (Objects.equals(b, ZERO)) {
return a;
}
if (Objects.equals(a, ZERO)) {
return b;
}
b.mergeFrom(a);
return b;
}
@Override
public Encoder<Relation> bufferEncoder() {
return Encoders.kryo(Relation.class);

3
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPropagateRelation.java
View File

@ -3,6 +3,7 @@ package eu.dnetlib.dhp.oa.dedup;
import static org.apache.spark.sql.functions.col;
import com.google.common.base.Joiner;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
@ -108,7 +109,7 @@ public class SparkPropagateRelation extends AbstractSparkAction {
private Dataset<Relation> distinctRelations(Dataset<Relation> rels) {
return rels
.filter(getRelationFilterFunction())
.groupByKey((MapFunction<Relation, String>) r -> ModelSupport.idFn().apply(r), Encoders.STRING())
.groupByKey((MapFunction<Relation, String>) r -> String.join(r.getSource(), r.getTarget(), r.getRelType(), r.getSubRelType(), r.getRelClass()), Encoders.STRING())
.agg(new RelationAggregator().toColumn())
.map((MapFunction<Tuple2<String, Relation>, Relation>) t -> t._2(), Encoders.bean(Relation.class));
}