package eu.dnetlib.dhp.oa.graph.merge; import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession; import java.util.*; import java.util.stream.Collectors; import javax.xml.crypto.Data; import org.apache.commons.io.IOUtils; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.FilterFunction; import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.SaveMode; import org.apache.spark.sql.SparkSession; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.fasterxml.jackson.databind.ObjectMapper; import eu.dnetlib.dhp.application.ArgumentApplicationParser; import eu.dnetlib.dhp.common.HdfsSupport; import eu.dnetlib.dhp.oa.graph.clean.CleanGraphSparkJob; import eu.dnetlib.dhp.schema.common.ModelSupport; import eu.dnetlib.dhp.schema.oaf.*; import scala.Tuple2; /** * Combines the content from two aggregator graph tables of the same type, entities (or relationships) with the same ids * are picked preferring those from the BETA aggregator rather then from PROD. The identity of a relationship is defined * by eu.dnetlib.dhp.schema.common.ModelSupport#idFn() */ public class MergeGraphTableSparkJob { private static final Logger log = LoggerFactory.getLogger(MergeGraphTableSparkJob.class); private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private static final String PRIORITY_DEFAULT = "BETA"; // BETA | PROD private static final Datasource DATASOURCE = new Datasource(); static { Qualifier compatibility = new Qualifier(); compatibility.setClassid("UNKNOWN"); DATASOURCE.setOpenairecompatibility(compatibility); } public static void main(String[] args) throws Exception { String jsonConfiguration = IOUtils .toString( CleanGraphSparkJob.class .getResourceAsStream( "/eu/dnetlib/dhp/oa/graph/merge_graphs_parameters.json")); final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration); parser.parseArgument(args); String priority = Optional .ofNullable(parser.get("priority")) .orElse(PRIORITY_DEFAULT); log.info("priority: {}", priority); Boolean isSparkSessionManaged = Optional .ofNullable(parser.get("isSparkSessionManaged")) .map(Boolean::valueOf) .orElse(Boolean.TRUE); log.info("isSparkSessionManaged: {}", isSparkSessionManaged); String betaInputPath = parser.get("betaInputPath"); log.info("betaInputPath: {}", betaInputPath); String prodInputPath = parser.get("prodInputPath"); log.info("prodInputPath: {}", prodInputPath); String outputPath = parser.get("outputPath"); log.info("outputPath: {}", outputPath); String graphTableClassName = parser.get("graphTableClassName"); log.info("graphTableClassName: {}", graphTableClassName); Class extends OafEntity> entityClazz = (Class extends OafEntity>) Class.forName(graphTableClassName); SparkConf conf = new SparkConf(); conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); conf.registerKryoClasses(ModelSupport.getOafModelClasses()); runWithSparkSession( conf, isSparkSessionManaged, spark -> { removeOutputDir(spark, outputPath); mergeGraphTable(spark, priority, betaInputPath, prodInputPath, entityClazz, entityClazz, outputPath); }); } private static
void mergeGraphTable( SparkSession spark, String priority, String betaInputPath, String prodInputPath, Class
p_clazz,
Class b_clazz,
String outputPath) {
Dataset p = Optional.ofNullable(value._1()).map(Tuple2::_2);
Optional b = Optional.ofNullable(value._2()).map(Tuple2::_2);
if (p.orElse((P) b.orElse((B) DATASOURCE)) instanceof Datasource) {
return mergeDatasource(p, b);
}
switch (priority) {
default:
case "BETA":
return mergeWithPriorityToBETA(p, b);
case "PROD":
return mergeWithPriorityToPROD(p, b);
}
}, Encoders.bean(p_clazz))
.filter((FilterFunction ) Objects::nonNull)
.filter((FilterFunction ) o -> {
HashSet Datasource class type from PROD
* @param Datasource class type from BETA
* @return the datasource from PROD with the highest compatibility level.
*/
protected static P mergeDatasource(Optional p, Optional b) {
if (p.isPresent() & !b.isPresent()) {
return p.get();
}
if (b.isPresent() & !p.isPresent()) {
return (P) b.get();
}
if (!b.isPresent() & !p.isPresent()) {
return null; // unlikely, at least one should be produced by the join operation
}
Datasource dp = (Datasource) p.get();
Datasource db = (Datasource) b.get();
List P mergeWithPriorityToPROD(Optional p, Optional b) {
if (b.isPresent() & !p.isPresent()) {
return (P) b.get();
}
if (p.isPresent()) {
return p.get();
}
return null;
}
private static P mergeWithPriorityToBETA(Optional p, Optional b) {
if (p.isPresent() & !b.isPresent()) {
return p.get();
}
if (b.isPresent()) {
return (P) b.get();
}
return null;
}
private static