refactoring

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/BipDeserialize.java

@@ -1,3 +1,4 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
 import java.io.Serializable;
@@ -5,19 +6,18 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 
+public class BipDeserialize extends HashMap<String, List<Score>> implements Serializable {
 
-public class BipDeserialize extends HashMap<String, List<Score>> implements Serializable  {
-
-    public BipDeserialize(){
-        super();
-    }
+	public BipDeserialize() {
+		super();
+	}
 
-    public List<Score> get(String key) {
+	public List<Score> get(String key) {
 
-        if (super.get(key) == null) {
-            return new ArrayList<>();
-        }
-        return super.get(key);
-    }
+		if (super.get(key) == null) {
+			return new ArrayList<>();
+		}
+		return super.get(key);
+	}
 
 }

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/BipScore.java

@@ -1,25 +1,26 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
 import java.io.Serializable;
 import java.util.List;
 
 public class BipScore implements Serializable {
-    private String id;
-    private List<Score> scoreList;
+	private String id;
+	private List<Score> scoreList;
 
-    public String getId() {
-        return id;
-    }
+	public String getId() {
+		return id;
+	}
 
-    public void setId(String id) {
-        this.id = id;
-    }
+	public void setId(String id) {
+		this.id = id;
+	}
 
-    public List<Score> getScoreList() {
-        return scoreList;
-    }
+	public List<Score> getScoreList() {
+		return scoreList;
+	}
 
-    public void setScoreList(List<Score> scoreList) {
-        this.scoreList = scoreList;
-    }
+	public void setScoreList(List<Score> scoreList) {
+		this.scoreList = scoreList;
+	}
 }

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/KeyValue.java

@@ -1,25 +1,26 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
 import java.io.Serializable;
 
 public class KeyValue implements Serializable {
 
-    private String key;
-    private String value;
+	private String key;
+	private String value;
 
-    public String getKey() {
-        return key;
-    }
+	public String getKey() {
+		return key;
+	}
 
-    public void setKey(String key) {
-        this.key = key;
-    }
+	public void setKey(String key) {
+		this.key = key;
+	}
 
-    public String getValue() {
-        return value;
-    }
+	public String getValue() {
+		return value;
+	}
 
-    public void setValue(String value) {
-        this.value = value;
-    }
+	public void setValue(String value) {
+		this.value = value;
+	}
 }

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/PreparedResult.java

@@ -1,24 +1,25 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
 import java.io.Serializable;
 
 public class PreparedResult implements Serializable {
-    private String id; //openaire id
-    private String value; //doi
+	private String id; // openaire id
+	private String value; // doi
 
-    public String getId() {
-        return id;
-    }
+	public String getId() {
+		return id;
+	}
 
-    public void setId(String id) {
-        this.id = id;
-    }
+	public void setId(String id) {
+		this.id = id;
+	}
 
-    public String getValue() {
-        return value;
-    }
+	public String getValue() {
+		return value;
+	}
 
-    public void setValue(String value) {
-        this.value = value;
-    }
+	public void setValue(String value) {
+		this.value = value;
+	}
 }

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/Score.java

@@ -1,3 +1,4 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
 import java.io.Serializable;
@@ -5,22 +6,22 @@ import java.util.List;
 
 public class Score implements Serializable {
 
-    private String id;
-    private List<KeyValue> unit;
+	private String id;
+	private List<KeyValue> unit;
 
-    public String getId() {
-        return id;
-    }
+	public String getId() {
+		return id;
+	}
 
-    public void setId(String id) {
-        this.id = id;
-    }
+	public void setId(String id) {
+		this.id = id;
+	}
 
-    public List<KeyValue> getUnit() {
-        return unit;
-    }
+	public List<KeyValue> getUnit() {
+		return unit;
+	}
 
-    public void setUnit(List<KeyValue> unit) {
-        this.unit = unit;
-    }
+	public void setUnit(List<KeyValue> unit) {
+		this.unit = unit;
+	}
 }

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/bipfinder/SparkAtomicActionScoreJob.java

@@ -1,11 +1,13 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import eu.dnetlib.dhp.application.ArgumentApplicationParser;
-import eu.dnetlib.dhp.common.HdfsSupport;
-import eu.dnetlib.dhp.schema.action.AtomicAction;
-import eu.dnetlib.dhp.schema.oaf.*;
-import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
 import org.apache.commons.io.IOUtils;
 import org.apache.hadoop.hdfs.qjournal.protocol.QJournalProtocolProtos;
 import org.apache.hadoop.io.Text;
@@ -22,182 +24,188 @@ import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.Tuple2;
 
-import java.io.Serializable;
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
-import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+import eu.dnetlib.dhp.common.HdfsSupport;
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.*;
+import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import scala.Tuple2;
 
 public class SparkAtomicActionScoreJob implements Serializable {
 
-    private static String DOI = "doi";
-    private static final Logger log = LoggerFactory.getLogger(SparkAtomicActionScoreJob.class);
-    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
-    public static <I extends Result> void main(String[] args) throws Exception {
-
-        String jsonConfiguration = IOUtils
-                .toString(
-                        SparkAtomicActionScoreJob.class
-                                .getResourceAsStream(
-                                        "/eu/dnetlib/dhp/actionmanager/bipfinder/input_parameters.json"));
-
-        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+	private static String DOI = "doi";
+	private static final Logger log = LoggerFactory.getLogger(SparkAtomicActionScoreJob.class);
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
-        parser.parseArgument(args);
-
-        Boolean isSparkSessionManaged = Optional
-                .ofNullable(parser.get("isSparkSessionManaged"))
-                .map(Boolean::valueOf)
-                .orElse(Boolean.TRUE);
-
-        log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
-
-        final String inputPath = parser.get("inputPath");
-        log.info("inputPath {}: ", inputPath);
-
-        final String outputPath = parser.get("outputPath");
-        log.info("outputPath {}: ", outputPath);
-
-        final String bipScorePath = parser.get("bipScorePath");
-        log.info("bipScorePath: {}", bipScorePath);
-
-        final String resultClassName = parser.get("resultTableName");
-        log.info("resultTableName: {}", resultClassName);
-
-        Class<I> inputClazz = (Class<I>) Class.forName(resultClassName);
-
-        SparkConf conf = new SparkConf();
-
-        runWithSparkSession(
-                conf,
-                isSparkSessionManaged,
-                spark -> {
-                    removeOutputDir(spark, outputPath);
-                    prepareResults(spark, inputPath, outputPath, bipScorePath, inputClazz);
-                });
-    }
-
-    private static <I extends Result> void prepareResults(SparkSession spark, String inputPath, String outputPath,
-                                                         String bipScorePath, Class<I> inputClazz) {
-
-        final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
-
-        JavaRDD<BipDeserialize> bipDeserializeJavaRDD = sc.textFile(bipScorePath)
-                .map(item -> OBJECT_MAPPER.readValue(item, BipDeserialize.class));
-
-
-        Dataset<BipScore> bipScores = spark.createDataset(bipDeserializeJavaRDD.flatMap(entry ->
-                entry.keySet().stream().map(key -> {
-                    BipScore bs = new BipScore();
-                    bs.setId(key);
-                    bs.setScoreList(entry.get(key));
-                    return bs;
-                }).collect(Collectors.toList()).iterator()
-        ).rdd(), Encoders.bean(BipScore.class));
-
-        System.out.println(bipScores.count());
-
-        Dataset<I> results = readPath(spark, inputPath, inputClazz);
-
-        results.createOrReplaceTempView("result");
-
-        Dataset<PreparedResult> preparedResult = spark.sql("select pIde.value value, id " +
-                "from result " +
-                "lateral view explode (pid) p as pIde " +
-                "where dataInfo.deletedbyinference = false and pIde.qualifier.classid = '" + DOI + "'")
-                .as(Encoders.bean(PreparedResult.class));
-
-        Dataset<BipScore> tmp = bipScores.joinWith(preparedResult, bipScores.col("id").equalTo(preparedResult.col("value")),
-                "inner")
-                .map((MapFunction<Tuple2<BipScore, PreparedResult>, BipScore>) value -> {
-                    BipScore ret = value._1();
-                    ret.setId(value._2().getId());
-                    return ret;
-                }, Encoders.bean(BipScore.class));
-
-        tmp.groupByKey((MapFunction<BipScore, String>) value -> value.getId(), Encoders.STRING())
-        .mapGroups((MapGroupsFunction<String, BipScore, I>) (k, it) ->
-        {
-            Result ret = inputClazz.newInstance();
-            BipScore first = it.next();
-            ret.setId(first.getId());
-
-            ret.setMeasures(getMeasure(first));
-            it.forEachRemaining(value -> ret.getMeasures().addAll(getMeasure(value)));
-
-            return (I)ret;
-        }, Encoders.bean(inputClazz))
-                .toJavaRDD()
-                .map(p -> new AtomicAction(inputClazz, p))
-                .mapToPair(
-                        aa -> new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
-                                new Text(OBJECT_MAPPER.writeValueAsString(aa))))
-                .saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class);
-        ;
-    }
-
-    public static Dataset<BipScore> getBipScoreDataset(Dataset<BipDeserialize> bipdeserialized){
-        return  bipdeserialized.flatMap((FlatMapFunction<BipDeserialize, BipScore>) bip ->
-                        bip.keySet().stream().map(key -> {
-                            BipScore bs = new BipScore();
-                            bs.setId(key);
-                            bs.setScoreList(bip.get(key));
-                            return bs;
-                        }).collect(Collectors.toList()).iterator()
-                , Encoders.bean(BipScore.class));
-    }
-
-    private static List<Measure> getMeasure(BipScore value) {
-        return value.getScoreList()
-                .stream()
-                .map(score -> {
-                    Measure m = new Measure();
-                    m.setId(score.getId());
-                    m.setUnit(score.getUnit().stream()
-                            .map(unit -> {
-                                KeyValue kv = new KeyValue();
-                                kv.setValue(unit.getValue());
-                                kv.setKey(unit.getKey());
-                                kv.setDataInfo(getDataInfo());
-                                return kv;
-                            }).collect(Collectors.toList()));
-                    return m;
-                }).collect(Collectors.toList());
-    }
-
-
-
-    private static DataInfo getDataInfo() {
-        DataInfo di = new DataInfo();
-        di.setInferred(false);
-        di.setInvisible(false);
-        di.setDeletedbyinference(false);
-        di.setTrust("");
-        Qualifier qualifier = new Qualifier();
-        qualifier.setClassid("sysimport:actionset");
-        qualifier.setClassname("Harvested");
-        qualifier.setSchemename("dnet:provenanceActions");
-        qualifier.setSchemeid("dnet:provenanceActions");
-        di.setProvenanceaction(qualifier);
-        return di;
-    }
-
-
-    private static void removeOutputDir(SparkSession spark, String path) {
-        HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
-    }
-
-    public static <R> Dataset<R> readPath(
-            SparkSession spark, String inputPath, Class<R> clazz) {
-        return spark
-                .read()
-                .textFile(inputPath)
-                .map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
-    }
+	public static <I extends Result> void main(String[] args) throws Exception {
 
+		String jsonConfiguration = IOUtils
+			.toString(
+				SparkAtomicActionScoreJob.class
+					.getResourceAsStream(
+						"/eu/dnetlib/dhp/actionmanager/bipfinder/input_parameters.json"));
+
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
+
+		parser.parseArgument(args);
+
+		Boolean isSparkSessionManaged = Optional
+			.ofNullable(parser.get("isSparkSessionManaged"))
+			.map(Boolean::valueOf)
+			.orElse(Boolean.TRUE);
+
+		log.info("isSparkSessionManaged: {}", isSparkSessionManaged);
+
+		final String inputPath = parser.get("inputPath");
+		log.info("inputPath {}: ", inputPath);
+
+		final String outputPath = parser.get("outputPath");
+		log.info("outputPath {}: ", outputPath);
+
+		final String bipScorePath = parser.get("bipScorePath");
+		log.info("bipScorePath: {}", bipScorePath);
+
+		final String resultClassName = parser.get("resultTableName");
+		log.info("resultTableName: {}", resultClassName);
+
+		Class<I> inputClazz = (Class<I>) Class.forName(resultClassName);
+
+		SparkConf conf = new SparkConf();
+
+		runWithSparkSession(
+			conf,
+			isSparkSessionManaged,
+			spark -> {
+				removeOutputDir(spark, outputPath);
+				prepareResults(spark, inputPath, outputPath, bipScorePath, inputClazz);
+			});
+	}
+
+	private static <I extends Result> void prepareResults(SparkSession spark, String inputPath, String outputPath,
+		String bipScorePath, Class<I> inputClazz) {
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<BipDeserialize> bipDeserializeJavaRDD = sc
+			.textFile(bipScorePath)
+			.map(item -> OBJECT_MAPPER.readValue(item, BipDeserialize.class));
+
+		Dataset<BipScore> bipScores = spark
+			.createDataset(bipDeserializeJavaRDD.flatMap(entry -> entry.keySet().stream().map(key -> {
+				BipScore bs = new BipScore();
+				bs.setId(key);
+				bs.setScoreList(entry.get(key));
+				return bs;
+			}).collect(Collectors.toList()).iterator()).rdd(), Encoders.bean(BipScore.class));
+
+		System.out.println(bipScores.count());
+
+		Dataset<I> results = readPath(spark, inputPath, inputClazz);
+
+		results.createOrReplaceTempView("result");
+
+		Dataset<PreparedResult> preparedResult = spark
+			.sql(
+				"select pIde.value value, id " +
+					"from result " +
+					"lateral view explode (pid) p as pIde " +
+					"where dataInfo.deletedbyinference = false and pIde.qualifier.classid = '" + DOI + "'")
+			.as(Encoders.bean(PreparedResult.class));
+
+		Dataset<BipScore> tmp = bipScores
+			.joinWith(
+				preparedResult, bipScores.col("id").equalTo(preparedResult.col("value")),
+				"inner")
+			.map((MapFunction<Tuple2<BipScore, PreparedResult>, BipScore>) value -> {
+				BipScore ret = value._1();
+				ret.setId(value._2().getId());
+				return ret;
+			}, Encoders.bean(BipScore.class));
+
+		tmp
+			.groupByKey((MapFunction<BipScore, String>) value -> value.getId(), Encoders.STRING())
+			.mapGroups((MapGroupsFunction<String, BipScore, I>) (k, it) -> {
+				Result ret = inputClazz.newInstance();
+				BipScore first = it.next();
+				ret.setId(first.getId());
+
+				ret.setMeasures(getMeasure(first));
+				it.forEachRemaining(value -> ret.getMeasures().addAll(getMeasure(value)));
+
+				return (I) ret;
+			}, Encoders.bean(inputClazz))
+			.toJavaRDD()
+			.map(p -> new AtomicAction(inputClazz, p))
+			.mapToPair(
+				aa -> new Tuple2<>(new Text(aa.getClazz().getCanonicalName()),
+					new Text(OBJECT_MAPPER.writeValueAsString(aa))))
+			.saveAsHadoopFile(outputPath, Text.class, Text.class, SequenceFileOutputFormat.class);
+		;
+	}
+
+	public static Dataset<BipScore> getBipScoreDataset(Dataset<BipDeserialize> bipdeserialized) {
+		return bipdeserialized
+			.flatMap((FlatMapFunction<BipDeserialize, BipScore>) bip -> bip.keySet().stream().map(key -> {
+				BipScore bs = new BipScore();
+				bs.setId(key);
+				bs.setScoreList(bip.get(key));
+				return bs;
+			}).collect(Collectors.toList()).iterator(), Encoders.bean(BipScore.class));
+	}
+
+	private static List<Measure> getMeasure(BipScore value) {
+		return value
+			.getScoreList()
+			.stream()
+			.map(score -> {
+				Measure m = new Measure();
+				m.setId(score.getId());
+				m
+					.setUnit(
+						score
+							.getUnit()
+							.stream()
+							.map(unit -> {
+								KeyValue kv = new KeyValue();
+								kv.setValue(unit.getValue());
+								kv.setKey(unit.getKey());
+								kv.setDataInfo(getDataInfo());
+								return kv;
+							})
+							.collect(Collectors.toList()));
+				return m;
+			})
+			.collect(Collectors.toList());
+	}
+
+	private static DataInfo getDataInfo() {
+		DataInfo di = new DataInfo();
+		di.setInferred(false);
+		di.setInvisible(false);
+		di.setDeletedbyinference(false);
+		di.setTrust("");
+		Qualifier qualifier = new Qualifier();
+		qualifier.setClassid("sysimport:actionset");
+		qualifier.setClassname("Harvested");
+		qualifier.setSchemename("dnet:provenanceActions");
+		qualifier.setSchemeid("dnet:provenanceActions");
+		di.setProvenanceaction(qualifier);
+		return di;
+	}
+
+	private static void removeOutputDir(SparkSession spark, String path) {
+		HdfsSupport.remove(path, spark.sparkContext().hadoopConfiguration());
+	}
+
+	public static <R> Dataset<R> readPath(
+		SparkSession spark, String inputPath, Class<R> clazz) {
+		return spark
+			.read()
+			.textFile(inputPath)
+			.map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
+	}
 
 }

dhp-workflows/dhp-aggregation/src/test/java/eu/dnetlib/dhp/actionmanager/bipfinder/SparkAtomicActionScoreJobTest.java

@@ -1,9 +1,12 @@
+
 package eu.dnetlib.dhp.actionmanager.bipfinder;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
-import eu.dnetlib.dhp.schema.action.AtomicAction;
-import eu.dnetlib.dhp.schema.oaf.*;
-import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.SequenceFileOutputFormat;
@@ -24,241 +27,305 @@ import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.Tuple2;
-
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.List;
-import java.util.stream.Collectors;
-
-public class SparkAtomicActionScoreJobTest {
-
-    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
-    private static SparkSession spark;
-
-    private static Path workingDir;
-    private static final Logger log = LoggerFactory
-            .getLogger(SparkAtomicActionScoreJobTest.class);
-
-    @BeforeAll
-    public static void beforeAll() throws IOException {
-        workingDir = Files
-                .createTempDirectory(SparkAtomicActionScoreJobTest.class.getSimpleName());
-        log.info("using work dir {}", workingDir);
-
-        SparkConf conf = new SparkConf();
-        conf.setAppName(SparkAtomicActionScoreJobTest.class.getSimpleName());
-
-        conf.setMaster("local[*]");
-        conf.set("spark.driver.host", "localhost");
-        conf.set("hive.metastore.local", "true");
-        conf.set("spark.ui.enabled", "false");
-        conf.set("spark.sql.warehouse.dir", workingDir.toString());
-        conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
-
-        spark = SparkSession
-                .builder()
-                .appName(SparkAtomicActionScoreJobTest.class.getSimpleName())
-                .config(conf)
-                .getOrCreate();
-    }
-
-    @AfterAll
-    public static void afterAll() throws IOException {
-        FileUtils.deleteDirectory(workingDir.toFile());
-        spark.stop();
-    }
-
-    @Test
-    public void matchOne() throws Exception {
-        String bipScoresPath =  getClass().getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json").getPath();
-        String inputPath = getClass()
-                .getResource(
-                        "/eu/dnetlib/dhp/actionmanager/bipfinder/publication.json")
-                .getPath();
-
-        SparkAtomicActionScoreJob.main(
-                        new String[] {
-                                "-isSparkSessionManaged",
-                                Boolean.FALSE.toString(),
-                                "-inputPath",
-                                inputPath,
-                                "-bipScorePath",
-                               bipScoresPath,
-                                "-resultTableName",
-                                "eu.dnetlib.dhp.schema.oaf.Publication",
-                                "-outputPath",
-                                workingDir.toString() + "/actionSet"
-                        });
-
-
-        final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
-
-        JavaRDD<Publication> tmp = sc
-                .sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
-                .map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
-                .map(aa -> ((Publication) aa.getPayload()));
-
-        Assertions.assertTrue(tmp.count() == 1);
-
-        Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
-        verificationDataset.createOrReplaceTempView("publication");
-
-        Dataset<Row> execVerification = spark.sql("Select p.id oaid, mes.id, mUnit.value from publication p " +
-                "lateral view explode(measures) m as mes " +
-                "lateral view explode(mes.unit) u as mUnit ");
-
-        Assertions.assertEquals(2, execVerification.count());
-
-        Assertions.assertEquals("50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb",
-                execVerification.select("oaid").collectAsList().get(0).getString(0));
-
-        Assertions.assertEquals("1.47565045883e-08",
-                execVerification.filter("id = 'influence'").select("value").collectAsList().get(0).getString(0));
-
-        Assertions.assertEquals("0.227515392",
-                execVerification.filter("id = 'popularity'").select("value").collectAsList().get(0).getString(0));
-
-    }
-
-    @Test
-    public void matchOneWithTwo() throws Exception {
-        String bipScoresPath =  getClass().getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json").getPath();
-        String inputPath = getClass()
-                .getResource(
-                        "/eu/dnetlib/dhp/actionmanager/bipfinder/publication_2.json")
-                .getPath();
-
-        SparkAtomicActionScoreJob.main(
-                new String[] {
-                        "-isSparkSessionManaged",
-                        Boolean.FALSE.toString(),
-                        "-inputPath",
-                        inputPath,
-                        "-bipScorePath",
-                        bipScoresPath,
-                        "-resultTableName",
-                        "eu.dnetlib.dhp.schema.oaf.Publication",
-                        "-outputPath",
-                        workingDir.toString() + "/actionSet"
-                });
-
-
-        final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
-
-        JavaRDD<Publication> tmp = sc
-                .sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
-                .map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
-                .map(aa -> ((Publication) aa.getPayload()));
-
-        Assertions.assertTrue(tmp.count() == 1);
-
-        Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
-        verificationDataset.createOrReplaceTempView("publication");
 
-        Dataset<Row> execVerification = spark.sql("Select p.id oaid, mes.id, mUnit.value from publication p " +
-                "lateral view explode(measures) m as mes " +
-                "lateral view explode(mes.unit) u as mUnit ");
-
-        Assertions.assertEquals(4, execVerification.count());
-
-
-        Assertions.assertEquals("50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb",
-                execVerification.select("oaid").collectAsList().get(0).getString(0));
-
-        Assertions.assertEquals(2,
-                execVerification.filter("id = 'influence'").count());
-
-        Assertions.assertEquals(2,
-                execVerification.filter("id = 'popularity'").count());
-
-        List<Row> tmp_ds = execVerification.filter("id = 'influence'").select("value").collectAsList();
-        String tmp_influence = tmp_ds.get(0).getString(0);
-        Assertions.assertTrue("1.47565045883e-08".equals(tmp_influence)||
-                "1.98956540239e-08".equals(tmp_influence) );
-
-        tmp_influence = tmp_ds.get(1).getString(0);
-        Assertions.assertTrue("1.47565045883e-08".equals(tmp_influence)||
-                "1.98956540239e-08".equals(tmp_influence) );
-
-        Assertions.assertTrue(!tmp_ds.get(0).getString(0).equals(tmp_ds.get(1).getString(0)));
-
-
-    }
-
-    @Test
-    public void matchTwo() throws Exception {
-        String bipScoresPath =  getClass().getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json").getPath();
-        String inputPath = getClass()
-                .getResource(
-                        "/eu/dnetlib/dhp/actionmanager/bipfinder/publication_3.json")
-                .getPath();
-
-        SparkAtomicActionScoreJob.main(
-                new String[] {
-                        "-isSparkSessionManaged",
-                        Boolean.FALSE.toString(),
-                        "-inputPath",
-                        inputPath,
-                        "-bipScorePath",
-                        bipScoresPath,
-                        "-resultTableName",
-                        "eu.dnetlib.dhp.schema.oaf.Publication",
-                        "-outputPath",
-                        workingDir.toString() + "/actionSet"
-                });
-
-
-        final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
-
-        JavaRDD<Publication> tmp = sc
-                .sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
-                .map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
-                .map(aa -> ((Publication) aa.getPayload()));
-
-        Assertions.assertTrue(tmp.count() == 2);
-
-        Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
-        verificationDataset.createOrReplaceTempView("publication");
-
-        Dataset<Row> execVerification = spark.sql("Select p.id oaid, mes.id, mUnit.value from publication p " +
-                "lateral view explode(measures) m as mes " +
-                "lateral view explode(mes.unit) u as mUnit ");
-
-        Assertions.assertEquals(4, execVerification.count());
-
-
-        Assertions.assertEquals(2,
-                execVerification.filter("oaid = '50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb'").count());
-
-        Assertions.assertEquals(2,
-                execVerification.filter("oaid = '50|acm_________::faed5b7a1bd8f51118d13ed29cfaee09'").count());
-
-
-        Assertions.assertEquals(2,
-                execVerification.filter("id = 'influence'").count());
-
-        Assertions.assertEquals(2,
-                execVerification.filter("id = 'popularity'").count());
-
-        Assertions.assertEquals("1.47565045883e-08" ,
-                execVerification.filter("oaid = '50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb' " +
-                        "and id = 'influence'").select("value").collectAsList().get(0).getString(0));
-
-        Assertions.assertEquals(
-                "1.98956540239e-08", execVerification.filter("oaid = '50|acm_________::faed5b7a1bd8f51118d13ed29cfaee09' " +
-                        "and id = 'influence'").select("value").collectAsList().get(0).getString(0));
+import com.fasterxml.jackson.databind.ObjectMapper;
 
-        Assertions.assertEquals(
-                "0.282046161584", execVerification.filter("oaid = '50|acm_________::faed5b7a1bd8f51118d13ed29cfaee09' " +
-                        "and id = 'popularity'").select("value").collectAsList().get(0).getString(0));
+import eu.dnetlib.dhp.schema.action.AtomicAction;
+import eu.dnetlib.dhp.schema.oaf.*;
+import eu.dnetlib.dhp.schema.oaf.KeyValue;
+import scala.Tuple2;
 
-        Assertions.assertEquals("0.227515392" ,
-                execVerification.filter("oaid = '50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb' " +
-                        "and id = 'popularity'").select("value").collectAsList().get(0).getString(0));
+public class SparkAtomicActionScoreJobTest {
 
-    }
+	private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+	private static SparkSession spark;
+
+	private static Path workingDir;
+	private static final Logger log = LoggerFactory
+		.getLogger(SparkAtomicActionScoreJobTest.class);
+
+	@BeforeAll
+	public static void beforeAll() throws IOException {
+		workingDir = Files
+			.createTempDirectory(SparkAtomicActionScoreJobTest.class.getSimpleName());
+		log.info("using work dir {}", workingDir);
+
+		SparkConf conf = new SparkConf();
+		conf.setAppName(SparkAtomicActionScoreJobTest.class.getSimpleName());
+
+		conf.setMaster("local[*]");
+		conf.set("spark.driver.host", "localhost");
+		conf.set("hive.metastore.local", "true");
+		conf.set("spark.ui.enabled", "false");
+		conf.set("spark.sql.warehouse.dir", workingDir.toString());
+		conf.set("hive.metastore.warehouse.dir", workingDir.resolve("warehouse").toString());
+
+		spark = SparkSession
+			.builder()
+			.appName(SparkAtomicActionScoreJobTest.class.getSimpleName())
+			.config(conf)
+			.getOrCreate();
+	}
+
+	@AfterAll
+	public static void afterAll() throws IOException {
+		FileUtils.deleteDirectory(workingDir.toFile());
+		spark.stop();
+	}
+
+	@Test
+	public void matchOne() throws Exception {
+		String bipScoresPath = getClass()
+			.getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json")
+			.getPath();
+		String inputPath = getClass()
+			.getResource(
+				"/eu/dnetlib/dhp/actionmanager/bipfinder/publication.json")
+			.getPath();
+
+		SparkAtomicActionScoreJob
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-inputPath",
+					inputPath,
+					"-bipScorePath",
+					bipScoresPath,
+					"-resultTableName",
+					"eu.dnetlib.dhp.schema.oaf.Publication",
+					"-outputPath",
+					workingDir.toString() + "/actionSet"
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<Publication> tmp = sc
+			.sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
+			.map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
+			.map(aa -> ((Publication) aa.getPayload()));
+
+		Assertions.assertTrue(tmp.count() == 1);
+
+		Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
+		verificationDataset.createOrReplaceTempView("publication");
+
+		Dataset<Row> execVerification = spark
+			.sql(
+				"Select p.id oaid, mes.id, mUnit.value from publication p " +
+					"lateral view explode(measures) m as mes " +
+					"lateral view explode(mes.unit) u as mUnit ");
+
+		Assertions.assertEquals(2, execVerification.count());
+
+		Assertions
+			.assertEquals(
+				"50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb",
+				execVerification.select("oaid").collectAsList().get(0).getString(0));
+
+		Assertions
+			.assertEquals(
+				"1.47565045883e-08",
+				execVerification.filter("id = 'influence'").select("value").collectAsList().get(0).getString(0));
+
+		Assertions
+			.assertEquals(
+				"0.227515392",
+				execVerification.filter("id = 'popularity'").select("value").collectAsList().get(0).getString(0));
+
+	}
+
+	@Test
+	public void matchOneWithTwo() throws Exception {
+		String bipScoresPath = getClass()
+			.getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json")
+			.getPath();
+		String inputPath = getClass()
+			.getResource(
+				"/eu/dnetlib/dhp/actionmanager/bipfinder/publication_2.json")
+			.getPath();
+
+		SparkAtomicActionScoreJob
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-inputPath",
+					inputPath,
+					"-bipScorePath",
+					bipScoresPath,
+					"-resultTableName",
+					"eu.dnetlib.dhp.schema.oaf.Publication",
+					"-outputPath",
+					workingDir.toString() + "/actionSet"
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<Publication> tmp = sc
+			.sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
+			.map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
+			.map(aa -> ((Publication) aa.getPayload()));
+
+		Assertions.assertTrue(tmp.count() == 1);
+
+		Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
+		verificationDataset.createOrReplaceTempView("publication");
+
+		Dataset<Row> execVerification = spark
+			.sql(
+				"Select p.id oaid, mes.id, mUnit.value from publication p " +
+					"lateral view explode(measures) m as mes " +
+					"lateral view explode(mes.unit) u as mUnit ");
+
+		Assertions.assertEquals(4, execVerification.count());
+
+		Assertions
+			.assertEquals(
+				"50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb",
+				execVerification.select("oaid").collectAsList().get(0).getString(0));
+
+		Assertions
+			.assertEquals(
+				2,
+				execVerification.filter("id = 'influence'").count());
+
+		Assertions
+			.assertEquals(
+				2,
+				execVerification.filter("id = 'popularity'").count());
+
+		List<Row> tmp_ds = execVerification.filter("id = 'influence'").select("value").collectAsList();
+		String tmp_influence = tmp_ds.get(0).getString(0);
+		Assertions
+			.assertTrue(
+				"1.47565045883e-08".equals(tmp_influence) ||
+					"1.98956540239e-08".equals(tmp_influence));
+
+		tmp_influence = tmp_ds.get(1).getString(0);
+		Assertions
+			.assertTrue(
+				"1.47565045883e-08".equals(tmp_influence) ||
+					"1.98956540239e-08".equals(tmp_influence));
+
+		Assertions.assertTrue(!tmp_ds.get(0).getString(0).equals(tmp_ds.get(1).getString(0)));
+
+	}
+
+	@Test
+	public void matchTwo() throws Exception {
+		String bipScoresPath = getClass()
+			.getResource("/eu/dnetlib/dhp/actionmanager/bipfinder/bip_scores.json")
+			.getPath();
+		String inputPath = getClass()
+			.getResource(
+				"/eu/dnetlib/dhp/actionmanager/bipfinder/publication_3.json")
+			.getPath();
+
+		SparkAtomicActionScoreJob
+			.main(
+				new String[] {
+					"-isSparkSessionManaged",
+					Boolean.FALSE.toString(),
+					"-inputPath",
+					inputPath,
+					"-bipScorePath",
+					bipScoresPath,
+					"-resultTableName",
+					"eu.dnetlib.dhp.schema.oaf.Publication",
+					"-outputPath",
+					workingDir.toString() + "/actionSet"
+				});
+
+		final JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+
+		JavaRDD<Publication> tmp = sc
+			.sequenceFile(workingDir.toString() + "/actionSet", Text.class, Text.class)
+			.map(value -> OBJECT_MAPPER.readValue(value._2().toString(), AtomicAction.class))
+			.map(aa -> ((Publication) aa.getPayload()));
+
+		Assertions.assertTrue(tmp.count() == 2);
+
+		Dataset<Publication> verificationDataset = spark.createDataset(tmp.rdd(), Encoders.bean(Publication.class));
+		verificationDataset.createOrReplaceTempView("publication");
+
+		Dataset<Row> execVerification = spark
+			.sql(
+				"Select p.id oaid, mes.id, mUnit.value from publication p " +
+					"lateral view explode(measures) m as mes " +
+					"lateral view explode(mes.unit) u as mUnit ");
+
+		Assertions.assertEquals(4, execVerification.count());
+
+		Assertions
+			.assertEquals(
+				2,
+				execVerification.filter("oaid = '50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb'").count());
+
+		Assertions
+			.assertEquals(
+				2,
+				execVerification.filter("oaid = '50|acm_________::faed5b7a1bd8f51118d13ed29cfaee09'").count());
+
+		Assertions
+			.assertEquals(
+				2,
+				execVerification.filter("id = 'influence'").count());
+
+		Assertions
+			.assertEquals(
+				2,
+				execVerification.filter("id = 'popularity'").count());
+
+		Assertions
+			.assertEquals(
+				"1.47565045883e-08",
+				execVerification
+					.filter(
+						"oaid = '50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb' " +
+							"and id = 'influence'")
+					.select("value")
+					.collectAsList()
+					.get(0)
+					.getString(0));
+
+		Assertions
+			.assertEquals(
+				"1.98956540239e-08",
+				execVerification
+					.filter(
+						"oaid = '50|acm_________::faed5b7a1bd8f51118d13ed29cfaee09' " +
+							"and id = 'influence'")
+					.select("value")
+					.collectAsList()
+					.get(0)
+					.getString(0));
+
+		Assertions
+			.assertEquals(
+				"0.282046161584",
+				execVerification
+					.filter(
+						"oaid = '50|acm_________::faed5b7a1bd8f51118d13ed29cfaee09' " +
+							"and id = 'popularity'")
+					.select("value")
+					.collectAsList()
+					.get(0)
+					.getString(0));
+
+		Assertions
+			.assertEquals(
+				"0.227515392",
+				execVerification
+					.filter(
+						"oaid = '50|355e65625b88::ffa5bad14f4adc0c9a15c00efbbccddb' " +
+							"and id = 'popularity'")
+					.select("value")
+					.collectAsList()
+					.get(0)
+					.getString(0));
+
+	}
 
 }