michele.artini
/
dnet-hadoop
forked from D-Net/dnet-hadoop
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

read dataset from files instead of using hive tables

This commit is contained in:
Miriam Baglioni 2020-03-23 17:09:27 +01:00
parent a7bf037306
commit 9418e3d4fa
1 changed files with 69 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
dhp-workflows/dhp-propagation/src/main/java/eu/dnetlib/dhp/countrypropagation/SparkCountryPropagationJob.java
View File

@ -59,35 +59,87 @@ public class SparkCountryPropagationJob {
} }
private static void datasource(SparkSession spark, List<String> whitelist, String outputPath, String inputPath){ private static void datasource(SparkSession spark, List<String> whitelist, String outputPath, String inputPath){
String whitelisted = ""; String whitelisted = "";
for (String i : whitelist){ for (String i : whitelist){
whitelisted += " OR id = '" + i + "'"; whitelisted += " OR id = '" + i + "'";
} }
final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
Dataset<Datasource> datasource = spark.createDataset(sc.textFile(inputPath + "/datasource")
.map(item -> new ObjectMapper().readValue(item, Datasource.class)).rdd(), Encoders.bean(Datasource.class));
Dataset<Relation> relation = spark.createDataset(sc.textFile(inputPath + "/relation")
.map(item -> new ObjectMapper().readValue(item, Relation.class)).rdd(), Encoders.bean(Relation.class));
Dataset<Organization> organization = spark.createDataset(sc.textFile(inputPath + "/organization")
.map(item -> new ObjectMapper().readValue(item, Organization.class)).rdd(), Encoders.bean(Organization.class));
Dataset<eu.dnetlib.dhp.schema.oaf.Dataset> dataset = spark.createDataset(sc.textFile(inputPath + "/dataset")
.map(item -> new ObjectMapper().readValue(item, eu.dnetlib.dhp.schema.oaf.Dataset.class)).rdd(),
Encoders.bean(eu.dnetlib.dhp.schema.oaf.Dataset.class));
Dataset<OtherResearchProduct> other = spark.createDataset(sc.textFile(inputPath + "/otherresearchproduct")
.map(item -> new ObjectMapper().readValue(item, eu.dnetlib.dhp.schema.oaf.OtherResearchProduct.class)).rdd(),
Encoders.bean(eu.dnetlib.dhp.schema.oaf.OtherResearchProduct.class));
Dataset<Software> software = spark.createDataset(sc.textFile(inputPath + "/software")
.map(item -> new ObjectMapper().readValue(item, eu.dnetlib.dhp.schema.oaf.Software.class)).rdd(),
Encoders.bean(eu.dnetlib.dhp.schema.oaf.Software.class));
Dataset<Publication> publication = spark.createDataset(sc.textFile(inputPath + "/publication")
.map(item -> new ObjectMapper().readValue(item, eu.dnetlib.dhp.schema.oaf.Publication.class)).rdd(),
Encoders.bean(eu.dnetlib.dhp.schema.oaf.Publication.class));
datasource.createOrReplaceTempView("datasource");
relation.createOrReplaceTempView("relation");
organization.createOrReplaceTempView("organization");
// String query = "SELECT source ds, target org, country.classid country " +
// "FROM ( SELECT id " +
// "FROM openaire.datasource " +
// "WHERE datasourcetype.classid = 'pubsrepository::institutional' " +
// "AND (datainfo.deletedbyinference = false " + whitelisted + ") ) d " +
// "JOIN ( SELECT source, target " +
// "FROM openaire.relation " +
// "WHERE relclass = 'provides' " +
// "AND datainfo.deletedbyinference = false ) rel " +
// "ON d.id = rel.source " +
// "JOIN (SELECT id, country " +
// "FROM openaire.organization " +
// "WHERE datainfo.deletedbyinference = false ) o " +
// "ON o.id = rel.target";
String query = "SELECT source ds, target org, country.classid country " + String query = "SELECT source ds, target org, country.classid country " +
"FROM ( SELECT id " + "FROM ( SELECT id " +
"FROM openaire.datasource " + "FROM datasource " +
"WHERE datasourcetype.classid = 'pubsrepository::institutional' " + "WHERE datasourcetype.classid = 'pubsrepository::institutional' " +
"AND (datainfo.deletedbyinference = false " + whitelisted + ") ) d " + "AND (datainfo.deletedbyinference = false " + whitelisted + ") ) d " +
"JOIN ( SELECT source, target " + "JOIN ( SELECT source, target " +
"FROM openaire.relation " + "FROM relation " +
"WHERE relclass = 'provides' " + "WHERE relclass = 'provides' " +
"AND datainfo.deletedbyinference = false ) rel " + "AND datainfo.deletedbyinference = false ) rel " +
"ON d.id = rel.source " + "ON d.id = rel.source " +
"JOIN (SELECT id, country " + "JOIN (SELECT id, country " +
"FROM openaire.organization " + "FROM organization " +
"WHERE datainfo.deletedbyinference = false ) o " + "WHERE datainfo.deletedbyinference = false ) o " +
"ON o.id = rel.target"; "ON o.id = rel.target";
Dataset<Row> rels = spark.sql(query); Dataset<Row> rels = spark.sql(query);
rels.createOrReplaceTempView("rels"); rels.createOrReplaceTempView("rels");
software.createOrReplaceTempView("software");
final JavaRDD<Row> toupdateresultsoftware = propagateOnResult(spark, "software");
final JavaRDD<Row> toupdateresultsoftware = propagateOnResult(spark, "openaire.software"); dataset.createOrReplaceTempView("dataset");
final JavaRDD<Row> toupdateresultdataset = propagateOnResult(spark, "openaire.dataset"); final JavaRDD<Row> toupdateresultdataset = propagateOnResult(spark, "dataset");
final JavaRDD<Row> toupdateresultother = propagateOnResult(spark, "openaire.otherresearchproduct");
final JavaRDD<Row> toupdateresultpublication = propagateOnResult(spark, "openaire.publication"); other.createOrReplaceTempView("other");
final JavaRDD<Row> toupdateresultother = propagateOnResult(spark, "other");
publication.createOrReplaceTempView("publication");
final JavaRDD<Row> toupdateresultpublication = propagateOnResult(spark, "publication");
writeUpdates(toupdateresultsoftware, toupdateresultdataset, toupdateresultother, toupdateresultpublication, outputPath); writeUpdates(toupdateresultsoftware, toupdateresultdataset, toupdateresultother, toupdateresultpublication, outputPath);