diff --git a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/Country.java b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/Country.java new file mode 100644 index 0000000000..accd278d72 --- /dev/null +++ b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/Country.java @@ -0,0 +1,24 @@ +package eu.dnetlib.dhp.countrypropagation; + +import java.io.Serializable; + +public class Country implements Serializable { + private String classid; + private String classname; + + public String getClassid() { + return classid; + } + + public void setClassid(String classid) { + this.classid = classid; + } + + public String getClassname() { + return classname; + } + + public void setClassname(String classname) { + this.classname = classname; + } +} diff --git a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/DatasourceCountry.java b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/DatasourceCountry.java index 460764e166..3cce8f4bbc 100644 --- a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/DatasourceCountry.java +++ b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/DatasourceCountry.java @@ -4,7 +4,7 @@ import java.io.Serializable; public class DatasourceCountry implements Serializable { private String dataSourceId; - private String country; + private Country country; public String getDataSourceId() { return dataSourceId; @@ -14,11 +14,11 @@ public class DatasourceCountry implements Serializable { this.dataSourceId = dataSourceId; } - public String getCountry() { + public Country getCountry() { return country; } - public void setCountry(String country) { + public void setCountry(Country country) { this.country = country; } } diff --git a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java index b84c4afb54..887ddbc691 100644 --- a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java +++ b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/PrepareDatasourceCountryAssociation.java @@ -29,14 +29,14 @@ import org.slf4j.LoggerFactory; * relation provides between Datasource and Organization is exploited to get the country for the datasource */ -public class PrepareResultCountryAssociation { - private static final Logger log = LoggerFactory.getLogger(PrepareResultCountryAssociation.class); +public class PrepareDatasourceCountryAssociation { + private static final Logger log = LoggerFactory.getLogger(PrepareDatasourceCountryAssociation.class); private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); public static void main(String[] args) throws Exception { - String jsonConfiguration = IOUtils.toString(PrepareResultCountryAssociation.class + String jsonConfiguration = IOUtils.toString(PrepareDatasourceCountryAssociation.class .getResourceAsStream("/eu/dnetlib/dhp/countrypropagation/input_prepare_dc_assoc.json")); final ArgumentApplicationParser parser = new ArgumentApplicationParser( @@ -97,7 +97,7 @@ public class PrepareResultCountryAssociation { relation.createOrReplaceTempView("relation"); organization.createOrReplaceTempView("organization"); - String query = "SELECT source ds, country.classid country " + + String query = "SELECT source ds, named_struct('classid', country.classid, 'classname', country.classname) country " + "FROM ( SELECT id " + " FROM datasource " + " WHERE (datainfo.deletedbyinference = false " + whitelisted + ") " + diff --git a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/ResultCountrySet.java b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/ResultCountrySet.java new file mode 100644 index 0000000000..4978d91821 --- /dev/null +++ b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/ResultCountrySet.java @@ -0,0 +1,25 @@ +package eu.dnetlib.dhp.countrypropagation; + +import java.io.Serializable; +import java.util.Set; + +public class ResultCountrySet implements Serializable { + private String resultId; + private Set countrySet; + + public String getResultId() { + return resultId; + } + + public void setResultId(String resultId) { + this.resultId = resultId; + } + + public Set getCountrySet() { + return countrySet; + } + + public void setCountrySet(Set countrySet) { + this.countrySet = countrySet; + } +} diff --git a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java index 2ebd818cb7..cf80649b60 100644 --- a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java +++ b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java @@ -100,7 +100,7 @@ public class SparkCountryPropagationJob { datasource.createOrReplaceTempView("datasource"); relation.createOrReplaceTempView("relation"); organization.createOrReplaceTempView("organization"); - +//todo add length(country.classid)>0 String query = "SELECT source ds, target org, country.classid country " + "FROM ( SELECT id " + "FROM datasource " + @@ -118,6 +118,7 @@ public class SparkCountryPropagationJob { "WHERE datainfo.deletedbyinference = false ) o " + "ON o.id = rel.target"; + //todo broadcast Dataset rels = spark.sql(query); rels.createOrReplaceTempView("rels"); @@ -268,11 +269,11 @@ public class SparkCountryPropagationJob { }); } - private static JavaRDD propagateOnResult(SparkSession spark, String table) { + private static JavaRDD propagateOnResult(SparkSession spark, String result_type) { String query; query = "SELECT id, inst.collectedfrom.key cf , inst.hostedby.key hb " + "FROM ( SELECT id, instance " + - "FROM " + table + + "FROM " + result_type + " WHERE datainfo.deletedbyinference = false) ds " + "LATERAL VIEW EXPLODE(instance) i AS inst"; Dataset cfhb = spark.sql(query); diff --git a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob2.java b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob2.java index 210d42f9ca..967c940b50 100644 --- a/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob2.java +++ b/dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob2.java @@ -5,7 +5,6 @@ import eu.dnetlib.dhp.application.ArgumentApplicationParser; import eu.dnetlib.dhp.schema.oaf.*; import org.apache.commons.io.IOUtils; import org.apache.spark.SparkConf; -import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.MapFunction; @@ -41,7 +40,7 @@ public class SparkCountryPropagationJob2 { final String outputPath = "/tmp/provision/propagation/countrytoresultfrominstitutionalrepositories"; final String datasourcecountrypath = outputPath + "/prepared_datasource_country"; - final String resultClassName = parser.get("resultClazz"); + final String resultClassName = parser.get("resultTableName"); final String resultType = resultClassName.substring(resultClassName.lastIndexOf(".")+1); @@ -67,7 +66,8 @@ public class SparkCountryPropagationJob2 { //broadcasting the result of the preparation step Broadcast> broadcast_datasourcecountryassoc = sc.broadcast(datasourcecountryassoc); - Dataset potentialUpdates = getPotentialResultToUpdate(spark, inputPath, resultClazz, broadcast_datasourcecountryassoc); + Dataset potentialUpdates = getPotentialResultToUpdate(spark, inputPath, resultClazz, + broadcast_datasourcecountryassoc).as(Encoders.bean(ResultCountrySet.class)); if(writeUpdates){ writeUpdates(potentialUpdates.toJavaRDD(), outputPath + "/update_" + resultType); @@ -80,7 +80,7 @@ public class SparkCountryPropagationJob2 { } - private static void updateResultTable(SparkSession spark, Dataset potentialUpdates, + private static void updateResultTable(SparkSession spark, Dataset potentialUpdates, String inputPath, Class resultClazz, String outputPath) { @@ -91,23 +91,24 @@ public class SparkCountryPropagationJob2 { .map(r -> new Tuple2<>(r.getId(), r), Encoders.tuple(Encoders.STRING(), Encoders.bean(resultClazz))); - Dataset> potential_update_pair = potentialUpdates.map(pu -> new Tuple2<>(pu.getString(0), pu.getList(1)), - Encoders.tuple(Encoders.STRING(), Encoders.bean(List.class))); + Dataset> potential_update_pair = potentialUpdates.map(pu -> new Tuple2<>(pu.getResultId(), + pu), + Encoders.tuple(Encoders.STRING(), Encoders.bean(ResultCountrySet.class))); Dataset new_table = result_pair .joinWith(potential_update_pair, result_pair.col("_1").equalTo(potential_update_pair.col("_1")), "left") - .map((MapFunction, Tuple2>, R>) value -> { + .map((MapFunction, Tuple2>, R>) value -> { R r = value._1()._2(); - Optional> potentialNewCountries = Optional.ofNullable(value._2()).map(Tuple2::_2); - if (potentialNewCountries != null) { + Optional potentialNewCountries = Optional.ofNullable(value._2()).map(Tuple2::_2); + if (potentialNewCountries.isPresent()) { HashSet countries = new HashSet<>(); for (Qualifier country : r.getCountry()) { countries.add(country.getClassid()); } - for (Object country : potentialNewCountries.get()) { - if (!countries.contains(country)) { - r.getCountry().add(getCountry((String) country)); + for (Country country : potentialNewCountries.get().getCountrySet()) { + if (!countries.contains(country.getClassid())) { + r.getCountry().add(getCountry(country.getClassid(),country.getClassname())); } } } @@ -117,7 +118,7 @@ public class SparkCountryPropagationJob2 { log.info("Saving graph table to path: {}", outputPath); - result + new_table .toJSON() .write() .option("compression", "gzip") @@ -151,6 +152,7 @@ public class SparkCountryPropagationJob2 { private static Dataset countryPropagationAssoc(SparkSession spark, Broadcast> broadcast_datasourcecountryassoc){ + Dataset datasource_country = broadcast_datasourcecountryassoc.value(); datasource_country.createOrReplaceTempView("datasource_country"); @@ -183,7 +185,7 @@ public class SparkCountryPropagationJob2 { .as(Encoders.bean(DatasourceCountry.class)); } - private static void writeUpdates(JavaRDD potentialUpdates, String outputPath){ + private static void writeUpdates(JavaRDD potentialUpdates, String outputPath){ potentialUpdates.map(u -> OBJECT_MAPPER.writeValueAsString(u)) .saveAsTextFile(outputPath); }