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

Use json loading in place of text loading + jackson mapper

This commit is contained in:
Giambattista Bloisi 2024-12-20 15:24:35 +01:00
parent 1a3fc87599
commit 79e4728afd
7 changed files with 28 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dhp-shade-package/pom.xml
View File

@ -45,11 +45,11 @@
<!-- <artifactId>dhp-broker-events</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-dedup-openaire</artifactId>-->
<!-- <version>${project.version}</version>-->
<!-- </dependency>-->
<dependency>
<groupId>eu.dnetlib.dhp</groupId>
<artifactId>dhp-dedup-openaire</artifactId>
<version>${project.version}</version>
</dependency>
<!-- <dependency>-->
<!-- <groupId>eu.dnetlib.dhp</groupId>-->
<!-- <artifactId>dhp-enrichment</artifactId>-->

9
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/AbstractSparkAction.java
View File

@ -128,13 +128,12 @@ abstract class AbstractSparkAction implements Serializable {
.collect(Collectors.joining(SP_SEPARATOR));
}
protected static MapFunction<String, Relation> patchRelFn() {
protected static MapFunction<Relation, Relation> patchRelFn() {
return value -> {
final Relation rel = OBJECT_MAPPER.readValue(value, Relation.class);
if (rel.getDataInfo() == null) {
rel.setDataInfo(new DataInfo());
if (value.getDataInfo() == null) {
value.setDataInfo(new DataInfo());
}
return rel;
return value;
};
}

16
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCopyOpenorgsMergeRels.java
View File

@ -68,22 +68,20 @@ public class SparkCopyOpenorgsMergeRels extends AbstractSparkAction {
final String relationPath = DedupUtility.createEntityPath(graphBasePath, "relation");
// collect organization merge relations from openorgs database
JavaRDD<Relation> mergeRelsRDD = spark
Dataset<Relation> relations = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(this::isOpenorgs) // take only openorgs relations
.filter(this::isMergeRel); // take merges and isMergedIn relations
log.info("Number of Openorgs Merge Relations collected: {}", mergeRelsRDD.count());
final Dataset<Relation> relations = spark
.createDataset(
mergeRelsRDD.rdd(),
Encoders.bean(Relation.class));
relations.cache();
log.info("Number of Openorgs Merge Relations collected: {}", relations.count());
saveParquet(relations, outputPath, SaveMode.Append);
relations.unpersist();
}
private boolean isMergeRel(Relation rel) {

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCopyOpenorgsSimRels.java
View File

@ -69,7 +69,9 @@ public class SparkCopyOpenorgsSimRels extends AbstractSparkAction {
Dataset<Relation> rawRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.filter(this::filterOpenorgsRels);

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCopyRelationsNoOpenorgs.java
View File

@ -58,7 +58,9 @@ public class SparkCopyRelationsNoOpenorgs extends AbstractSparkAction {
JavaRDD<Relation> simRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(x -> !isOpenorgsDedupRel(x));

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPrepareNewOrgs.java
View File

@ -111,7 +111,9 @@ public class SparkPrepareNewOrgs extends AbstractSparkAction {
// collect diffrels from the raw graph relations: <other id, "diffRel">
JavaPairRDD<String, String> diffRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(r -> filterRels(r, ModelSupport.getMainType(EntityType.organization)))

4
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkPrepareOrgRels.java
View File

@ -133,7 +133,9 @@ public class SparkPrepareOrgRels extends AbstractSparkAction {
// collect diffrels from the raw graph relations: <<best id, other id>, "diffRel">
JavaRDD<Tuple2<Tuple2<String, String>, String>> diffRels = spark
.read()
.textFile(relationPath)
.schema(Encoders.bean(Relation.class).schema())
.json(relationPath)
.as(Encoders.bean(Relation.class))
.map(patchRelFn(), Encoders.bean(Relation.class))
.toJavaRDD()
.filter(r -> filterRels(r, "organization"))