antonis.lempesis
/
dnet-hadoop
forked from D-Net/dnet-hadoop
1
1
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity

Filter ids by dataInfo.deletedbyinference and DataInfo.invisible flags 

Filter relations also by dataInfo.invisible flag
This commit is contained in:
Giambattista Bloisi 2023-08-07 10:24:11 +02:00
parent af49424b59
commit 97b6d1dc45
1 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dhp-workflows/dhp-dedup-openaire/src/main/java/eu/dnetlib/dhp/oa/dedup/SparkCleanRelation.scala
View File

@ -46,22 +46,26 @@ class SparkCleanRelation(parser: ArgumentApplicationParser, spark: SparkSession)
val entities = val entities =
Seq("datasource", "project", "organization", "publication", "dataset", "software", "otherresearchproduct") Seq("datasource", "project", "organization", "publication", "dataset", "software", "otherresearchproduct")
val idsSchema = StructType.fromDDL("`id` STRING, `dataInfo` STRUCT<`deletedbyinference`:BOOLEAN,`invisible`:BOOLEAN>")
val emptyIds = spark.createDataFrame(spark.sparkContext.emptyRDD[Row].setName("empty"), val emptyIds = spark.createDataFrame(spark.sparkContext.emptyRDD[Row].setName("empty"),
new StructType().add(StructField("id", DataTypes.StringType, true))) idsSchema)
val ids = entities val ids = entities
.foldLeft(emptyIds)((ds, entity) => { .foldLeft(emptyIds)((ds, entity) => {
val entityPath = graphBasePath + '/' + entity val entityPath = graphBasePath + '/' + entity
if (HdfsSupport.exists(entityPath, spark.sparkContext.hadoopConfiguration)) { if (HdfsSupport.exists(entityPath, spark.sparkContext.hadoopConfiguration)) {
ds.union(spark.read.schema("`id` STRING").json(entityPath)) ds.union(spark.read.schema(idsSchema).json(entityPath))
} else { } else {
ds ds
} }
}) })
.filter("dataInfo.deletedbyinference != true AND dataInfo.invisible != true")
.select("id")
.distinct() .distinct()
val relations = spark.read.schema(Encoders.bean(classOf[Relation]).schema).json(inputPath) val relations = spark.read.schema(Encoders.bean(classOf[Relation]).schema).json(inputPath)
.filter(col("dataInfo.deletedbyinference").isNull || col("dataInfo.deletedbyinference") === false) .filter("dataInfo.deletedbyinference != true AND dataInfo.invisible != true")
AbstractSparkAction.save( AbstractSparkAction.save(
relations relations