2020-06-25 18:43:29 +02:00
|
|
|
|
|
|
|
package eu.dnetlib.doiboost.orcidnodoi;
|
|
|
|
|
2020-11-12 17:43:32 +01:00
|
|
|
import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
|
|
|
|
|
|
|
|
import java.io.IOException;
|
|
|
|
import java.util.Objects;
|
|
|
|
import java.util.Optional;
|
|
|
|
|
2020-06-25 18:43:29 +02:00
|
|
|
import org.apache.commons.io.IOUtils;
|
|
|
|
import org.apache.hadoop.io.Text;
|
2020-11-11 17:38:50 +01:00
|
|
|
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
|
2020-06-25 18:43:29 +02:00
|
|
|
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;
|
|
|
|
import org.apache.spark.sql.Dataset;
|
|
|
|
import org.apache.spark.sql.Encoders;
|
2020-07-30 13:38:33 +02:00
|
|
|
import org.apache.spark.util.LongAccumulator;
|
2020-06-25 18:43:29 +02:00
|
|
|
import org.slf4j.Logger;
|
|
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
|
2020-11-12 17:43:32 +01:00
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
|
import com.google.gson.Gson;
|
|
|
|
import com.google.gson.JsonElement;
|
|
|
|
import com.google.gson.JsonParser;
|
2020-06-25 18:43:29 +02:00
|
|
|
|
2020-11-12 17:43:32 +01:00
|
|
|
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
|
|
|
|
import eu.dnetlib.dhp.schema.action.AtomicAction;
|
|
|
|
import eu.dnetlib.dhp.schema.oaf.Publication;
|
|
|
|
import eu.dnetlib.dhp.schema.orcid.AuthorData;
|
|
|
|
import eu.dnetlib.doiboost.orcid.json.JsonHelper;
|
|
|
|
import eu.dnetlib.doiboost.orcidnodoi.model.WorkDataNoDoi;
|
|
|
|
import eu.dnetlib.doiboost.orcidnodoi.oaf.PublicationToOaf;
|
|
|
|
import eu.dnetlib.doiboost.orcidnodoi.similarity.AuthorMatcher;
|
|
|
|
import scala.Tuple2;
|
2020-06-25 18:43:29 +02:00
|
|
|
|
2020-09-15 11:32:49 +02:00
|
|
|
/**
|
|
|
|
* This spark job generates one parquet file, containing orcid publications dataset
|
|
|
|
*/
|
|
|
|
|
2020-06-25 18:43:29 +02:00
|
|
|
public class SparkGenEnrichedOrcidWorks {
|
|
|
|
|
2020-06-29 18:03:16 +02:00
|
|
|
static Logger logger = LoggerFactory.getLogger(SparkGenEnrichedOrcidWorks.class);
|
|
|
|
|
2020-11-11 17:38:50 +01:00
|
|
|
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
|
|
|
|
|
2020-06-25 18:43:29 +02:00
|
|
|
public static void main(String[] args) throws IOException, Exception {
|
|
|
|
|
|
|
|
final ArgumentApplicationParser parser = new ArgumentApplicationParser(
|
|
|
|
IOUtils
|
|
|
|
.toString(
|
|
|
|
SparkGenEnrichedOrcidWorks.class
|
|
|
|
.getResourceAsStream(
|
|
|
|
"/eu/dnetlib/dhp/doiboost/gen_enriched_orcid_works_parameters.json")));
|
|
|
|
parser.parseArgument(args);
|
|
|
|
Boolean isSparkSessionManaged = Optional
|
|
|
|
.ofNullable(parser.get("isSparkSessionManaged"))
|
|
|
|
.map(Boolean::valueOf)
|
|
|
|
.orElse(Boolean.TRUE);
|
|
|
|
final String workingPath = parser.get("workingPath");
|
|
|
|
final String outputEnrichedWorksPath = parser.get("outputEnrichedWorksPath");
|
|
|
|
final String outputWorksPath = parser.get("outputWorksPath");
|
2020-11-11 17:38:50 +01:00
|
|
|
final String hdfsServerUri = parser.get("hdfsServerUri");
|
2020-06-25 18:43:29 +02:00
|
|
|
|
|
|
|
SparkConf conf = new SparkConf();
|
|
|
|
runWithSparkSession(
|
|
|
|
conf,
|
|
|
|
isSparkSessionManaged,
|
|
|
|
spark -> {
|
|
|
|
JavaSparkContext sc = JavaSparkContext.fromSparkContext(spark.sparkContext());
|
|
|
|
|
|
|
|
JavaPairRDD<Text, Text> summariesRDD = sc
|
2020-11-06 13:47:50 +01:00
|
|
|
.sequenceFile(workingPath + "authors/authors.seq", Text.class, Text.class);
|
2020-06-25 18:43:29 +02:00
|
|
|
Dataset<AuthorData> summariesDataset = spark
|
|
|
|
.createDataset(
|
|
|
|
summariesRDD.map(seq -> loadAuthorFromJson(seq._1(), seq._2())).rdd(),
|
|
|
|
Encoders.bean(AuthorData.class));
|
2020-06-29 18:03:16 +02:00
|
|
|
logger.info("Authors data loaded: " + summariesDataset.count());
|
2020-06-25 18:43:29 +02:00
|
|
|
|
|
|
|
JavaPairRDD<Text, Text> activitiesRDD = sc
|
2020-06-29 18:03:16 +02:00
|
|
|
.sequenceFile(workingPath + outputWorksPath + "*.seq", Text.class, Text.class);
|
2020-06-25 18:43:29 +02:00
|
|
|
Dataset<WorkDataNoDoi> activitiesDataset = spark
|
|
|
|
.createDataset(
|
|
|
|
activitiesRDD.map(seq -> loadWorkFromJson(seq._1(), seq._2())).rdd(),
|
|
|
|
Encoders.bean(WorkDataNoDoi.class));
|
2020-06-29 18:03:16 +02:00
|
|
|
logger.info("Works data loaded: " + activitiesDataset.count());
|
2020-06-25 18:43:29 +02:00
|
|
|
|
2020-06-29 18:03:16 +02:00
|
|
|
JavaRDD<Tuple2<String, String>> enrichedWorksRDD = activitiesDataset
|
2020-06-26 17:27:34 +02:00
|
|
|
.joinWith(
|
|
|
|
summariesDataset,
|
|
|
|
activitiesDataset.col("oid").equalTo(summariesDataset.col("oid")), "inner")
|
|
|
|
.map(
|
2020-06-29 18:03:16 +02:00
|
|
|
(MapFunction<Tuple2<WorkDataNoDoi, AuthorData>, Tuple2<String, String>>) value -> {
|
2020-06-26 17:27:34 +02:00
|
|
|
WorkDataNoDoi w = value._1;
|
|
|
|
AuthorData a = value._2;
|
|
|
|
AuthorMatcher.match(a, w.getContributors());
|
2020-06-29 18:03:16 +02:00
|
|
|
return new Tuple2<>(a.getOid(), JsonHelper.createOidWork(w));
|
2020-06-26 17:27:34 +02:00
|
|
|
},
|
2020-06-29 18:03:16 +02:00
|
|
|
Encoders.tuple(Encoders.STRING(), Encoders.STRING()))
|
2020-06-26 17:27:34 +02:00
|
|
|
.filter(Objects::nonNull)
|
2020-06-29 18:03:16 +02:00
|
|
|
.toJavaRDD();
|
2020-11-06 13:47:50 +01:00
|
|
|
logger.info("Enriched works RDD ready.");
|
2020-07-30 13:38:33 +02:00
|
|
|
|
|
|
|
final LongAccumulator parsedPublications = spark.sparkContext().longAccumulator("parsedPublications");
|
|
|
|
final LongAccumulator enrichedPublications = spark
|
|
|
|
.sparkContext()
|
|
|
|
.longAccumulator("enrichedPublications");
|
|
|
|
final LongAccumulator errorsGeneric = spark.sparkContext().longAccumulator("errorsGeneric");
|
|
|
|
final LongAccumulator errorsInvalidTitle = spark.sparkContext().longAccumulator("errorsInvalidTitle");
|
|
|
|
final LongAccumulator errorsNotFoundAuthors = spark
|
|
|
|
.sparkContext()
|
|
|
|
.longAccumulator("errorsNotFoundAuthors");
|
|
|
|
final LongAccumulator errorsInvalidType = spark.sparkContext().longAccumulator("errorsInvalidType");
|
|
|
|
final PublicationToOaf publicationToOaf = new PublicationToOaf(
|
|
|
|
parsedPublications,
|
|
|
|
enrichedPublications,
|
|
|
|
errorsGeneric,
|
|
|
|
errorsInvalidTitle,
|
|
|
|
errorsNotFoundAuthors,
|
|
|
|
errorsInvalidType);
|
|
|
|
JavaRDD<Publication> oafPublicationRDD = enrichedWorksRDD
|
|
|
|
.map(
|
|
|
|
e -> {
|
|
|
|
return (Publication) publicationToOaf
|
|
|
|
.generatePublicationActionsFromJson(e._2());
|
|
|
|
})
|
|
|
|
.filter(p -> p != null);
|
2020-07-03 23:30:31 +02:00
|
|
|
|
2020-11-13 12:45:31 +01:00
|
|
|
sc.hadoopConfiguration().set("mapreduce.output.fileoutputformat.compress", "true");
|
|
|
|
|
2020-11-11 17:38:50 +01:00
|
|
|
oafPublicationRDD
|
|
|
|
.mapToPair(
|
|
|
|
p -> new Tuple2<>(p.getClass().toString(),
|
|
|
|
OBJECT_MAPPER.writeValueAsString(new AtomicAction<>(Publication.class, (Publication) p))))
|
|
|
|
.mapToPair(t -> new Tuple2(new Text(t._1()), new Text(t._2())))
|
|
|
|
.saveAsNewAPIHadoopFile(
|
|
|
|
workingPath.concat(outputEnrichedWorksPath),
|
|
|
|
Text.class,
|
|
|
|
Text.class,
|
|
|
|
SequenceFileOutputFormat.class,
|
|
|
|
sc.hadoopConfiguration());
|
2020-07-30 13:38:33 +02:00
|
|
|
|
|
|
|
logger.info("parsedPublications: " + parsedPublications.value().toString());
|
|
|
|
logger.info("enrichedPublications: " + enrichedPublications.value().toString());
|
|
|
|
logger.info("errorsGeneric: " + errorsGeneric.value().toString());
|
|
|
|
logger.info("errorsInvalidTitle: " + errorsInvalidTitle.value().toString());
|
|
|
|
logger.info("errorsNotFoundAuthors: " + errorsNotFoundAuthors.value().toString());
|
|
|
|
logger.info("errorsInvalidType: " + errorsInvalidType.value().toString());
|
2020-06-25 18:43:29 +02:00
|
|
|
});
|
|
|
|
}
|
|
|
|
|
|
|
|
private static AuthorData loadAuthorFromJson(Text orcidId, Text json) {
|
|
|
|
AuthorData authorData = new AuthorData();
|
|
|
|
authorData.setOid(orcidId.toString());
|
|
|
|
JsonElement jElement = new JsonParser().parse(json.toString());
|
|
|
|
authorData.setName(getJsonValue(jElement, "name"));
|
|
|
|
authorData.setSurname(getJsonValue(jElement, "surname"));
|
|
|
|
authorData.setCreditName(getJsonValue(jElement, "creditname"));
|
|
|
|
return authorData;
|
|
|
|
}
|
|
|
|
|
|
|
|
private static WorkDataNoDoi loadWorkFromJson(Text orcidId, Text json) {
|
2020-06-29 18:03:16 +02:00
|
|
|
|
2020-06-25 18:43:29 +02:00
|
|
|
WorkDataNoDoi workData = new Gson().fromJson(json.toString(), WorkDataNoDoi.class);
|
|
|
|
return workData;
|
|
|
|
}
|
|
|
|
|
|
|
|
private static String getJsonValue(JsonElement jElement, String property) {
|
|
|
|
if (jElement.getAsJsonObject().has(property)) {
|
|
|
|
JsonElement name = null;
|
|
|
|
name = jElement.getAsJsonObject().get(property);
|
|
|
|
if (name != null && !name.isJsonNull()) {
|
|
|
|
return name.getAsString();
|
|
|
|
}
|
|
|
|
}
|
2020-10-22 14:21:31 +02:00
|
|
|
return new String("");
|
2020-06-25 18:43:29 +02:00
|
|
|
}
|
|
|
|
}
|