Merge pull request 'hostedbymap' (#136) from hostedbymap into beta

Reviewed-on: #136

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/utils/CSVParser.java

@@ -16,10 +16,15 @@ import org.apache.commons.lang.reflect.FieldUtils;
 public class CSVParser {
 
 	public <R> List<R> parse(String csvFile, String classForName)
+		throws ClassNotFoundException, IOException, IllegalAccessException, InstantiationException {
+		return parse(csvFile, classForName, ';');
+	}
+
+	public <R> List<R> parse(String csvFile, String classForName, char delimiter)
 		throws ClassNotFoundException, IOException, IllegalAccessException, InstantiationException {
 		final CSVFormat format = CSVFormat.EXCEL
 			.withHeader()
-			.withDelimiter(';')
+			.withDelimiter(delimiter)
 			.withQuote('"')
 			.withTrim();
 		List<R> ret = new ArrayList<>();

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/utils/ReadCSV.java

@@ -6,6 +6,7 @@ import java.io.Closeable;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.nio.charset.StandardCharsets;
+import java.util.Optional;
 
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.logging.Log;
@@ -29,6 +30,7 @@ public class ReadCSV implements Closeable {
 	private final BufferedWriter writer;
 	private final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 	private final String csvFile;
+	private final char delimiter;
 
 	public static void main(final String[] args) throws Exception {
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
@@ -44,19 +46,23 @@ public class ReadCSV implements Closeable {
 		final String hdfsPath = parser.get("hdfsPath");
 		final String hdfsNameNode = parser.get("hdfsNameNode");
 		final String classForName = parser.get("classForName");
-
-		try (final ReadCSV readCSV = new ReadCSV(hdfsPath, hdfsNameNode, fileURL)) {
+		Optional<String> delimiter = Optional.ofNullable(parser.get("delimiter"));
+		char del = ';';
+		if (delimiter.isPresent())
+			del = delimiter.get().charAt(0);
+		try (final ReadCSV readCSV = new ReadCSV(hdfsPath, hdfsNameNode, fileURL, del)) {
 
 			log.info("Getting CSV file...");
 			readCSV.execute(classForName);
 		}
+
 	}
 
 	public void execute(final String classForName)
 		throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException {
 		CSVParser csvParser = new CSVParser();
 		csvParser
-			.parse(csvFile, classForName)
+			.parse(csvFile, classForName, delimiter)
 			.stream()
 			.forEach(this::write);
 	}
@@ -69,7 +75,8 @@ public class ReadCSV implements Closeable {
 	public ReadCSV(
 		final String hdfsPath,
 		final String hdfsNameNode,
-		final String fileURL)
+		final String fileURL,
+		char delimiter)
 		throws Exception {
 		Configuration conf = new Configuration();
 		conf.set("fs.defaultFS", hdfsNameNode);
@@ -84,6 +91,7 @@ public class ReadCSV implements Closeable {
 
 		this.writer = new BufferedWriter(new OutputStreamWriter(fos, StandardCharsets.UTF_8));
 		this.csvFile = httpConnector.getInputSource(fileURL);
+		this.delimiter = delimiter;
 	}
 
 	protected void write(final Object p) {

dhp-workflows/dhp-aggregation/src/main/java/eu/dnetlib/dhp/actionmanager/project/utils/ReadExcel.java

@@ -33,7 +33,7 @@ public class ReadExcel implements Closeable {
 		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
 			IOUtils
 				.toString(
-					ReadCSV.class
+					ReadExcel.class
 						.getResourceAsStream(
 							"/eu/dnetlib/dhp/actionmanager/project/parameters.json")));
 

dhp-workflows/dhp-aggregation/src/main/resources/eu/dnetlib/dhp/actionmanager/project/parameters.json

@@ -28,6 +28,11 @@
   "paramLongName" : "sheetName",
   "paramDescription" : "the name of the sheet in case the file is excel",
   "paramRequired" : false
+}, {
+  "paramName": "d",
+  "paramLongName" : "delimiter",
+  "paramDescription" : "the delimiter between fields in case it is not ;",
+  "paramRequired" : false
 }
 
 

dhp-workflows/dhp-doiboost/src/main/java/eu/dnetlib/doiboost/DoiBoostMappingUtil.scala

@@ -232,6 +232,8 @@ object DoiBoostMappingUtil {
       if (item != null) {
         hb.setValue(item.officialname)
         hb.setKey(generateDSId(item.id))
+        //TODO replace with the one above as soon as  the new HBM will be used
+        //hb.setKey(item.id)
         if (item.openAccess) {
           i.setAccessright(getOpenAccessQualifier())
           i.getAccessright.setOpenAccessRoute(OpenAccessRoute.gold)

dhp-workflows/dhp-doiboost/src/main/resources/eu/dnetlib/dhp/doiboost/download_unibi_issn_gold_parameters.json

@@ -0,0 +1,39 @@
+[
+
+  {
+    "paramName": "fu",
+    "paramLongName" : "fileURL",
+    "paramDescription" : "the url of the file to download",
+    "paramRequired" : true
+  },
+  {
+    "paramName": "hp",
+    "paramLongName" : "hdfsPath",
+    "paramDescription" : "where to save the file",
+    "paramRequired" : true
+  },
+  {
+    "paramName": "hnn",
+    "paramLongName" : "hdfsNameNode",
+    "paramDescription" : "the name node",
+    "paramRequired" : true
+  },
+  {
+  "paramName": "cfn",
+  "paramLongName" : "classForName",
+  "paramDescription" : "the name of the class to deserialize the csv to",
+  "paramRequired" : true
+}, {
+  "paramName": "sn",
+  "paramLongName" : "sheetName",
+  "paramDescription" : "the name of the sheet in case the file is excel",
+  "paramRequired" : false
+}, {
+  "paramName": "d",
+  "paramLongName" : "delimiter",
+  "paramDescription" : "the delimiter between fields in case it is not ;",
+  "paramRequired" : false
+}
+
+
+]

dhp-workflows/dhp-doiboost/src/main/resources/eu/dnetlib/dhp/doiboost/preprocess/oozie_app/workflow.xml

@@ -76,6 +76,7 @@
         <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
     </kill>
 
+
     <action name="ImportCrossRef">
         <java>
             <job-tracker>${jobTracker}</job-tracker>

dhp-workflows/dhp-graph-mapper/pom.xml

@@ -122,6 +122,10 @@
             <groupId>org.json4s</groupId>
             <artifactId>json4s-jackson_2.11</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.opencsv</groupId>
+            <artifactId>opencsv</artifactId>
+        </dependency>
 
     </dependencies>
 

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/Aggregators.scala

@@ -0,0 +1,142 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.EntityInfo
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, TypedColumn}
+import org.apache.spark.sql.expressions.Aggregator
+
+
+case class HostedByItemType(id: String, officialname: String, issn: String, eissn: String, lissn: String, openAccess: Boolean) {}
+case class HostedByInfo(id: String, officialname: String, journal_id: String, provenance : String, id_type: String) {}
+
+object Aggregators {
+
+
+
+  def getId(s1:String, s2:String) : String = {
+    if (s1.startsWith("10|")){
+      return s1}
+    s2
+  }
+
+  def getValue(s1:String, s2:String) : String = {
+    if(!s1.equals("")){
+      return s1
+    }
+    s2
+  }
+
+
+  def explodeHostedByItemType(df: Dataset[(String, HostedByItemType)]): Dataset[(String, HostedByItemType)] = {
+    val transformedData : Dataset[(String, HostedByItemType)] = df
+      .groupByKey(_._1)(Encoders.STRING)
+      .agg(Aggregators.hostedByAggregator)
+      .map{
+        case (id:String , res:(String, HostedByItemType)) => res
+      }(Encoders.tuple(Encoders.STRING, Encoders.product[HostedByItemType]))
+
+    transformedData
+  }
+
+  val hostedByAggregator: TypedColumn[(String, HostedByItemType), (String, HostedByItemType)] = new Aggregator[(String, HostedByItemType), (String, HostedByItemType), (String, HostedByItemType)] {
+    override def zero: (String, HostedByItemType) = ("", HostedByItemType("","","","","",false))
+    override def reduce(b: (String, HostedByItemType), a:(String,HostedByItemType)): (String, HostedByItemType) = {
+      return merge(b, a)
+    }
+    override def merge(b1: (String, HostedByItemType), b2: (String, HostedByItemType)): (String, HostedByItemType) = {
+      if (b1 == null){
+        return b2
+      }
+      if(b2 == null){
+        return b1
+      }
+      if(b1._2.id.startsWith("10|")){
+        return (b1._1, HostedByItemType(b1._2.id, b1._2.officialname, b1._2.issn, b1._2.eissn, b1._2.lissn, b1._2.openAccess || b2._2.openAccess))
+
+      }
+      return (b2._1, HostedByItemType(b2._2.id, b2._2.officialname, b2._2.issn, b2._2.eissn, b2._2.lissn, b1._2.openAccess || b2._2.openAccess))
+
+    }
+    override def finish(reduction: (String,HostedByItemType)): (String, HostedByItemType) = reduction
+    override def bufferEncoder: Encoder[(String,HostedByItemType)] = Encoders.tuple(Encoders.STRING,Encoders.product[HostedByItemType])
+
+    override def outputEncoder: Encoder[(String,HostedByItemType)] = Encoders.tuple(Encoders.STRING,Encoders.product[HostedByItemType])
+  }.toColumn
+
+
+
+
+  def resultToSingleIdAggregator: TypedColumn[EntityInfo, EntityInfo] = new Aggregator[EntityInfo, EntityInfo, EntityInfo]{
+    override def zero: EntityInfo = EntityInfo.newInstance("","","")
+
+    override def reduce(b: EntityInfo, a:EntityInfo):  EntityInfo = {
+      return merge(b, a)
+    }
+    override def merge(b1: EntityInfo, b2: EntityInfo): EntityInfo = {
+      if (b1 == null){
+        return b2
+      }
+      if(b2 == null){
+        return b1
+      }
+      if(!b1.getHb_id.equals("")){
+        b1.setOpenaccess(b1.getOpenaccess || b2.getOpenaccess)
+        return b1
+      }
+      b2.setOpenaccess(b1.getOpenaccess || b2.getOpenaccess)
+      b2
+
+    }
+    override def finish(reduction: EntityInfo):  EntityInfo = reduction
+    override def bufferEncoder: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+    override def outputEncoder: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+  }.toColumn
+
+  def resultToSingleId(df:Dataset[EntityInfo]): Dataset[EntityInfo] = {
+    val transformedData : Dataset[EntityInfo] = df
+      .groupByKey(_.getId)(Encoders.STRING)
+      .agg(Aggregators.resultToSingleIdAggregator)
+      .map{
+        case (id:String , res: EntityInfo) => res
+      }(Encoders.bean(classOf[EntityInfo]))
+
+    transformedData
+  }
+
+  def datasourceToSingleIdAggregator: TypedColumn[EntityInfo, EntityInfo] = new Aggregator[EntityInfo, EntityInfo, EntityInfo]{
+    override def zero: EntityInfo = EntityInfo.newInstance("","","")
+
+    override def reduce(b: EntityInfo, a:EntityInfo):  EntityInfo = {
+      return merge(b, a)
+    }
+    override def merge(b1: EntityInfo, b2: EntityInfo): EntityInfo = {
+      if (b1 == null){
+        return b2
+      }
+      if(b2 == null){
+        return b1
+      }
+      if(!b1.getHb_id.equals("")){
+        return b1
+      }
+      b2
+
+    }
+    override def finish(reduction: EntityInfo):  EntityInfo = reduction
+    override def bufferEncoder: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+    override def outputEncoder: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+  }.toColumn
+
+
+  def datasourceToSingleId(df:Dataset[EntityInfo]): Dataset[EntityInfo] = {
+    val transformedData : Dataset[EntityInfo] = df
+      .groupByKey(_.getHb_id)(Encoders.STRING)
+      .agg(Aggregators.datasourceToSingleIdAggregator)
+      .map{
+        case (id:String , res: EntityInfo) => res
+      }(Encoders.bean(classOf[EntityInfo]))
+
+    transformedData
+  }
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/Constants.java

@@ -0,0 +1,13 @@
+
+package eu.dnetlib.dhp.oa.graph.hostedbymap;
+
+public class Constants {
+
+	public static final String OPENAIRE = "openaire";
+	public static final String DOAJ = "doaj";
+	public static final String UNIBI = "unibi";
+
+	public static final String ISSN = "issn";
+	public static final String EISSN = "eissn";
+	public static final String ISSNL = "issnl";
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/GetCSV.java

@@ -0,0 +1,107 @@
+
+package eu.dnetlib.dhp.oa.graph.hostedbymap;
+
+import java.io.*;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.opencsv.bean.CsvToBeanBuilder;
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser;
+
+public class GetCSV {
+	private static final Log log = LogFactory.getLog(eu.dnetlib.dhp.oa.graph.hostedbymap.GetCSV.class);
+
+	public static void main(final String[] args) throws Exception {
+		final ArgumentApplicationParser parser = new ArgumentApplicationParser(
+			IOUtils
+				.toString(
+					GetCSV.class
+						.getResourceAsStream(
+							"/eu/dnetlib/dhp/oa/graph/hostedbymap/download_csv_parameters.json")));
+
+		parser.parseArgument(args);
+
+		final String fileURL = parser.get("fileURL");
+		final String hdfsPath = parser.get("workingPath");
+		final String hdfsNameNode = parser.get("hdfsNameNode");
+		final String classForName = parser.get("classForName");
+		final Boolean shouldReplace = Optional
+			.ofNullable((parser.get("replace")))
+			.map(Boolean::valueOf)
+			.orElse(false);
+
+		URLConnection connection = new URL(fileURL).openConnection();
+		connection
+			.setRequestProperty(
+				"User-Agent",
+				"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.95 Safari/537.11");
+		connection.connect();
+
+		BufferedReader in = new BufferedReader(
+			new InputStreamReader(connection.getInputStream(), Charset.forName("UTF-8")));
+
+		if (shouldReplace) {
+			PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter("/tmp/DOAJ.csv")));
+			String line = null;
+			while ((line = in.readLine()) != null) {
+				writer.println(line.replace("\\\"", "\""));
+			}
+			writer.close();
+			in.close();
+			in = new BufferedReader(new FileReader("/tmp/DOAJ.csv"));
+		}
+
+		Configuration conf = new Configuration();
+		conf.set("fs.defaultFS", hdfsNameNode);
+
+		FileSystem fileSystem = FileSystem.get(conf);
+		Path hdfsWritePath = new Path(hdfsPath);
+		FSDataOutputStream fsDataOutputStream = null;
+		if (fileSystem.exists(hdfsWritePath)) {
+			fileSystem.delete(hdfsWritePath, false);
+		}
+		fsDataOutputStream = fileSystem.create(hdfsWritePath);
+
+		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream, StandardCharsets.UTF_8));
+
+		Class<?> clazz = Class.forName(classForName);
+
+		ObjectMapper mapper = new ObjectMapper();
+
+		new CsvToBeanBuilder(in)
+			.withType(clazz)
+			.withMultilineLimit(1)
+			.build()
+			.parse()
+			.forEach(line -> {
+				try {
+					writer.write(mapper.writeValueAsString(line));
+					writer.newLine();
+				} catch (IOException e) {
+					throw new RuntimeException(e);
+				}
+			});
+
+		writer.close();
+		in.close();
+		if (shouldReplace) {
+			File f = new File("/tmp/DOAJ.csv");
+			f.delete();
+		}
+
+	}
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/SparkApplyHostedByMapToDatasource.scala

@@ -0,0 +1,68 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import eu.dnetlib.dhp.oa.graph.hostedbymap.SparkApplyHostedByMapToResult.{applyHBtoPubs, getClass}
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.EntityInfo
+import eu.dnetlib.dhp.schema.common.ModelConstants
+import eu.dnetlib.dhp.schema.oaf.{Datasource, Publication}
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
+import org.json4s.DefaultFormats
+import org.slf4j.{Logger, LoggerFactory}
+
+object SparkApplyHostedByMapToDatasource {
+
+  def applyHBtoDats(join: Dataset[EntityInfo], dats: Dataset[Datasource]): Dataset[Datasource] = {
+    dats.joinWith(join, dats.col("id").equalTo(join.col("hb_id")), "left")
+      .map(t2 => {
+        val d: Datasource = t2._1
+        if (t2._2 != null) {
+          if (d.getOpenairecompatibility.getClassid.equals(ModelConstants.UNKNOWN)) {
+            d.getOpenairecompatibility.setClassid("hostedBy")
+            d.getOpenairecompatibility.setClassname("collected from a compatible aggregator")
+          }
+        }
+        d
+      })(Encoders.bean((classOf[Datasource])))
+  }
+  def main(args: Array[String]): Unit = {
+
+
+    val logger: Logger = LoggerFactory.getLogger(getClass)
+    val conf: SparkConf = new SparkConf()
+    val parser = new ArgumentApplicationParser(IOUtils.toString(getClass.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_apply_params.json")))
+    parser.parseArgument(args)
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .config(conf)
+        .appName(getClass.getSimpleName)
+        .master(parser.get("master")).getOrCreate()
+
+
+    val graphPath = parser.get("graphPath")
+
+    val outputPath = parser.get("outputPath")
+    val preparedInfoPath = parser.get("preparedInfoPath")
+
+
+    implicit val formats = DefaultFormats
+
+
+    implicit val mapEncoderPubs: Encoder[Datasource] = Encoders.bean(classOf[Datasource])
+    implicit val mapEncoderEinfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+    val mapper = new ObjectMapper()
+
+    val dats : Dataset[Datasource] = spark.read.textFile(graphPath + "/datasource")
+      .map(r => mapper.readValue(r, classOf[Datasource]))
+
+    val pinfo : Dataset[EntityInfo] = Aggregators.datasourceToSingleId( spark.read.textFile(preparedInfoPath)
+      .map(ei => mapper.readValue(ei, classOf[EntityInfo])))
+
+    applyHBtoDats(pinfo, dats).write.mode(SaveMode.Overwrite).option("compression","gzip").json(outputPath)
+  }
+
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/SparkApplyHostedByMapToResult.scala

@@ -0,0 +1,83 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.EntityInfo
+import eu.dnetlib.dhp.schema.common.ModelConstants
+import eu.dnetlib.dhp.schema.oaf.utils.OafMapperUtils
+import eu.dnetlib.dhp.schema.oaf.{Datasource, Instance, OpenAccessRoute, Publication}
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
+import org.json4s.DefaultFormats
+import org.slf4j.{Logger, LoggerFactory}
+
+import scala.collection.JavaConverters._
+
+
+object SparkApplyHostedByMapToResult {
+
+  def applyHBtoPubs(join: Dataset[EntityInfo], pubs: Dataset[Publication]) = {
+    pubs.joinWith(join, pubs.col("id").equalTo(join.col("id")), "left")
+      .map(t2 => {
+        val p: Publication = t2._1
+        if (t2._2 != null) {
+          val ei: EntityInfo = t2._2
+          val i = p.getInstance().asScala
+          if (i.size == 1) {
+            val inst: Instance = i(0)
+            inst.getHostedby.setKey(ei.getHb_id)
+            inst.getHostedby.setValue(ei.getName)
+            if (ei.getOpenaccess) {
+              inst.setAccessright(OafMapperUtils.accessRight(ModelConstants.ACCESS_RIGHT_OPEN, "Open Access", ModelConstants.DNET_ACCESS_MODES, ModelConstants.DNET_ACCESS_MODES))
+              inst.getAccessright.setOpenAccessRoute(OpenAccessRoute.hybrid)
+              p.setBestaccessright(OafMapperUtils.createBestAccessRights(p.getInstance()));
+            }
+
+          }
+        }
+        p
+      })(Encoders.bean(classOf[Publication]))
+  }
+  def main(args: Array[String]): Unit = {
+
+
+    val logger: Logger = LoggerFactory.getLogger(getClass)
+    val conf: SparkConf = new SparkConf()
+    val parser = new ArgumentApplicationParser(IOUtils.toString(getClass.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_apply_params.json")))
+    parser.parseArgument(args)
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .config(conf)
+        .appName(getClass.getSimpleName)
+        .master(parser.get("master")).getOrCreate()
+
+
+    val graphPath = parser.get("graphPath")
+
+    val outputPath = parser.get("outputPath")
+    val preparedInfoPath = parser.get("preparedInfoPath")
+
+
+    implicit val formats = DefaultFormats
+
+
+    implicit val mapEncoderPubs: Encoder[Publication] = Encoders.bean(classOf[Publication])
+    implicit val mapEncoderEinfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+    val mapper = new ObjectMapper()
+
+    val pubs : Dataset[Publication] = spark.read.textFile(graphPath + "/publication")
+      .map(r => mapper.readValue(r, classOf[Publication]))
+
+    val pinfo : Dataset[EntityInfo] = spark.read.textFile(preparedInfoPath)
+        .map(ei => mapper.readValue(ei, classOf[EntityInfo]))
+
+    applyHBtoPubs(pinfo, pubs).write.mode(SaveMode.Overwrite).option("compression","gzip").json(outputPath)
+
+
+
+  }
+
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/SparkPrepareHostedByInfoToApply.scala

@@ -0,0 +1,125 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.EntityInfo
+
+import eu.dnetlib.dhp.schema.oaf.{Journal, Publication}
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
+import org.json4s
+import org.json4s.DefaultFormats
+import org.json4s.jackson.JsonMethods.parse
+import org.slf4j.{Logger, LoggerFactory}
+
+
+
+object SparkPrepareHostedByInfoToApply {
+
+  implicit val mapEncoderPInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+  def getList(id: String, j: Journal, name: String ) : List[EntityInfo] = {
+    var lst:List[EntityInfo] = List()
+
+
+    if (j.getIssnLinking != null && !j.getIssnLinking.equals("")){
+      lst = EntityInfo.newInstance(id, j.getIssnLinking, name) :: lst
+    }
+    if (j.getIssnOnline != null && !j.getIssnOnline.equals("")){
+      lst = EntityInfo.newInstance(id, j.getIssnOnline, name) :: lst
+    }
+    if (j.getIssnPrinted != null && !j.getIssnPrinted.equals("")){
+      lst = EntityInfo.newInstance(id, j.getIssnPrinted, name) :: lst
+    }
+    lst
+  }
+
+  def prepareResultInfo(spark:SparkSession, publicationPath:String) : Dataset[EntityInfo] = {
+    implicit val mapEncoderPubs: Encoder[Publication] = Encoders.bean(classOf[Publication])
+
+    val mapper = new ObjectMapper()
+
+    val dd : Dataset[Publication] = spark.read.textFile(publicationPath)
+      .map(r => mapper.readValue(r, classOf[Publication]))
+
+    dd.filter(p => p.getJournal != null ).flatMap(p => getList(p.getId, p.getJournal, ""))
+
+  }
+
+
+  def toEntityInfo(input:String): EntityInfo = {
+    implicit lazy val formats: DefaultFormats.type = org.json4s.DefaultFormats
+
+    lazy val json: json4s.JValue = parse(input)
+    val c :Map[String,HostedByItemType] = json.extract[Map[String, HostedByItemType]]
+    toEntityItem(c.keys.head, c.values.head)
+  }
+
+
+  def toEntityItem(journal_id: String , hbi: HostedByItemType): EntityInfo = {
+
+    EntityInfo.newInstance(hbi.id, journal_id, hbi.officialname, hbi.openAccess)
+
+  }
+
+  def joinResHBM(res: Dataset[EntityInfo], hbm: Dataset[EntityInfo]): Dataset[EntityInfo] = {
+    Aggregators.resultToSingleId(res.joinWith(hbm, res.col("journal_id").equalTo(hbm.col("journal_id")), "left")
+      .map(t2 => {
+        val res: EntityInfo = t2._1
+        if(t2._2 != null ){
+          val ds = t2._2
+          res.setHb_id(ds.getId)
+          res.setOpenaccess(ds.getOpenaccess)
+          res.setName(ds.getName)
+        }
+        res
+      }))
+  }
+
+  def main(args: Array[String]): Unit = {
+
+
+    val logger: Logger = LoggerFactory.getLogger(getClass)
+    val conf: SparkConf = new SparkConf()
+    val parser = new ArgumentApplicationParser(IOUtils.toString(getClass.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_prepare_params.json")))
+    parser.parseArgument(args)
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .config(conf)
+        .appName(getClass.getSimpleName)
+        .master(parser.get("master")).getOrCreate()
+
+
+    val graphPath = parser.get("graphPath")
+
+    val outputPath = parser.get("preparedInfoPath")
+    val hostedByMapPath = parser.get("hostedByMapPath")
+
+
+    implicit val formats = DefaultFormats
+
+
+    logger.info("Getting the Datasources")
+
+    import spark.implicits._
+
+
+    //STEP1: read the hostedbymap and transform it in EntityInfo
+    val hostedByInfo:Dataset[EntityInfo] = spark.createDataset(spark.sparkContext.textFile(hostedByMapPath)).map(toEntityInfo)
+
+   //STEP2: create association (publication, issn), (publication, eissn), (publication, lissn)
+    val resultInfoDataset:Dataset[EntityInfo] = prepareResultInfo(spark, graphPath + "/publication")
+
+    //STEP3: left join resultInfo with hostedByInfo on journal_id. Reduction of all the results with the same id in just
+    //one entry (one result could be associated to issn and eissn and so possivly matching more than once against the map)
+    //to this entry we add the id of the datasource for the next step
+    joinResHBM(resultInfoDataset, hostedByInfo)
+      .write.mode(SaveMode.Overwrite).option("compression", "gzip").json(outputPath)
+
+
+  }
+
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/SparkProduceHostedByMap.scala

@@ -0,0 +1,222 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import eu.dnetlib.dhp.application.ArgumentApplicationParser
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.{DOAJModel, UnibiGoldModel}
+import eu.dnetlib.dhp.schema.oaf.Datasource
+import org.apache.commons.io.IOUtils
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
+import org.json4s.DefaultFormats
+import org.slf4j.{Logger, LoggerFactory}
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.FileSystem
+import org.apache.hadoop.fs.Path
+import java.io.PrintWriter
+
+import org.apache.hadoop.io.compress.GzipCodec
+
+
+object SparkProduceHostedByMap {
+
+
+  implicit val tupleForJoinEncoder: Encoder[(String, HostedByItemType)] = Encoders.tuple(Encoders.STRING, Encoders.product[HostedByItemType])
+
+
+  def toHostedByItemType(input: ((HostedByInfo, HostedByInfo), HostedByInfo)) : HostedByItemType = {
+    val openaire: HostedByInfo = input._1._1
+    val doaj: HostedByInfo = input._1._2
+    val gold: HostedByInfo = input._2
+    val isOpenAccess: Boolean = doaj == null && gold == null
+
+    openaire.journal_id match {
+      case Constants.ISSN =>  HostedByItemType(openaire.id, openaire.officialname, openaire.journal_id, "", "", isOpenAccess)
+      case Constants.EISSN =>  HostedByItemType(openaire.id, openaire.officialname, "", openaire.journal_id, "", isOpenAccess)
+      case Constants.ISSNL =>  HostedByItemType(openaire.id, openaire.officialname, "", "", openaire.journal_id, isOpenAccess)
+
+      // catch the default with a variable so you can print it
+      case whoa =>  null
+    }
+  }
+
+  def toHostedByMap(input: (String, HostedByItemType)): String = {
+    import org.json4s.jackson.Serialization
+
+    implicit val formats = org.json4s.DefaultFormats
+
+    val map: Map [String, HostedByItemType] = Map (input._1 -> input._2 )
+
+    Serialization.write(map)
+
+
+  }
+
+
+
+  def getHostedByItemType(id:String, officialname: String, issn:String, eissn:String, issnl:String, oa:Boolean): HostedByItemType = {
+    if(issn != null){
+      if(eissn != null){
+        if(issnl != null){
+          HostedByItemType(id, officialname, issn, eissn, issnl  , oa)
+        }else{
+          HostedByItemType(id, officialname, issn, eissn, ""  , oa)
+        }
+      }else{
+        if(issnl != null){
+          HostedByItemType(id, officialname, issn, "", issnl  , oa)
+        }else{
+          HostedByItemType(id, officialname, issn, "", ""  , oa)
+        }
+      }
+    }else{
+      if(eissn != null){
+        if(issnl != null){
+          HostedByItemType(id, officialname, "", eissn, issnl  , oa)
+        }else{
+          HostedByItemType(id, officialname, "", eissn, ""  , oa)
+        }
+      }else{
+        if(issnl != null){
+          HostedByItemType(id, officialname, "", "", issnl  , oa)
+        }else{
+          HostedByItemType("", "", "", "", ""  , oa)
+        }
+      }
+    }
+  }
+
+  def oaToHostedbyItemType(dats: Datasource): HostedByItemType = {
+    if (dats.getJournal != null) {
+
+      return getHostedByItemType(dats.getId, dats.getOfficialname.getValue, dats.getJournal.getIssnPrinted, dats.getJournal.getIssnOnline, dats.getJournal.getIssnLinking, false)
+    }
+    HostedByItemType("","","","","",false)
+  }
+
+  def oaHostedByDataset(spark:SparkSession, datasourcePath : String) : Dataset[HostedByItemType] = {
+
+    import spark.implicits._
+
+
+    val mapper = new ObjectMapper()
+
+    implicit var encoderD = Encoders.kryo[Datasource]
+
+    val dd : Dataset[Datasource] = spark.read.textFile(datasourcePath)
+      .map(r => mapper.readValue(r, classOf[Datasource]))
+
+    dd.map{ddt => oaToHostedbyItemType(ddt)}.filter(hb => !(hb.id.equals("")))
+
+  }
+
+
+  def goldToHostedbyItemType(gold: UnibiGoldModel): HostedByItemType = {
+    return getHostedByItemType(Constants.UNIBI, gold.getTitle, gold.getIssn, "", gold.getIssn_l, true)
+  }
+
+
+  def goldHostedByDataset(spark:SparkSession, datasourcePath:String) : Dataset[HostedByItemType] = {
+    import spark.implicits._
+
+    implicit val mapEncoderUnibi: Encoder[UnibiGoldModel] = Encoders.kryo[UnibiGoldModel]
+
+    val mapper = new ObjectMapper()
+
+    val dd : Dataset[UnibiGoldModel] = spark.read.textFile(datasourcePath)
+      .map(r => mapper.readValue(r, classOf[UnibiGoldModel]))
+
+    dd.map{ddt => goldToHostedbyItemType(ddt)}.filter(hb => !(hb.id.equals("")))
+
+  }
+
+  def doajToHostedbyItemType(doaj: DOAJModel): HostedByItemType = {
+
+    return getHostedByItemType(Constants.DOAJ, doaj.getJournalTitle, doaj.getIssn, doaj.getEissn, "", true)
+  }
+
+  def doajHostedByDataset(spark:SparkSession, datasourcePath:String) : Dataset[HostedByItemType] = {
+    import spark.implicits._
+
+    implicit val mapEncoderDOAJ: Encoder[DOAJModel] = Encoders.kryo[DOAJModel]
+
+    val mapper = new ObjectMapper()
+
+    val dd : Dataset[DOAJModel] = spark.read.textFile(datasourcePath)
+      .map(r => mapper.readValue(r, classOf[DOAJModel]))
+
+    dd.map{ddt => doajToHostedbyItemType(ddt)}.filter(hb => !(hb.id.equals("")))
+
+  }
+
+  def toList(input: HostedByItemType): List[(String, HostedByItemType)] = {
+    var lst : List[(String, HostedByItemType)] = List()
+    if(!input.issn.equals("")){
+      lst = (input.issn, input) :: lst
+    }
+    if(!input.eissn.equals("")){
+      lst = (input.eissn, input) :: lst
+    }
+    if(!input.lissn.equals("")){
+      lst = (input.lissn, input) :: lst
+    }
+    lst
+  }
+
+
+
+  def writeToHDFS(input: Array[String], outputPath: String, hdfsNameNode : String):Unit = {
+    val conf = new Configuration()
+
+    conf.set("fs.defaultFS", hdfsNameNode)
+    val fs= FileSystem.get(conf)
+    val output = fs.create(new Path(outputPath))
+    val writer = new PrintWriter(output)
+    try {
+      input.foreach(hbi => writer.println(hbi))
+    }
+    finally {
+      writer.close()
+
+    }
+
+  }
+
+
+
+  def main(args: Array[String]): Unit = {
+
+    val logger: Logger = LoggerFactory.getLogger(getClass)
+    val conf: SparkConf = new SparkConf()
+    val parser = new ArgumentApplicationParser(IOUtils.toString(getClass.getResourceAsStream("/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_params.json")))
+    parser.parseArgument(args)
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .config(conf)
+        .appName(getClass.getSimpleName)
+        .master(parser.get("master")).getOrCreate()
+
+
+    val datasourcePath = parser.get("datasourcePath")
+    val workingDirPath = parser.get("workingPath")
+    val outputPath = parser.get("outputPath")
+
+
+    implicit val formats = DefaultFormats
+
+
+    logger.info("Getting the Datasources")
+
+
+    Aggregators.explodeHostedByItemType(oaHostedByDataset(spark, datasourcePath)
+      .union(goldHostedByDataset(spark, workingDirPath + "/unibi_gold"))
+      .union(doajHostedByDataset(spark, workingDirPath + "/doaj"))
+      .flatMap(hbi => toList(hbi))).filter(hbi => hbi._2.id.startsWith("10|"))
+      .map(hbi => toHostedByMap(hbi))(Encoders.STRING)
+      .rdd.saveAsTextFile(outputPath , classOf[GzipCodec])
+
+
+  }
+
+
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/model/DOAJModel.java

@@ -0,0 +1,52 @@
+
+package eu.dnetlib.dhp.oa.graph.hostedbymap.model;
+
+import java.io.Serializable;
+
+import com.opencsv.bean.CsvBindByName;
+
+public class DOAJModel implements Serializable {
+	@CsvBindByName(column = "Journal title")
+	private String journalTitle;
+
+	@CsvBindByName(column = "Journal ISSN (print version)")
+	private String issn;
+
+	@CsvBindByName(column = "Journal EISSN (online version)")
+	private String eissn;
+
+	@CsvBindByName(column = "Review process")
+	private String reviewProcess;
+
+	public String getJournalTitle() {
+		return journalTitle;
+	}
+
+	public void setJournalTitle(String journalTitle) {
+		this.journalTitle = journalTitle;
+	}
+
+	public String getIssn() {
+		return issn;
+	}
+
+	public void setIssn(String issn) {
+		this.issn = issn;
+	}
+
+	public String getEissn() {
+		return eissn;
+	}
+
+	public void setEissn(String eissn) {
+		this.eissn = eissn;
+	}
+
+	public String getReviewProcess() {
+		return reviewProcess;
+	}
+
+	public void setReviewProcess(String reviewProcess) {
+		this.reviewProcess = reviewProcess;
+	}
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/model/EntityInfo.java

@@ -0,0 +1,68 @@
+
+package eu.dnetlib.dhp.oa.graph.hostedbymap.model;
+
+import java.io.Serializable;
+
+public class EntityInfo implements Serializable {
+	private String id;
+	private String journal_id;
+	private String name;
+	private Boolean openaccess;
+	private String hb_id;
+
+	public String getId() {
+		return id;
+	}
+
+	public void setId(String id) {
+		this.id = id;
+	}
+
+	public String getJournal_id() {
+		return journal_id;
+	}
+
+	public void setJournal_id(String journal_id) {
+		this.journal_id = journal_id;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public Boolean getOpenaccess() {
+		return openaccess;
+	}
+
+	public void setOpenaccess(Boolean openaccess) {
+		this.openaccess = openaccess;
+	}
+
+	public String getHb_id() {
+		return hb_id;
+	}
+
+	public void setHb_id(String hb_id) {
+		this.hb_id = hb_id;
+	}
+
+	public static EntityInfo newInstance(String id, String j_id, String name) {
+		return newInstance(id, j_id, name, false);
+
+	}
+
+	public static EntityInfo newInstance(String id, String j_id, String name, Boolean openaccess) {
+		EntityInfo pi = new EntityInfo();
+		pi.id = id;
+		pi.journal_id = j_id;
+		pi.name = name;
+		pi.openaccess = openaccess;
+		pi.hb_id = "";
+		return pi;
+
+	}
+}

dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/hostedbymap/model/UnibiGoldModel.java

@@ -0,0 +1,45 @@
+
+package eu.dnetlib.dhp.oa.graph.hostedbymap.model;
+
+import java.io.Serializable;
+
+import com.opencsv.bean.CsvBindByName;
+
+public class UnibiGoldModel implements Serializable {
+	@CsvBindByName(column = "ISSN")
+	private String issn;
+	@CsvBindByName(column = "ISSN_L")
+	private String issn_l;
+	@CsvBindByName(column = "TITLE")
+	private String title;
+	@CsvBindByName(column = "TITLE_SOURCE")
+	private String title_source;
+
+	public String getIssn() {
+		return issn;
+	}
+
+	public void setIssn(String issn) {
+		this.issn = issn;
+	}
+
+	public String getIssn_l() {
+		return issn_l;
+	}
+
+	public String getTitle() {
+		return title;
+	}
+
+	public void setTitle(String title) {
+		this.title = title;
+	}
+
+	public String getTitle_source() {
+		return title_source;
+	}
+
+	public void setTitle_source(String title_source) {
+		this.title_source = title_source;
+	}
+}

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/download_csv_parameters.json

@@ -0,0 +1,37 @@
+
+[
+
+  {
+    "paramName":"fu",
+    "paramLongName":"fileURL",
+    "paramDescription": "the url to download the csv file ",
+    "paramRequired": true
+  },
+
+  {
+    "paramName":"wp",
+    "paramLongName":"workingPath",
+    "paramDescription": "the path where to find the pre-processed data for unibi gold list and doj artciles",
+    "paramRequired": true
+  },
+  {
+    "paramName": "hnn",
+    "paramLongName": "hdfsNameNode",
+    "paramDescription": "the path used to store the HostedByMap",
+    "paramRequired": true
+  },
+  {
+    "paramName": "cfn",
+    "paramLongName": "classForName",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  },
+  {
+    "paramName": "sr",
+    "paramLongName": "replace",
+    "paramDescription": "true if the input file has to be cleaned before parsing",
+    "paramRequired": false
+  }
+]
+
+

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_apply_params.json

@@ -0,0 +1,36 @@
+
+[
+
+  {
+    "paramName":"pip",
+    "paramLongName":"preparedInfoPath",
+    "paramDescription": "the path where to find the pre-processed data for unibi gold list and doj artciles",
+    "paramRequired": true
+  },
+
+  {
+    "paramName": "ssm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": false
+  },
+  {
+    "paramName": "m",
+    "paramLongName": "master",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  },
+
+  {
+    "paramName": "gp",
+    "paramLongName": "graphPath",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  },
+  {
+    "paramName": "out",
+    "paramLongName": "outputPath",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_params.json

@@ -0,0 +1,38 @@
+
+[
+
+  {
+    "paramName":"dsp",
+    "paramLongName":"datasourcePath",
+    "paramDescription": "the path to the datasource ",
+    "paramRequired": true
+  },
+
+  {
+    "paramName":"wp",
+    "paramLongName":"workingPath",
+    "paramDescription": "the path where to find the pre-processed data for unibi gold list and doj artciles",
+    "paramRequired": true
+  },
+  {
+    "paramName": "out",
+    "paramLongName": "outputPath",
+    "paramDescription": "the path used to store the HostedByMap",
+    "paramRequired": true
+  },
+  {
+    "paramName": "ssm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": false
+  },
+  {
+    "paramName": "m",
+    "paramLongName": "master",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  }
+]
+
+
+

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedby_prepare_params.json

@@ -0,0 +1,37 @@
+
+[
+
+  {
+    "paramName":"hbmp",
+    "paramLongName":"hostedByMapPath",
+    "paramDescription": "the path to the datasource ",
+    "paramRequired": true
+  },
+
+  {
+    "paramName":"pip",
+    "paramLongName":"preparedInfoPath",
+    "paramDescription": "the path where to find the pre-processed data for unibi gold list and doj artciles",
+    "paramRequired": true
+  },
+
+  {
+    "paramName": "ssm",
+    "paramLongName": "isSparkSessionManaged",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": false
+  },
+  {
+    "paramName": "m",
+    "paramLongName": "master",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  },
+
+  {
+    "paramName": "gp",
+    "paramLongName": "graphPath",
+    "paramDescription": "true if the spark session is managed, false otherwise",
+    "paramRequired": true
+  }
+]

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/oozie_app/config-default.xml

@@ -0,0 +1,30 @@
+<configuration>
+    <property>
+        <name>jobTracker</name>
+        <value>yarnRM</value>
+    </property>
+    <property>
+        <name>nameNode</name>
+        <value>hdfs://nameservice1</value>
+    </property>
+    <property>
+        <name>oozie.use.system.libpath</name>
+        <value>true</value>
+    </property>
+    <property>
+        <name>hiveMetastoreUris</name>
+        <value>thrift://iis-cdh5-test-m3.ocean.icm.edu.pl:9083</value>
+    </property>
+    <property>
+        <name>hiveJdbcUrl</name>
+        <value>jdbc:hive2://iis-cdh5-test-m3.ocean.icm.edu.pl:10000</value>
+    </property>
+    <property>
+        <name>hiveDbName</name>
+        <value>openaire</value>
+    </property>
+    <property>
+        <name>oozie.launcher.mapreduce.user.classpath.first</name>
+        <value>true</value>
+    </property>
+</configuration>

dhp-workflows/dhp-graph-mapper/src/main/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/oozie_app/workflow.xml

@@ -0,0 +1,238 @@
+<workflow-app name="hosted_by_map" xmlns="uri:oozie:workflow:0.5">
+
+    <parameters>
+        <property>
+            <name>sourcePath</name>
+            <description>the source path</description>
+        </property>
+        <property>
+            <name>outputPath</name>
+            <description>the output path</description>
+        </property>
+        <property>
+            <name>hostedByMapPath</name>
+            <description>the output path</description>
+        </property>
+        <property>
+            <name>sparkDriverMemory</name>
+            <description>memory for driver process</description>
+        </property>
+        <property>
+            <name>sparkExecutorMemory</name>
+            <description>memory for individual executor</description>
+        </property>
+        <property>
+            <name>sparkExecutorCores</name>
+            <description>number of cores used by single executor</description>
+        </property>
+        <property>
+            <name>oozieActionShareLibForSpark2</name>
+            <description>oozie action sharelib for spark 2.*</description>
+        </property>
+        <property>
+            <name>spark2ExtraListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorAppListener</value>
+            <description>spark 2.* extra listeners classname</description>
+        </property>
+        <property>
+            <name>spark2SqlQueryExecutionListeners</name>
+            <value>com.cloudera.spark.lineage.NavigatorQueryListener</value>
+            <description>spark 2.* sql query execution listeners classname</description>
+        </property>
+        <property>
+            <name>spark2YarnHistoryServerAddress</name>
+            <description>spark 2.* yarn history server address</description>
+        </property>
+        <property>
+            <name>spark2EventLogDir</name>
+            <description>spark 2.* event log dir location</description>
+        </property>
+    </parameters>
+
+    <global>
+        <job-tracker>${jobTracker}</job-tracker>
+        <name-node>${nameNode}</name-node>
+        <configuration>
+            <property>
+                <name>mapreduce.job.queuename</name>
+                <value>${queueName}</value>
+            </property>
+            <property>
+                <name>oozie.launcher.mapred.job.queue.name</name>
+                <value>${oozieLauncherQueueName}</value>
+            </property>
+            <property>
+                <name>oozie.action.sharelib.for.spark</name>
+                <value>${oozieActionShareLibForSpark2}</value>
+            </property>
+
+        </configuration>
+    </global>
+
+    <start to="resume_from"/>
+
+    <decision name="resume_from">
+        <switch>
+            <case to="remove_hbmpath">${wf:conf('resumeFrom') eq 'ProduceHBM'}</case>
+            <default to="prepareInfo"/>
+        </switch>
+    </decision>
+
+
+    <kill name="Kill">
+        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+    </kill>
+
+    <action name="remove_hbmpath">
+        <fs>
+            <delete path="${hostedByMapPath}"/>
+<!--            <mkdir path="${hostedByMapPath}"/>-->
+        </fs>
+        <ok to="fork_downloads_csv"/>
+        <error to="Kill"/>
+    </action>
+
+
+    <fork name="fork_downloads_csv">
+        <path start="download_gold"/>
+        <path start="download_doaj"/>
+    </fork>
+
+    <action name="download_gold">
+        <java>
+            <main-class>eu.dnetlib.dhp.oa.graph.hostedbymap.GetCSV</main-class>
+            <arg>--hdfsNameNode</arg><arg>${nameNode}</arg>
+            <arg>--fileURL</arg><arg>${unibiFileURL}</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/unibi_gold</arg>
+            <arg>--classForName</arg><arg>eu.dnetlib.dhp.oa.graph.hostedbymap.model.UnibiGoldModel</arg>
+        </java>
+        <ok to="join_download"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="download_doaj">
+        <java>
+            <main-class>eu.dnetlib.dhp.oa.graph.hostedbymap.GetCSV</main-class>
+            <arg>--hdfsNameNode</arg><arg>${nameNode}</arg>
+            <arg>--fileURL</arg><arg>${doajFileURL}</arg>
+            <arg>--workingPath</arg><arg>${workingDir}/doaj</arg>
+            <arg>--classForName</arg><arg>eu.dnetlib.dhp.oa.graph.hostedbymap.model.DOAJModel</arg>
+            <arg>--replace</arg><arg>true</arg>
+        </java>
+        <ok to="join_download"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="join_download" to="produceHBM"/>
+
+    <action name="produceHBM">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn-cluster</master>
+            <name>Produce the new HostedByMap</name>
+            <class>eu.dnetlib.dhp.oa.graph.hostedbymap.SparkProduceHostedByMap</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-memory=${sparkExecutorMemory}
+                --executor-cores=${sparkExecutorCores}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.warehouse.dir=${sparkSqlWarehouseDir}
+            </spark-opts>
+            <arg>--datasourcePath</arg><arg>${sourcePath}/datasource</arg>
+            <arg>--workingPath</arg><arg>${workingDir}</arg>
+            <arg>--outputPath</arg><arg>${hostedByMapPath}</arg>
+            <arg>--master</arg><arg>yarn-cluster</arg>
+        </spark>
+        <ok to="prepareInfo"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="prepareInfo">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn-cluster</master>
+            <name>Prepare info to apply the hbm</name>
+            <class>eu.dnetlib.dhp.oa.graph.hostedbymap.SparkPrepareHostedByInfoToApply</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-memory=${sparkExecutorMemory}
+                --executor-cores=${sparkExecutorCores}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.warehouse.dir=${sparkSqlWarehouseDir}
+            </spark-opts>
+            <arg>--hostedByMapPath</arg><arg>${hostedByMapPath}</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo</arg>
+            <arg>--graphPath</arg><arg>${sourcePath}</arg>
+            <arg>--master</arg><arg>yarn-cluster</arg>
+        </spark>
+        <ok to="fork_apply"/>
+        <error to="Kill"/>
+    </action>
+
+    <fork name="fork_apply">
+        <path start="apply_result"/>
+        <path start="apply_datasource"/>
+    </fork>
+
+    <action name="apply_result">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn-cluster</master>
+            <name>Apply hbm to result</name>
+            <class>eu.dnetlib.dhp.oa.graph.hostedbymap.SparkApplyHostedByMapToResult</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-memory=${sparkExecutorMemory}
+                --executor-cores=${sparkExecutorCores}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.warehouse.dir=${sparkSqlWarehouseDir}
+            </spark-opts>
+            <arg>--outputPath</arg><arg>${outputPath}/publication</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo</arg>
+            <arg>--graphPath</arg><arg>${sourcePath}</arg>
+            <arg>--master</arg><arg>yarn-cluster</arg>
+        </spark>
+        <ok to="join_apply"/>
+        <error to="Kill"/>
+    </action>
+
+    <action name="apply_datasource">
+        <spark xmlns="uri:oozie:spark-action:0.2">
+            <master>yarn-cluster</master>
+            <name>Apply hbm to datasource</name>
+            <class>eu.dnetlib.dhp.oa.graph.hostedbymap.SparkApplyHostedByMapToDatasource</class>
+            <jar>dhp-graph-mapper-${projectVersion}.jar</jar>
+            <spark-opts>
+                --executor-memory=${sparkExecutorMemory}
+                --executor-cores=${sparkExecutorCores}
+                --driver-memory=${sparkDriverMemory}
+                --conf spark.extraListeners=${spark2ExtraListeners}
+                --conf spark.sql.queryExecutionListeners=${spark2SqlQueryExecutionListeners}
+                --conf spark.yarn.historyServer.address=${spark2YarnHistoryServerAddress}
+                --conf spark.eventLog.dir=${nameNode}${spark2EventLogDir}
+                --conf spark.sql.warehouse.dir=${sparkSqlWarehouseDir}
+            </spark-opts>
+            <arg>--outputPath</arg><arg>${outputPath}/datasource</arg>
+            <arg>--preparedInfoPath</arg><arg>${workingDir}/preparedInfo</arg>
+            <arg>--graphPath</arg><arg>${sourcePath}</arg>
+            <arg>--master</arg><arg>yarn-cluster</arg>
+        </spark>
+        <ok to="join_apply"/>
+        <error to="Kill"/>
+    </action>
+
+    <join name="join_apply" to="End"/>
+
+
+    <end name="End"/>
+
+</workflow-app>

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/hostedbymap/TestApply.scala

@@ -0,0 +1,138 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.EntityInfo
+import eu.dnetlib.dhp.schema.common.ModelConstants
+import eu.dnetlib.dhp.schema.oaf.{Datasource, OpenAccessRoute, Publication}
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Test
+
+class TestApply extends java.io.Serializable{
+
+  @Test
+  def testApplyOnResult (): Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val pub = getClass.getResource("publication.json").getPath
+    val hbm = getClass.getResource("preparedInfo.json").getPath
+
+    val mapper:ObjectMapper = new ObjectMapper()
+
+    implicit val mapEncoderDSInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+    implicit val mapEncoderPubInfo: Encoder[Publication] = Encoders.bean(classOf[Publication])
+
+
+    val pub_ds :Dataset[Publication] = spark.read.textFile(pub).map(p => mapper.readValue(p, classOf[Publication]))
+    val hbm_ds :Dataset[EntityInfo] = spark.read.textFile(hbm).map(p => mapper.readValue(p, classOf[EntityInfo]))
+
+
+    assertEquals(13, pub_ds.count())
+
+    val ds:Dataset[Publication] = SparkApplyHostedByMapToResult.applyHBtoPubs(hbm_ds, pub_ds)
+
+    assertEquals(13,  ds.count)
+
+    val temp: Dataset[(Publication, Publication)] = pub_ds.joinWith(ds, pub_ds.col("id").equalTo(ds.col("id")), "left")
+    assertEquals(13, temp.count())
+    temp.foreach(t2 => {
+      val pb : Publication = t2._1
+      val pa : Publication = t2._2
+      assertEquals(1, pa.getInstance().size())
+      assertEquals(1, pb.getInstance().size())
+      assertTrue(t2._1.getId.equals(t2._2.getId))
+      if(pb.getId.equals("50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9")){
+        assertTrue(pa.getInstance().get(0).getHostedby.getKey.equals("10|issn___print::e4b6d6d978f67520f6f37679a98c5735"))
+        assertTrue(pa.getInstance().get(0).getHostedby.getValue.equals("Academic Therapy"))
+        assertTrue(pa.getInstance().get(0).getAccessright.getClassid.equals("OPEN"))
+        assertTrue(pa.getInstance().get(0).getAccessright.getClassname.equals("Open Access"))
+        assertTrue(pa.getInstance().get(0).getAccessright.getOpenAccessRoute.equals(OpenAccessRoute.hybrid))
+        assertTrue(pa.getBestaccessright.getClassid.equals("OPEN"))
+        assertTrue(pa.getBestaccessright.getClassname.equals("Open Access"))
+
+
+        assertTrue(pb.getInstance().get(0).getHostedby.getKey.equals("10|openaire____::0b74b6a356bbf23c245f9ae9a748745c"))
+        assertTrue(pb.getInstance().get(0).getHostedby.getValue.equals("Revistas de investigación Universidad Nacional Mayor de San Marcos"))
+        assertTrue(pb.getInstance().get(0).getAccessright.getClassname.equals("not available"))
+        assertTrue(pb.getInstance().get(0).getAccessright.getClassid.equals("UNKNOWN"))
+        assertTrue(pb.getInstance().get(0).getAccessright.getOpenAccessRoute == null)
+        assertTrue(pb.getBestaccessright.getClassid.equals("UNKNOWN"))
+        assertTrue(pb.getBestaccessright.getClassname.equals("not available"))
+
+      }else{
+        assertTrue(pa.getInstance().get(0).getHostedby.getKey.equals(pb.getInstance().get(0).getHostedby.getKey))
+        assertTrue(pa.getInstance().get(0).getHostedby.getValue.equals(pb.getInstance().get(0).getHostedby.getValue))
+        assertTrue(pa.getInstance().get(0).getAccessright.getClassid.equals(pb.getInstance().get(0).getAccessright.getClassid))
+        assertTrue(pa.getInstance().get(0).getAccessright.getClassname.equals(pb.getInstance().get(0).getAccessright.getClassname))
+        assertTrue(pa.getInstance().get(0).getAccessright.getOpenAccessRoute == pb.getInstance().get(0).getAccessright.getOpenAccessRoute)
+
+      }
+    })
+
+    spark.close()
+  }
+
+
+  @Test
+  def testApplyOnDatasource():Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val dats = getClass.getResource("datasource.json").getPath
+    val hbm = getClass.getResource("preparedInfo2.json").getPath
+
+    val mapper:ObjectMapper = new ObjectMapper()
+
+    implicit val mapEncoderDSInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+    implicit val mapEncoderPubInfo: Encoder[Datasource] = Encoders.bean(classOf[Datasource])
+
+
+    val dats_ds :Dataset[Datasource] = spark.read.textFile(dats).map(p => mapper.readValue(p, classOf[Datasource]))
+    val hbm_ds :Dataset[EntityInfo] = Aggregators.datasourceToSingleId(spark.read.textFile(hbm).map(p => mapper.readValue(p, classOf[EntityInfo])))
+
+
+    assertEquals(10, dats_ds.count())
+
+    val ds:Dataset[Datasource] = SparkApplyHostedByMapToDatasource.applyHBtoDats(hbm_ds, dats_ds)
+
+    assertEquals(10,  ds.count)
+
+    val temp: Dataset[(Datasource, Datasource)] = dats_ds.joinWith(ds, dats_ds.col("id").equalTo(ds.col("id")), "left")
+    assertEquals(10, temp.count())
+    temp.foreach(t2 => {
+      val pb : Datasource = t2._1
+      val pa : Datasource = t2._2
+      assertTrue(t2._1.getId.equals(t2._2.getId))
+      if(pb.getId.equals("10|doajarticles::0ab37b7620eb9a73ac95d3ca4320c97d")){
+        assertTrue(pa.getOpenairecompatibility().getClassid.equals("hostedBy"))
+        assertTrue(pa.getOpenairecompatibility().getClassname.equals("collected from a compatible aggregator"))
+
+        assertTrue(pb.getOpenairecompatibility().getClassid.equals(ModelConstants.UNKNOWN))
+
+
+      }else{
+        assertTrue(pa.getOpenairecompatibility().getClassid.equals(pb.getOpenairecompatibility.getClassid))
+        assertTrue(pa.getOpenairecompatibility().getClassname.equals(pb.getOpenairecompatibility.getClassname))
+
+      }
+    })
+
+    spark.close()
+
+  }
+
+}

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/hostedbymap/TestPrepare.scala

@@ -0,0 +1,159 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import eu.dnetlib.dhp.oa.graph.hostedbymap.SparkPrepareHostedByInfoToApply.{joinResHBM, prepareResultInfo, toEntityInfo}
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.EntityInfo
+import eu.dnetlib.dhp.schema.oaf.{Datasource, OpenAccessRoute, Publication}
+import javax.management.openmbean.OpenMBeanAttributeInfo
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession}
+import org.json4s
+import org.json4s.DefaultFormats
+import eu.dnetlib.dhp.schema.common.ModelConstants
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.Test
+
+class TestPrepare extends java.io.Serializable{
+
+  def getString(input:HostedByItemType):String = {
+
+    import org.json4s.jackson.Serialization.write
+    implicit val formats = DefaultFormats
+
+    write(input)
+  }
+
+
+  @Test
+  def testHostedByMaptoEntityInfo() : Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val hbm = getClass.getResource("hostedbymap.json").getPath
+
+
+    import spark.implicits._
+
+    val mapper:ObjectMapper = new ObjectMapper()
+
+    implicit val mapEncoderDSInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+    val ds :Dataset[EntityInfo] = spark.createDataset(spark.sparkContext.textFile(hbm)).map(toEntityInfo)
+
+    ds.foreach(e => println(mapper.writeValueAsString(e)))
+
+    assertEquals(20,  ds.count)
+    spark.close()
+  }
+
+  @Test
+  def testPublicationtoEntityInfo() : Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val path = getClass.getResource("publication.json").getPath
+
+    val mapper:ObjectMapper = new ObjectMapper()
+
+    implicit val mapEncoderDSInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+    val ds :Dataset[EntityInfo] = prepareResultInfo(spark, path)
+
+    ds.foreach(e => println(mapper.writeValueAsString(e)))
+
+    assertEquals(2,  ds.count)
+
+    assertEquals("50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9", ds.filter(ei => ei.getJournal_id.equals("1728-5852")).first().getId)
+    assertEquals("50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9", ds.filter(ei => ei.getJournal_id.equals("0001-396X")).first().getId)
+
+    spark.close()
+  }
+
+  @Test
+  def testJoinResHBM (): Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val pub = getClass.getResource("iteminfofrompublication").getPath
+    val hbm = getClass.getResource("iteminfofromhostedbymap.json").getPath
+
+    val mapper:ObjectMapper = new ObjectMapper()
+
+    implicit val mapEncoderDSInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+    val pub_ds :Dataset[EntityInfo] = spark.read.textFile(pub).map(p => mapper.readValue(p, classOf[EntityInfo]))
+    val hbm_ds :Dataset[EntityInfo] = spark.read.textFile(hbm).map(p => mapper.readValue(p, classOf[EntityInfo]))
+
+    val ds: Dataset[EntityInfo] = joinResHBM(pub_ds, hbm_ds)
+
+    assertEquals(1,  ds.count)
+
+    val ei:EntityInfo = ds.first()
+
+    assertEquals("50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9", ei.getId)
+    assertEquals("10|issn___print::e4b6d6d978f67520f6f37679a98c5735", ei.getHb_id)
+    assertEquals("0001-396X", ei.getJournal_id)
+    assertEquals("Academic Therapy", ei.getName)
+    assertTrue(!ei.getOpenaccess)
+
+    spark.close()
+  }
+
+  @Test
+  def testJoinResHBM2 (): Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val pub = getClass.getResource("iteminfofrompublication2").getPath
+    val hbm = getClass.getResource("iteminfofromhostedbymap2.json").getPath
+
+    val mapper:ObjectMapper = new ObjectMapper()
+
+    implicit val mapEncoderDSInfo: Encoder[EntityInfo] = Encoders.bean(classOf[EntityInfo])
+
+    val pub_ds :Dataset[EntityInfo] = spark.read.textFile(pub).map(p => mapper.readValue(p, classOf[EntityInfo]))
+    val hbm_ds :Dataset[EntityInfo] = spark.read.textFile(hbm).map(p => mapper.readValue(p, classOf[EntityInfo]))
+
+    val ds: Dataset[EntityInfo] = joinResHBM(pub_ds, hbm_ds)
+
+    assertEquals(1,  ds.count)
+
+    val ei:EntityInfo = ds.first()
+
+    assertEquals("50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9", ei.getId)
+    assertEquals("10|issn___print::e4b6d6d978f67520f6f37679a98c5735", ei.getHb_id)
+    assertEquals("Academic Therapy", ei.getName)
+    assertTrue(ei.getOpenaccess)
+
+    ds.foreach(e => println(mapper.writeValueAsString(e)))
+
+    spark.close()
+  }
+
+
+
+}

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/hostedbymap/TestPreprocess.scala

@@ -0,0 +1,171 @@
+package eu.dnetlib.dhp.oa.graph.hostedbymap
+
+import eu.dnetlib.dhp.oa.graph.hostedbymap.{Aggregators, Constants, HostedByInfo, HostedByItemType, SparkProduceHostedByMap}
+import eu.dnetlib.dhp.schema.oaf.Datasource
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession}
+import org.json4s.DefaultFormats
+import org.junit.jupiter.api.Assertions.{assertNotNull, assertTrue}
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.Assertions._
+import org.json4s.jackson.Serialization.write
+
+class TestPreprocess extends java.io.Serializable{
+
+  implicit val mapEncoderDats: Encoder[Datasource] = Encoders.kryo[Datasource]
+  implicit val schema = Encoders.product[HostedByInfo]
+
+
+  def toHBIString (hbi:HostedByItemType): String = {
+    implicit val formats = DefaultFormats
+
+    write(hbi)
+  }
+
+  @Test
+  def readDatasource():Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val path = getClass.getResource("datasource.json").getPath
+
+    val ds :Dataset[HostedByItemType]= SparkProduceHostedByMap.oaHostedByDataset(spark, path)
+
+    assertEquals(9,  ds.count)
+
+    assertEquals(8, ds.filter(hbi => !hbi.issn.equals("")).count)
+    assertEquals(5, ds.filter(hbi => !hbi.eissn.equals("")).count)
+    assertEquals(0, ds.filter(hbi => !hbi.lissn.equals("")).count)
+
+    assertEquals(0, ds.filter(hbi => hbi.issn.equals("") && hbi.eissn.equals("") && hbi.lissn.equals("")).count)
+
+    assertTrue(ds.filter(hbi => hbi.issn.equals("0212-8365")).count == 1)
+    assertTrue(ds.filter(hbi => hbi.eissn.equals("2253-900X")).count == 1)
+    assertTrue(ds.filter(hbi => hbi.issn.equals("0212-8365") && hbi.eissn.equals("2253-900X")).count == 1)
+    assertTrue(ds.filter(hbi => hbi.issn.equals("0212-8365") && hbi.officialname.equals("Thémata")).count == 1)
+    assertTrue(ds.filter(hbi => hbi.issn.equals("0212-8365") && hbi.id.equals("10|doajarticles::abbc9265bea9ff62776a1c39785af00c")).count == 1)
+    ds.foreach(hbi => assertTrue(hbi.id.startsWith("10|")))
+    ds.foreach(hbi => println(toHBIString(hbi)))
+    spark.close()
+  }
+
+
+  @Test
+  def readGold():Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val path = getClass.getResource("unibi_transformed.json").getPath
+
+
+    val ds :Dataset[HostedByItemType]= SparkProduceHostedByMap.goldHostedByDataset(spark, path)
+
+    assertEquals(29, ds.count)
+
+    assertEquals(29, ds.filter(hbi => !hbi.issn.equals("")).count)
+    assertEquals(0, ds.filter(hbi => !hbi.eissn.equals("")).count)
+    assertEquals(29, ds.filter(hbi => !hbi.lissn.equals("")).count)
+
+    assertEquals(0, ds.filter(hbi => hbi.issn.equals("") && hbi.eissn.equals("") && hbi.lissn.equals("")).count)
+
+    assertTrue(ds.filter(hbi => hbi.issn.equals("2239-6101")).first().officialname.equals("European journal of sustainable development."))
+    assertTrue(ds.filter(hbi => hbi.issn.equals("2239-6101")).first().lissn.equals("2239-5938"))
+    assertTrue(ds.filter(hbi => hbi.issn.equals("2239-6101")).count == 1)
+    ds.foreach(hbi => assertTrue(hbi.id.equals(Constants.UNIBI)))
+    ds.foreach(hbi => println(toHBIString(hbi)))
+
+    spark.close()
+  }
+
+  @Test
+  def readDoaj():Unit = {
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+    val path = getClass.getResource("doaj_transformed.json").getPath
+
+    val ds :Dataset[HostedByItemType]= SparkProduceHostedByMap.doajHostedByDataset(spark, path)
+
+    assertEquals(25, ds.count)
+
+    assertEquals(14, ds.filter(hbi => !hbi.issn.equals("")).count)
+    assertEquals(21, ds.filter(hbi => !hbi.eissn.equals("")).count)
+    assertEquals(0, ds.filter(hbi => !hbi.lissn.equals("")).count)
+
+    assertEquals(0, ds.filter(hbi => hbi.issn.equals("") && hbi.eissn.equals("") && hbi.lissn.equals("")).count)
+
+    assertTrue(ds.filter(hbi => hbi.issn.equals("2077-3099")).first().officialname.equals("Journal of Space Technology"))
+    assertTrue(ds.filter(hbi => hbi.issn.equals("2077-3099")).first().eissn.equals("2411-5029"))
+    assertTrue(ds.filter(hbi => hbi.issn.equals("2077-3099")).count == 1)
+    assertTrue(ds.filter(hbi => hbi.eissn.equals("2077-2955")).first().issn.equals(""))
+    ds.foreach(hbi => assertTrue(hbi.id.equals(Constants.DOAJ)))
+    ds.foreach(hbi => println(toHBIString(hbi)))
+
+    spark.close()
+  }
+
+  @Test
+  def testAggregator() : Unit = {
+
+    val conf = new SparkConf()
+    conf.setMaster("local[*]")
+    conf.set("spark.driver.host", "localhost")
+    val spark: SparkSession =
+      SparkSession
+        .builder()
+        .appName(getClass.getSimpleName)
+        .config(conf)
+        .getOrCreate()
+
+
+    val tmp = SparkProduceHostedByMap.oaHostedByDataset(spark, getClass.getResource("datasource.json").getPath)
+      .union(SparkProduceHostedByMap.goldHostedByDataset(spark,getClass.getResource("unibi_transformed.json").getPath))
+      .union(SparkProduceHostedByMap.doajHostedByDataset(spark, getClass.getResource("doaj_transformed.json").getPath))
+      .flatMap(hbi => SparkProduceHostedByMap.toList(hbi))(Encoders.tuple(Encoders.STRING, Encoders.product[HostedByItemType]))
+
+    assertEquals(106, tmp.count)
+    assertEquals(82, tmp.map(i => i._1)(Encoders.STRING).distinct().count)
+
+
+    val ds :Dataset[(String, HostedByItemType)] = Aggregators.explodeHostedByItemType(SparkProduceHostedByMap.oaHostedByDataset(spark, getClass.getResource("datasource.json").getPath)
+      .union(SparkProduceHostedByMap.goldHostedByDataset(spark,getClass.getResource("unibi_transformed.json").getPath))
+      .union(SparkProduceHostedByMap.doajHostedByDataset(spark, getClass.getResource("doaj_transformed.json").getPath))
+      .flatMap(hbi => SparkProduceHostedByMap.toList(hbi))(Encoders.tuple(Encoders.STRING, Encoders.product[HostedByItemType])))
+
+    assertEquals(82,  ds.count)
+
+    assertEquals(13, ds.filter(i => i._2.id.startsWith("10|")).count)
+
+    assertTrue(ds.filter(i => i._1.equals("2077-3757")).first()._2.id.startsWith("10|"))
+    assertTrue(ds.filter(i => i._1.equals("2077-3757")).first()._2.openAccess)
+    assertEquals(1, ds.filter(i => i._1.equals("2077-3757")).count)
+
+    val hbmap : Dataset[String] = ds.filter(hbi => hbi._2.id.startsWith("10|")).map(SparkProduceHostedByMap.toHostedByMap)(Encoders.STRING)
+
+    hbmap.foreach(entry => println(entry))
+    spark.close()
+
+  }
+
+
+
+
+}

dhp-workflows/dhp-graph-mapper/src/test/java/eu/dnetlib/dhp/oa/graph/hostedbymap/TestReadCSV.java

@@ -0,0 +1,112 @@
+
+package eu.dnetlib.dhp.oa.graph.hostedbymap;
+
+import java.io.*;
+import java.net.URL;
+import java.net.URLConnection;
+import java.nio.charset.Charset;
+import java.util.List;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.opencsv.bean.CsvToBeanBuilder;
+
+import eu.dnetlib.dhp.oa.graph.hostedbymap.model.UnibiGoldModel;
+
+public class TestReadCSV {
+
+	@Test
+	public void testCSVUnibi() throws FileNotFoundException {
+
+		final String sourcePath = getClass()
+			.getResource("/eu/dnetlib/dhp/oa/graph/hostedbymap/unibiGold.csv")
+			.getPath();
+
+		List<UnibiGoldModel> beans = new CsvToBeanBuilder(new FileReader(sourcePath))
+			.withType(UnibiGoldModel.class)
+			.build()
+			.parse();
+
+		Assertions.assertEquals(36, beans.size());
+		Assertions.assertEquals(1, beans.stream().filter(e -> e.getIssn().equals("0001-625X")).count());
+		Assertions
+			.assertTrue(
+				beans
+					.stream()
+					.anyMatch(e -> e.getIssn().equals("0001-625X") && e.getTitle().equals("Acta Mycologica")));
+		Assertions.assertTrue(beans.stream().allMatch(e -> e.getIssn().equals(e.getIssn_l())));
+
+	}
+
+	@Disabled
+	@Test
+	public void testCSVUrlUnibi() throws IOException {
+
+		URL csv = new URL("https://pub.uni-bielefeld.de/download/2944717/2944718/issn_gold_oa_version_4.csv");
+
+		BufferedReader in = new BufferedReader(new InputStreamReader(csv.openStream()));
+		ObjectMapper mapper = new ObjectMapper();
+
+		new CsvToBeanBuilder(in)
+			.withType(eu.dnetlib.dhp.oa.graph.hostedbymap.model.UnibiGoldModel.class)
+			.build()
+			.parse()
+			.forEach(line ->
+
+			{
+				try {
+					System.out.println(mapper.writeValueAsString(line));
+				} catch (JsonProcessingException e) {
+					e.printStackTrace();
+				}
+			}
+
+			);
+	}
+
+	@Disabled
+	@Test
+	public void testCSVUrlDOAJ() throws IOException {
+
+		URLConnection connection = new URL("https://doaj.org/csv").openConnection();
+		connection
+			.setRequestProperty(
+				"User-Agent",
+				"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.95 Safari/537.11");
+		connection.connect();
+
+		BufferedReader in = new BufferedReader(
+			new InputStreamReader(connection.getInputStream(), Charset.forName("UTF-8")));
+		// BufferedReader in = new BufferedReader(new FileReader("/tmp/DOAJ.csv"));
+		PrintWriter writer = new PrintWriter(new BufferedWriter(new FileWriter("/tmp/DOAJ_1.csv")));
+		String line = null;
+		while ((line = in.readLine()) != null) {
+			writer.println(line.replace("\\\"", "\""));
+		}
+		writer.close();
+		in.close();
+		in = new BufferedReader(new FileReader("/tmp/DOAJ_1.csv"));
+		ObjectMapper mapper = new ObjectMapper();
+
+		new CsvToBeanBuilder(in)
+			.withType(eu.dnetlib.dhp.oa.graph.hostedbymap.model.DOAJModel.class)
+			.withMultilineLimit(1)
+			.build()
+			.parse()
+			.forEach(lline ->
+
+			{
+				try {
+					System.out.println(mapper.writeValueAsString(lline));
+				} catch (JsonProcessingException e) {
+					e.printStackTrace();
+				}
+			}
+
+			);
+	}
+}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/datasourceHostedByItem

@@ -0,0 +1,9 @@
+{"id":"10|doajarticles::0ab37b7620eb9a73ac95d3ca4320c97d","officialname":"Известия высших учебных заведений: Проблемы энергетики","issn":"1998-9903","eissn":"","lissn":"","openAccess":false}
+{"id":"10|doajarticles::abbc9265bea9ff62776a1c39785af00c","officialname":"Thémata","issn":"0212-8365","eissn":"2253-900X","lissn":"","openAccess":false}
+{"id":"10|issn___print::051e86306840dc8255d95c5671e97928","officialname":"Science Technology & Public Policy","issn":"2640-4613","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::4b2e7f05b6353940e5a7a592f2a87c94","officialname":"Cahiers d’études germaniques","issn":"0751-4239","eissn":"2605-8359","lissn":"","openAccess":false}
+{"id":"10|issn___print::4c950a72660642d69e767d1c2daad4a2","officialname":"Regional Economics Theory and Practice","issn":"2073-1477","eissn":"2311-8733","lissn":"","openAccess":false}
+{"id":"10|issn___print::9241f8ebd40dd55cbb179028b84ebb12","officialname":"Transplantation","issn":"0041-1337","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::982b4d2537d3f800b596fbec3dae0c7c","officialname":"International Journal of Operations Research and Information Systems","issn":"1947-9328","eissn":"1947-9336","lissn":"","openAccess":false}
+{"id":"10|issn___print::b9faf9c36c47169d4328e586eb62247c","officialname":"Bulletin of the British Mycological Society","issn":"0007-1528","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn__online::709e633c2ecf46396a4ed1b0096da1d0","officialname":"Journal of Technology and Innovation","issn":"","eissn":"2410-3993","lissn":"","openAccess":false}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/datasourceinfo.json

@@ -0,0 +1,9 @@
+{"id":"10|doajarticles::4f8b4cf7460320c0a80b6c6b64b3260f","officialname":"Известия высших учебных заведений: Проблемы энергетики","issn":"1998-9903","eissn":"","lissn":"","openAccess":false}
+{"id":"10|doajarticles::abbc9265bea9ff62776a1c39785af00c","officialname":"Thémata","issn":"0212-8365","eissn":"2253-900X","lissn":"","openAccess":false}
+{"id":"10|issn___print::051e86306840dc8255d95c5671e97928","officialname":"Science Technology & Public Policy","issn":"2077-3757","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","officialname":"Cahiers d’études germaniques","issn":"0751-4239","eissn":"2605-8359","lissn":"","openAccess":false}
+{"id":"10|issn___print::4c950a72660642d69e767d1c2daad4a2","officialname":"Regional Economics Theory and Practice","issn":"2073-1477","eissn":"2311-8733","lissn":"","openAccess":false}
+{"id":"10|issn___print::9241f8ebd40dd55cbb179028b84ebb12","officialname":"Transplantation","issn":"0041-1337","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::982b4d2537d3f800b596fbec3dae0c7c","officialname":"International Journal of Operations Research and Information Systems","issn":"1947-9328","eissn":"1947-9336","lissn":"","openAccess":false}
+{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","officialname":"Bulletin of the British Mycological Society","issn":"0007-1528","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::853ec7c7322ab252e0eca4d2840e7bd0","officialname":"Journal of Technology and Innovation","issn":"","eissn":"2410-3993","lissn":"","openAccess":false}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/doajHostedByItem

@@ -0,0 +1,25 @@
+{"id":"doaj","officialname":"Lëd i Sneg","issn":"2076-6734","eissn":"2412-3765","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Компьютерные исследования и моделирование","issn":"2076-7633","eissn":"2077-6853","lissn":"","openAccess":true}
+{"id":"doaj","officialname":" Историко-биологические исследования","issn":"2076-8176","eissn":"2500-1221","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Інформаційні технології і засоби навчання","issn":"2076-8184","eissn":"","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Revue Internationale de Pédagogie de l’Enseignement Supérieur","issn":"","eissn":"2076-8427","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Проблемы развития территории","issn":"2076-8915","eissn":"2409-9007","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Rambam Maimonides Medical Journal","issn":"","eissn":"2076-9172","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Membranes","issn":"2077-0375","eissn":"","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Journal of Clinical Medicine","issn":"","eissn":"2077-0383","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Agriculture","issn":"","eissn":"2077-0472","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Standartnye Obrazcy","issn":"2077-1177","eissn":"","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Металл и литье Украины","issn":"2077-1304","eissn":"2706-5529","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Journal of Marine Science and Engineering","issn":"","eissn":"2077-1312","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Religions","issn":"","eissn":"2077-1444","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"GW-Unterricht","issn":"2077-1517","eissn":"2414-4169","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"UCV-Scientia","issn":"2077-172X","eissn":"","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Sovremennye Issledovaniâ Socialʹnyh Problem","issn":"2077-1770","eissn":"2218-7405","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Granì","issn":"2077-1800","eissn":"2413-8738","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Journal of Economics Finance and Administrative Science","issn":"2077-1886","eissn":"2218-0648","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Science Education International","issn":"","eissn":"2077-2327","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Edumecentro","issn":"","eissn":"2077-2874","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Monteverdia","issn":"","eissn":"2077-2890","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Transformación","issn":"","eissn":"2077-2955","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Journal of Space Technology","issn":"2077-3099","eissn":"2411-5029","lissn":"","openAccess":true}
+{"id":"doaj","officialname":"Revue de Primatologie","issn":"","eissn":"2077-3757","lissn":"","openAccess":true}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/doaj_transformed.json

@@ -0,0 +1,25 @@
+{"journalTitle":"Lëd i Sneg","issn":"2076-6734","eissn":"2412-3765","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Компьютерные исследования и моделирование","issn":"2076-7633","eissn":"2077-6853","reviewProcess":"Blind peer review"}
+{"journalTitle":" Историко-биологические исследования","issn":"2076-8176","eissn":"2500-1221","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Інформаційні технології і засоби навчання","issn":"2076-8184","eissn":"","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Revue Internationale de Pédagogie de l’Enseignement Supérieur","issn":"","eissn":"2076-8427","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Проблемы развития территории","issn":"2076-8915","eissn":"2409-9007","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Rambam Maimonides Medical Journal","issn":"","eissn":"2076-9172","reviewProcess":"Peer review"}
+{"journalTitle":"Membranes","issn":"2077-0375","eissn":"","reviewProcess":"Blind peer review"}
+{"journalTitle":"Journal of Clinical Medicine","issn":"","eissn":"2077-0383","reviewProcess":"Blind peer review"}
+{"journalTitle":"Agriculture","issn":"","eissn":"2077-0472","reviewProcess":"Blind peer review"}
+{"journalTitle":"Standartnye Obrazcy","issn":"2077-1177","eissn":"","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Металл и литье Украины","issn":"2077-1304","eissn":"2706-5529","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Journal of Marine Science and Engineering","issn":"","eissn":"2077-1312","reviewProcess":"Blind peer review"}
+{"journalTitle":"Religions","issn":"","eissn":"2077-1444","reviewProcess":"Double blind peer review"}
+{"journalTitle":"GW-Unterricht","issn":"2077-1517","eissn":"2414-4169","reviewProcess":"Double blind peer review"}
+{"journalTitle":"UCV-Scientia","issn":"2077-172X","eissn":"","reviewProcess":"Peer review"}
+{"journalTitle":"Sovremennye Issledovaniâ Socialʹnyh Problem","issn":"2077-1770","eissn":"2218-7405","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Granì","issn":"2077-1800","eissn":"2413-8738","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Journal of Economics Finance and Administrative Science","issn":"2077-1886","eissn":"2218-0648","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Science Education International","issn":"","eissn":"2077-2327","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Edumecentro","issn":"","eissn":"2077-2874","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Monteverdia","issn":"","eissn":"2077-2890","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Transformación","issn":"","eissn":"2077-2955","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Journal of Space Technology","issn":"2077-3099","eissn":"2411-5029","reviewProcess":"Double blind peer review"}
+{"journalTitle":"Revue de Primatologie","issn":"","eissn":"2077-3757","reviewProcess":"Peer review"}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedbyitemtype.json

@@ -0,0 +1,17 @@
+{"id":"10|issn___print::e4b6d6d978f67520f6f37679a98c5735","officialname":"Academic Therapy","issn":"0001-396X","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","officialname":"Forschung im Ingenieurwesen","issn":"0015-7899","eissn":"1434-0860","lissn":"","openAccess":true}
+{"id":"10|issn___print::7977c16f0c47a3827536c7af137f6a81","officialname":"Review of Polarography","issn":"0034-6691","eissn":"1884-7692","lissn":"","openAccess":false}
+{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","officialname":"Land Economics","issn":"0023-7639","eissn":"1543-8325","lissn":"","openAccess":false}
+{"id":"10|issn___print::853ec7c7322ab252e0eca4d2840e7bd0","officialname":"Journal of Economic Entomology","issn":"0022-0493","eissn":"0022-0493","lissn":"","openAccess":false}
+{"id":"10|issn___print::480cbec18c06afa9bb7e0070948c97ff","officialname":"Brigham Young University science bulletin","issn":"0068-1024","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","officialname":"Journal of Contemporary Psychotherapy","issn":"0022-0116","eissn":"1573-3564","lissn":"","openAccess":false}
+{"id":"10|issn___print::cb21aba7985b1a0350abf99ee537302d","officialname":"Quarterly of Applied Mathematics","issn":"0033-569X","eissn":"1552-4485","lissn":"","openAccess":false}
+{"id":"10|issn___print::a10bce72f7ee20cae8fffc1a167d112f","officialname":"Revue de Synthèse","issn":"0035-1776","eissn":"1955-2343","lissn":"","openAccess":false}
+{"id":"10|issn___print::745f001e3f564f56a493dfea1faae501","officialname":"Journal of Statistical Physics","issn":"0022-4715","eissn":"1572-9613","lissn":"","openAccess":false}
+{"id":"10|issn___print::2a494a747066cafd64816e7495f32dc5","officialname":"Children s Literature in Education","issn":"0045-6713","eissn":"1573-1693","lissn":"","openAccess":false}
+{"id":"10|doajarticles::4f8b4cf7460320c0a80b6c6b64b3260f","officialname":"Slovenské divadlo","issn":"0037-699X","eissn":"1336-8605","lissn":"","openAccess":true}
+{"id":"10|issn___print::8cc8a1c0f0e11d4117014af5eccbbbb7","officialname":"Vistas in Astronomy","issn":"0083-6656","eissn":"","lissn":"","openAccess":false}
+{"id":"10|issn___print::a10bce72f7ee20cae8fffc1a167d112f","officialname":"Public Administration","issn":"0033-3298","eissn":"1467-9299","lissn":"","openAccess":false}
+{"id":"10|issn___print::55bb9eafabc7c310adb8bb0c336f2c26","officialname":"Memory & Cognition","issn":"0090-502X","eissn":"1532-5946","lissn":"","openAccess":false}
+{"id":"10|issn___print::dcde40f2d085cdf9c3a5b109d4978a9c","officialname":"Littérature","issn":"0047-4800","eissn":"1958-5926","lissn":"","openAccess":false}
+{"id":"10|issn___print::3c7f60a71f15ecc1611fbfe07509cd5c","officialname":"Proceedings of the Society for Analytical Chemistry","issn":"0037-9697","eissn":"","lissn":"","openAccess":false}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/hostedbymap.json

@@ -0,0 +1,20 @@
+{"0001-396X":{"id":"10|issn___print::e4b6d6d978f67520f6f37679a98c5735","officialname":"Academic Therapy","issn":"0001-396X","eissn":"","lissn":"","openAccess":false}}
+{"0015-7899":{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","officialname":"Forschung im Ingenieurwesen","issn":"0015-7899","eissn":"1434-0860","lissn":"","openAccess":false}}
+{"1434-0860":{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","officialname":"Forschung im Ingenieurwesen","issn":"0015-7899","eissn":"1434-0860","lissn":"","openAccess":true}}
+{"0022-0116":{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","officialname":"Journal of Contemporary Psychotherapy","issn":"0022-0116","eissn":"1573-3564","lissn":"","openAccess":false}}
+{"1573-3564":{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","officialname":"Journal of Contemporary Psychotherapy","issn":"0022-0116","eissn":"1573-3564","lissn":"","openAccess":false}}
+{"0022-0493":{"id":"10|issn___print::853ec7c7322ab252e0eca4d2840e7bd0","officialname":"Journal of Economic Entomology","issn":"0022-0493","eissn":"0022-0493","lissn":"","openAccess":false}}
+{"0022-4715":{"id":"10|issn___print::745f001e3f564f56a493dfea1faae501","officialname":"Journal of Statistical Physics","issn":"0022-4715","eissn":"1572-9613","lissn":"","openAccess":false}}
+{"1543-8325":{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","officialname":"Land Economics","issn":"0023-7639","eissn":"1543-8325","lissn":"","openAccess":false}}
+{"0023-7639":{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","officialname":"Land Economics","issn":"0023-7639","eissn":"1543-8325","lissn":"","openAccess":false}}
+{"0033-3298":{"id":"10|issn___print::91899e3872351895467856daeb798f63","officialname":"Public Administration","issn":"0033-3298","eissn":"1467-9299","lissn":"","openAccess":false}}
+{"0033-569X":{"id":"10|issn___print::cb21aba7985b1a0350abf99ee537302d","officialname":"Quarterly of Applied Mathematics","issn":"0033-569X","eissn":"1552-4485","lissn":"","openAccess":false}}
+{"0034-6691":{"id":"10|issn___print::7977c16f0c47a3827536c7af137f6a81","officialname":"Review of Polarography","issn":"0034-6691","eissn":"1884-7692","lissn":"","openAccess":false}}
+{"0035-1776":{"id":"10|issn___print::a10bce72f7ee20cae8fffc1a167d112f","officialname":"Revue de Synthèse","issn":"0035-1776","eissn":"1955-2343","lissn":"","openAccess":false}}
+{"0037-699X":{"id":"10|doajarticles::4f8b4cf7460320c0a80b6c6b64b3260f","officialname":"Slovenské divadlo","issn":"0037-699X","eissn":"1336-8605","lissn":"","openAccess":true}}
+{"0037-9697":{"id":"10|issn___print::3c7f60a71f15ecc1611fbfe07509cd5c","officialname":"Proceedings of the Society for Analytical Chemistry","issn":"0037-9697","eissn":"","lissn":"","openAccess":false}}
+{"0045-6713":{"id":"10|issn___print::2a494a747066cafd64816e7495f32dc5","officialname":"Children s Literature in Education","issn":"0045-6713","eissn":"1573-1693","lissn":"","openAccess":false}}
+{"0047-4800":{"id":"10|issn___print::dcde40f2d085cdf9c3a5b109d4978a9c","officialname":"Littérature","issn":"0047-4800","eissn":"1958-5926","lissn":"","openAccess":false}}
+{"0068-1024":{"id":"10|issn___print::480cbec18c06afa9bb7e0070948c97ff","officialname":"Brigham Young University science bulletin","issn":"0068-1024","eissn":"","lissn":"","openAccess":false}}
+{"0083-6656":{"id":"10|issn___print::8cc8a1c0f0e11d4117014af5eccbbbb7","officialname":"Vistas in Astronomy","issn":"0083-6656","eissn":"","lissn":"","openAccess":false}}
+{"0090-502X":{"id":"10|issn___print::55bb9eafabc7c310adb8bb0c336f2c26","officialname":"Memory & Cognition","issn":"0090-502X","eissn":"1532-5946","lissn":"","openAccess":false}}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/iteminfofromhostedbymap.json

@@ -0,0 +1,20 @@
+{"id":"10|issn___print::e4b6d6d978f67520f6f37679a98c5735","journal_id":"0001-396X","name":"Academic Therapy","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::cb21aba7985b1a0350abf99ee537302d","journal_id":"0033-569X","name":"Quarterly of Applied Mathematics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","journal_id":"0015-7899","name":"Forschung im Ingenieurwesen","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::7977c16f0c47a3827536c7af137f6a81","journal_id":"0034-6691","name":"Review of Polarography","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","journal_id":"1434-0860","name":"Forschung im Ingenieurwesen","openaccess":true,"hb_id":""}
+{"id":"10|issn___print::a10bce72f7ee20cae8fffc1a167d112f","journal_id":"0035-1776","name":"Revue de Synthèse","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","journal_id":"0022-0116","name":"Journal of Contemporary Psychotherapy","openaccess":false,"hb_id":""}
+{"id":"10|doajarticles::4f8b4cf7460320c0a80b6c6b64b3260f","journal_id":"0037-699X","name":"Slovenské divadlo","openaccess":true,"hb_id":""}
+{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","journal_id":"1573-3564","name":"Journal of Contemporary Psychotherapy","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::3c7f60a71f15ecc1611fbfe07509cd5c","journal_id":"0037-9697","name":"Proceedings of the Society for Analytical Chemistry","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::853ec7c7322ab252e0eca4d2840e7bd0","journal_id":"0022-0493","name":"Journal of Economic Entomology","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::2a494a747066cafd64816e7495f32dc5","journal_id":"0045-6713","name":"Children s Literature in Education","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::745f001e3f564f56a493dfea1faae501","journal_id":"0022-4715","name":"Journal of Statistical Physics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::dcde40f2d085cdf9c3a5b109d4978a9c","journal_id":"0047-4800","name":"Littérature","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","journal_id":"1543-8325","name":"Land Economics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::480cbec18c06afa9bb7e0070948c97ff","journal_id":"0068-1024","name":"Brigham Young University science bulletin","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","journal_id":"0023-7639","name":"Land Economics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::8cc8a1c0f0e11d4117014af5eccbbbb7","journal_id":"0083-6656","name":"Vistas in Astronomy","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::91899e3872351895467856daeb798f63","journal_id":"0033-3298","name":"Public Administration","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::55bb9eafabc7c310adb8bb0c336f2c26","journal_id":"0090-502X","name":"Memory & Cognition","openaccess":false,"hb_id":""}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/iteminfofromhostedbymap2.json

@@ -0,0 +1,20 @@
+{"id":"10|issn___print::e4b6d6d978f67520f6f37679a98c5735","journal_id":"0001-396X","name":"Academic Therapy","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::cb21aba7985b1a0350abf99ee537302d","journal_id":"0033-569X","name":"Quarterly of Applied Mathematics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::4b5605a395a243e12c95c1ecb8365107","journal_id":"0015-7899","name":"Forschung im Ingenieurwesen","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::7977c16f0c47a3827536c7af137f6a81","journal_id":"0034-6691","name":"Review of Polarography","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::e4b6d6d978f67520f6f37679a98c5735","journal_id":"1434-0860","name":"Academic Therapy","openaccess":true,"hb_id":""}
+{"id":"10|issn___print::a10bce72f7ee20cae8fffc1a167d112f","journal_id":"0035-1776","name":"Revue de Synthèse","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","journal_id":"0022-0116","name":"Journal of Contemporary Psychotherapy","openaccess":false,"hb_id":""}
+{"id":"10|doajarticles::4f8b4cf7460320c0a80b6c6b64b3260f","journal_id":"0037-699X","name":"Slovenské divadlo","openaccess":true,"hb_id":""}
+{"id":"10|issn___print::a4e08f7b862090b3f07e574e0159ff70","journal_id":"1573-3564","name":"Journal of Contemporary Psychotherapy","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::3c7f60a71f15ecc1611fbfe07509cd5c","journal_id":"0037-9697","name":"Proceedings of the Society for Analytical Chemistry","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::853ec7c7322ab252e0eca4d2840e7bd0","journal_id":"0022-0493","name":"Journal of Economic Entomology","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::2a494a747066cafd64816e7495f32dc5","journal_id":"0045-6713","name":"Children s Literature in Education","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::745f001e3f564f56a493dfea1faae501","journal_id":"0022-4715","name":"Journal of Statistical Physics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::dcde40f2d085cdf9c3a5b109d4978a9c","journal_id":"0047-4800","name":"Littérature","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","journal_id":"1543-8325","name":"Land Economics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::480cbec18c06afa9bb7e0070948c97ff","journal_id":"0068-1024","name":"Brigham Young University science bulletin","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::1aea1dc1fbc3153111099750884dc4e8","journal_id":"0023-7639","name":"Land Economics","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::8cc8a1c0f0e11d4117014af5eccbbbb7","journal_id":"0083-6656","name":"Vistas in Astronomy","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::91899e3872351895467856daeb798f63","journal_id":"0033-3298","name":"Public Administration","openaccess":false,"hb_id":""}
+{"id":"10|issn___print::55bb9eafabc7c310adb8bb0c336f2c26","journal_id":"0090-502X","name":"Memory & Cognition","openaccess":false,"hb_id":""}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/iteminfofrompublication

@@ -0,0 +1,2 @@
+{"id":"50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9","journal_id":"1728-5852","name":"","openaccess":false,"hb_id":""}
+{"id":"50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9","journal_id":"0001-396X","name":"","openaccess":false,"hb_id":""}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/iteminfofrompublication2

@@ -0,0 +1,2 @@
+{"id":"50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9","journal_id":"1434-0860","name":"","openaccess":false,"hb_id":""}
+{"id":"50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9","journal_id":"0001-396X","name":"","openaccess":false,"hb_id":""}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/preparedInfo.json

@@ -0,0 +1 @@
+{"id":"50|4dc99724cf04::ed1ba83e1add6ce292433729acd8b0d9","journal_id":"1434-0860","name":"Academic Therapy","openaccess":true,"hb_id":"10|issn___print::e4b6d6d978f67520f6f37679a98c5735"}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/preparedInfo2.json

@@ -0,0 +1,3 @@
+{"id":"pubid","journal_id":"issn","name":"ds_name","openaccess":true,"hb_id":"10|doajarticles::0ab37b7620eb9a73ac95d3ca4320c97d"}
+{"id":"pubid","journal_id":"issn","name":"ds_name","openaccess":true,"hb_id":"10|doajarticles::0ab37b7620eb9a73ac95d3ca4320c97d"}
+{"id":"pubid","journal_id":"issn","name":"ds_name","openaccess":true,"hb_id":"10|doajarticles::abbc9265bea9ff62776a1c39785af00c"}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/unibiGold.csv

@@ -0,0 +1,37 @@
+"ISSN","ISSN_L","ISSN_IN_DOAJ","ISSN_IN_ROAD","ISSN_IN_PMC","ISSN_IN_OAPC","ISSN_IN_WOS","ISSN_IN_SCOPUS","JOURNAL_IN_DOAJ","JOURNAL_IN_ROAD","JOURNAL_IN_PMC","JOURNAL_IN_OAPC","JOURNAL_IN_WOS","JOURNAL_IN_SCOPUS","TITLE","TITLE_SOURCE"
+"0001-625X","0001-625X",1,1,0,0,0,1,1,1,0,0,0,1,"Acta Mycologica","DOAJ"
+"0002-0397","0002-0397",1,1,0,0,1,1,1,1,0,0,1,1,"Africa Spectrum","DOAJ"
+"0003-2565","0003-2565",1,0,0,0,0,0,1,0,0,0,0,0,"Anali Pravnog Fakulteta u Beogradu","DOAJ"
+"0003-424X","0003-424X",0,1,0,0,1,0,0,1,0,0,1,0,"Annales de zootechnie.","ROAD"
+"0003-4827","0003-4827",0,1,0,0,0,1,0,1,0,0,0,1,"Annals of Iowa.","ROAD"
+"0004-0592","0004-0592",1,1,0,0,1,1,1,1,0,0,1,1,"Archivos de Zootecnia","DOAJ"
+"0004-282X","0004-282X",1,1,0,0,1,1,1,1,0,0,1,1,"Arquivos de Neuro-Psiquiatria","DOAJ"
+"0006-3096","0006-3096",0,1,0,0,0,0,0,1,0,0,0,0,"Biologia.","ROAD"
+"0006-8705","0006-8705",1,1,0,0,1,1,1,1,0,0,1,1,"Bragantia","DOAJ"
+"0007-5124","0007-5124",0,1,0,0,1,0,0,1,1,0,1,1,"Experimental animals.","ROAD"
+"0007-9502","0007-9502",0,1,0,0,0,0,0,1,0,0,0,0,"Caesaraugusta.","ROAD"
+"0008-7386","0008-7386",1,1,0,0,0,1,1,1,0,0,0,1,"Časopis pro Moderní Filologii","DOAJ"
+"0008-7629","0008-7629",1,0,0,0,0,0,1,0,0,0,0,0,"Catalogue and Index","DOAJ"
+"0015-573X","0015-573X",0,1,0,0,0,0,0,1,0,0,0,0,"Folia quaternaria.","ROAD"
+"0016-6987","0016-6987",1,0,0,0,1,1,1,0,0,0,1,1,"Genus","DOAJ"
+"0016-7789","0016-7789",1,1,0,0,0,1,1,1,0,0,0,1,"Geologija ","DOAJ"
+"0021-5007","0021-5007",0,1,0,0,0,1,0,1,0,0,0,1,"Nihon Seitai Gakkaishi.","ROAD"
+"0023-4001","0023-4001",0,1,0,0,1,1,0,1,0,0,1,1,"Korean Journal of Parasitology","ROAD"
+"0023-5415","0023-5415",1,1,0,0,0,0,1,1,0,0,0,0,"Kunst og Kultur","DOAJ"
+"0026-1165","0026-1165",1,0,0,0,1,1,1,0,0,0,1,1,"Journal of the Meteorological Society of Japan","DOAJ"
+"0029-0181","0029-0181",0,1,0,0,0,0,0,1,0,0,0,0,"Nihon butsuri gakkaishi.","ROAD"
+"0034-7000","0034-7000",1,1,0,0,0,1,1,1,0,0,0,1,"Revista Argentina de Cardiología","DOAJ"
+"0034-7523","0034-7523",0,1,0,0,0,1,0,1,0,0,0,1,"Revista cubana de medicina.","ROAD"
+"0034-8244","0034-8244",1,0,0,0,1,1,1,0,0,0,1,1,"Revista de Filosofia","DOAJ"
+"0034-8678","0034-8678",1,0,0,0,0,0,1,0,0,0,0,0,"Revista de Pedagogie","DOAJ"
+"0036-8709","0036-8709",1,1,1,0,1,1,1,1,1,0,1,1,"Scientia Pharmaceutica","DOAJ"
+"0044-4855","0044-4855",0,1,0,0,0,0,0,1,0,0,0,0,"Život i škola.","ROAD"
+"0048-7449","0048-7449",1,1,0,0,1,1,1,1,0,0,1,1,"Reumatismo","DOAJ"
+"0048-766X","0048-766X",0,1,0,0,0,1,0,1,0,0,0,1,"Revista chilena de obstetricia y ginecología.","ROAD"
+"0065-1400","0065-1400",0,1,0,0,1,1,0,1,0,0,1,1,"Acta Neurobiologiae Experimentalis.","ROAD"
+"0066-6742","0066-6742",1,0,0,0,1,1,1,0,0,0,1,1,"Archivo Español de Arqueología","DOAJ"
+"0073-2435","0073-2435",1,1,0,0,1,1,1,1,0,0,1,1,"Historia (Santiago)","DOAJ"
+"0073-4918","0073-4918",0,1,0,0,0,0,0,1,0,0,0,0,"Illinois Natural History Survey bulletin.","ROAD"
+"0075-7411","0075-7411",1,0,0,0,0,0,1,0,0,0,0,0,"Anales","DOAJ"
+"0077-2704","0077-2704",0,1,0,0,0,0,0,1,0,0,0,0,"Namn och bygd.","ROAD"
+"0078-5466","0078-5466",0,1,0,0,1,1,0,1,0,0,1,1,"Optica Applicata.","ROAD"

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/unibi_transformed.json

@@ -0,0 +1,29 @@
+{"issn":"2502-731X","issn_l":"2502-731X","title":"JIMKESMAS (Jurnal Ilmiah Mahasiswa Kesehatan Masyarakat)","title_source":"ROAD"}
+{"issn":"2502-7409","issn_l":"1411-0253","title":"Jurnal ilmu informasi, perpustakaan, dan kearsipan","title_source":"ROAD"}
+{"issn":"2502-7433","issn_l":"2502-7433","title":"At-Tadbir : jurnal ilmiah manajemen","title_source":"ROAD"}
+{"issn":"2502-745X","issn_l":"2502-745X","title":"Jurnal Kesehatan Panrita Husada.","title_source":"ROAD"}
+{"issn":"2502-7549","issn_l":"2502-7549","title":"ELang journal (An English Education journal)","title_source":"ROAD"}
+{"issn":"2423-3633","issn_l":"2423-3625","title":"̒Ulūm-i darmāngāhī-i dāmpizishkī-i Īrān.","title_source":"ROAD"}
+{"issn":"2423-5563","issn_l":"2423-3773","title":"Pizhūhishnāmah-i  ̒ilm/sanjī.","title_source":"ROAD"}
+{"issn":"1735-434X","issn_l":"1735-434X","title":"Iranian journal of animal biosystematics.","title_source":"ROAD"}
+{"issn":"2423-4435","issn_l":"2008-6113","title":"Majallah-i jangal-i Īrān.","title_source":"ROAD"}
+{"issn":"2423-4575","issn_l":"2423-4575","title":"Ābziyān-i zinatī.","title_source":"ROAD"}
+{"issn":"2423-4974","issn_l":"2423-4974","title":"Pizhūhishnāmah-i ravābiṭ-i  biyn/al- milal.","title_source":"ROAD"}
+{"issn":"2380-0607","issn_l":"2380-0607","title":"AIHM journal club.","title_source":"ROAD"}
+{"issn":"1085-4568","issn_l":"1085-4568","title":"Frontiers.","title_source":"ROAD"}
+{"issn":"2380-8845","issn_l":"2380-8845","title":"˜The œjournal of contemporary archival studies.","title_source":"ROAD"}
+{"issn":"2381-1803","issn_l":"2381-1803","title":"International journal of complementary & alternative medicine.","title_source":"ROAD"}
+{"issn":"2381-2478","issn_l":"2381-2478","title":"Palapala.","title_source":"ROAD"}
+{"issn":"2382-5170","issn_l":"2382-5170","title":"Asia pacific journal of environment ecology and sustainable development.","title_source":"ROAD"}
+{"issn":"2382-9737","issn_l":"2382-9737","title":"Majallah-i salāmat va bihdāsht","title_source":"ROAD"}
+{"issn":"2382-977X","issn_l":"2382-977X","title":"UCT journal of research in science ,engineering and technology","title_source":"ROAD"}
+{"issn":"2382-9974","issn_l":"2382-9974","title":"Bih/nizhādī-i giyāhān-i zirā̒ī va bāghī.","title_source":"ROAD"}
+{"issn":"2227-4782","issn_l":"2227-4782","title":"Problemi endokrinnoï patologìï.","title_source":"ROAD"}
+{"issn":"2685-0079","issn_l":"2597-4971","title":"Jurnal Kebijakan Pembangunan Daerah : Jurnal Penelitian dan Pengembangan Kebijakan Pembangunan Daerah.","title_source":"ROAD"}
+{"issn":"2574-0075","issn_l":"2574-0075","title":"Hypermedia magazine.","title_source":"ROAD"}
+{"issn":"2574-0296","issn_l":"2574-0296","title":"˜The œmuseum review.","title_source":"ROAD"}
+{"issn":"2574-0334","issn_l":"2574-0334","title":"Bioactive compounds in health and disease.","title_source":"ROAD"}
+{"issn":"2574-108X","issn_l":"2574-108X","title":"Journal of computer science integration.","title_source":"ROAD"}
+{"issn":"2574-254X","issn_l":"2574-254X","title":"Child and adolescent obesity.","title_source":"ROAD"}
+{"issn":"2574-3325","issn_l":"2574-3325","title":"Journal of research on the college president.","title_source":"ROAD"}
+{"issn":"2239-6101","issn_l":"2239-5938","title":"European journal of sustainable development.","title_source":"ROAD"}

dhp-workflows/dhp-graph-mapper/src/test/resources/eu/dnetlib/dhp/oa/graph/hostedbymap/unibyHostedByItem

@@ -0,0 +1,29 @@
+{"id":"unibi","officialname":"JIMKESMAS (Jurnal Ilmiah Mahasiswa Kesehatan Masyarakat)","issn":"2502-731X","eissn":"","lissn":"2502-731X","openAccess":true}
+{"id":"unibi","officialname":"Jurnal ilmu informasi, perpustakaan, dan kearsipan","issn":"2502-7409","eissn":"","lissn":"1411-0253","openAccess":true}
+{"id":"unibi","officialname":"At-Tadbir : jurnal ilmiah manajemen","issn":"2502-7433","eissn":"","lissn":"2502-7433","openAccess":true}
+{"id":"unibi","officialname":"Jurnal Kesehatan Panrita Husada.","issn":"2502-745X","eissn":"","lissn":"2502-745X","openAccess":true}
+{"id":"unibi","officialname":"ELang journal (An English Education journal)","issn":"2502-7549","eissn":"","lissn":"2502-7549","openAccess":true}
+{"id":"unibi","officialname":"̒Ulūm-i darmāngāhī-i dāmpizishkī-i Īrān.","issn":"2423-3633","eissn":"","lissn":"2423-3625","openAccess":true}
+{"id":"unibi","officialname":"Pizhūhishnāmah-i  ̒ilm/sanjī.","issn":"2423-5563","eissn":"","lissn":"2423-3773","openAccess":true}
+{"id":"unibi","officialname":"Iranian journal of animal biosystematics.","issn":"1735-434X","eissn":"","lissn":"1735-434X","openAccess":true}
+{"id":"unibi","officialname":"Majallah-i jangal-i Īrān.","issn":"2423-4435","eissn":"","lissn":"2008-6113","openAccess":true}
+{"id":"unibi","officialname":"Ābziyān-i zinatī.","issn":"2423-4575","eissn":"","lissn":"2423-4575","openAccess":true}
+{"id":"unibi","officialname":"Pizhūhishnāmah-i ravābiṭ-i  biyn/al- milal.","issn":"2423-4974","eissn":"","lissn":"2423-4974","openAccess":true}
+{"id":"unibi","officialname":"AIHM journal club.","issn":"2380-0607","eissn":"","lissn":"2380-0607","openAccess":true}
+{"id":"unibi","officialname":"Frontiers.","issn":"1085-4568","eissn":"","lissn":"1085-4568","openAccess":true}
+{"id":"unibi","officialname":"˜The œjournal of contemporary archival studies.","issn":"2380-8845","eissn":"","lissn":"2380-8845","openAccess":true}
+{"id":"unibi","officialname":"International journal of complementary & alternative medicine.","issn":"2381-1803","eissn":"","lissn":"2381-1803","openAccess":true}
+{"id":"unibi","officialname":"Palapala.","issn":"2381-2478","eissn":"","lissn":"2381-2478","openAccess":true}
+{"id":"unibi","officialname":"Asia pacific journal of environment ecology and sustainable development.","issn":"2382-5170","eissn":"","lissn":"2382-5170","openAccess":true}
+{"id":"unibi","officialname":"Majallah-i salāmat va bihdāsht","issn":"2382-9737","eissn":"","lissn":"2382-9737","openAccess":true}
+{"id":"unibi","officialname":"UCT journal of research in science ,engineering and technology","issn":"2382-977X","eissn":"","lissn":"2382-977X","openAccess":true}
+{"id":"unibi","officialname":"Bih/nizhādī-i giyāhān-i zirā̒ī va bāghī.","issn":"2382-9974","eissn":"","lissn":"2382-9974","openAccess":true}
+{"id":"unibi","officialname":"Problemi endokrinnoï patologìï.","issn":"2227-4782","eissn":"","lissn":"2227-4782","openAccess":true}
+{"id":"unibi","officialname":"Jurnal Kebijakan Pembangunan Daerah : Jurnal Penelitian dan Pengembangan Kebijakan Pembangunan Daerah.","issn":"2685-0079","eissn":"","lissn":"2597-4971","openAccess":true}
+{"id":"unibi","officialname":"Hypermedia magazine.","issn":"2574-0075","eissn":"","lissn":"2574-0075","openAccess":true}
+{"id":"unibi","officialname":"˜The œmuseum review.","issn":"2574-0296","eissn":"","lissn":"2574-0296","openAccess":true}
+{"id":"unibi","officialname":"Bioactive compounds in health and disease.","issn":"2574-0334","eissn":"","lissn":"2574-0334","openAccess":true}
+{"id":"unibi","officialname":"Journal of computer science integration.","issn":"2574-108X","eissn":"","lissn":"2574-108X","openAccess":true}
+{"id":"unibi","officialname":"Child and adolescent obesity.","issn":"2574-254X","eissn":"","lissn":"2574-254X","openAccess":true}
+{"id":"unibi","officialname":"Journal of research on the college president.","issn":"2574-3325","eissn":"","lissn":"2574-3325","openAccess":true}
+{"id":"unibi","officialname":"European journal of sustainable development.","issn":"2239-6101","eissn":"","lissn":"2239-5938","openAccess":true}

pom.xml

@@ -519,6 +519,12 @@
 				<version>${common.text.version}</version>
 			</dependency>
 
+			<dependency>
+				<groupId>com.opencsv</groupId>
+				<artifactId>opencsv</artifactId>
+				<version>5.5</version>
+			</dependency>
+
 		</dependencies>
 	</dependencyManagement>