This commit is contained in:
Miriam Baglioni 2020-06-11 10:49:01 +02:00
parent bb9f21d0e7
commit db27663750
3 changed files with 53 additions and 41 deletions

View File

@ -1,26 +1,11 @@
package eu.dnetlib.dhp.oa.graph.dump; package eu.dnetlib.dhp.oa.graph.dump;
import com.google.common.base.Joiner;
import com.google.common.collect.Maps;
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.utils.ISLookupClientFactory; import eu.dnetlib.dhp.utils.ISLookupClientFactory;
import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException; import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService; import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.Node;
import org.dom4j.io.SAXReader;
import java.io.StringReader;
import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map;
public class QueryInformationSystem { public class QueryInformationSystem {
private static final String XQUERY = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') " + private static final String XQUERY = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') " +
@ -33,27 +18,10 @@ public class QueryInformationSystem {
public static Map<String,String> getCommunityMap(final String isLookupUrl) public List<String> getCommunityMap(final String isLookupUrl)
throws ISLookUpException, DocumentException { throws ISLookUpException {
ISLookUpService isLookUp = ISLookupClientFactory.getLookUpService(isLookupUrl); ISLookUpService isLookUp = ISLookupClientFactory.getLookUpService(isLookupUrl);
final List<String> res = isLookUp.quickSearchProfile(XQUERY); return isLookUp.quickSearchProfile(XQUERY);
final Map<String, String> communityMap = new HashMap<>();
res.stream().forEach(xml -> {
final Document doc;
try {
doc = new SAXReader().read(new StringReader(xml));
Element root = doc.getRootElement();
communityMap.put(root.attribute("id").getValue(), root.attribute("label").getValue());
} catch (DocumentException e) {
e.printStackTrace();
}
});
return communityMap;
} }

View File

@ -3,29 +3,35 @@ package eu.dnetlib.dhp.oa.graph.dump;
import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession; import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
import java.io.Serializable; import java.io.Serializable;
import java.io.StringReader;
import java.util.*; import java.util.*;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import eu.dnetlib.dhp.schema.oaf.Context; import eu.dnetlib.dhp.schema.oaf.Context;
import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpException;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.spark.SparkConf; import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SaveMode; import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.SparkSession;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import eu.dnetlib.dhp.application.ArgumentApplicationParser; import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.schema.oaf.Result; import eu.dnetlib.dhp.schema.oaf.Result;
import javax.management.Query;
public class SparkDumpCommunityProducts implements Serializable { public class SparkDumpCommunityProducts implements Serializable {
private static final Logger log = LoggerFactory.getLogger(SparkDumpCommunityProducts.class); private static final Logger log = LoggerFactory.getLogger(SparkDumpCommunityProducts.class);
public static void main(String[] args) throws Exception { public static void main(String[] args) throws Exception {
String jsonConfiguration = IOUtils String jsonConfiguration = IOUtils
.toString( .toString(
@ -52,7 +58,7 @@ public class SparkDumpCommunityProducts implements Serializable {
final String resultClassName = parser.get("resultTableName"); final String resultClassName = parser.get("resultTableName");
log.info("resultTableName: {}", resultClassName); log.info("resultTableName: {}", resultClassName);
final String dumpClassName = parser.get("dumpClassName"); final String dumpClassName = parser.get("dumpTableName");
log.info("dumpClassName: {}", dumpClassName); log.info("dumpClassName: {}", dumpClassName);
final String isLookUpUrl = parser.get("isLookUpUrl"); final String isLookUpUrl = parser.get("isLookUpUrl");
@ -65,10 +71,19 @@ public class SparkDumpCommunityProducts implements Serializable {
Class<? extends eu.dnetlib.dhp.schema.dump.oaf.Result> dumpClazz = Class<? extends eu.dnetlib.dhp.schema.dump.oaf.Result> dumpClazz =
(Class<? extends eu.dnetlib.dhp.schema.dump.oaf.Result>) Class.forName(dumpClassName); (Class<? extends eu.dnetlib.dhp.schema.dump.oaf.Result>) Class.forName(dumpClassName);
SparkDumpCommunityProducts sdcp = new SparkDumpCommunityProducts(isLookUpUrl, isSparkSessionManaged, outputPath,
inputPath, inputClazz, dumpClazz);
}
public <I extends eu.dnetlib.dhp.schema.oaf.Result, O extends eu.dnetlib.dhp.schema.dump.oaf.Result>
SparkDumpCommunityProducts(String isLookUpUrl, Boolean isSparkSessionManaged, String outputPath, String inputPath,
Class<I> inputClazz, Class<O> dumpClazz) throws ISLookUpException {
SparkConf conf = new SparkConf(); SparkConf conf = new SparkConf();
Map<String,String> Map<String,String>
communityMap = QueryInformationSystem.getCommunityMap(isLookUpUrl); communityMap = getCommunityMap(isLookUpUrl);
runWithSparkSession( runWithSparkSession(
@ -76,11 +91,32 @@ public class SparkDumpCommunityProducts implements Serializable {
isSparkSessionManaged, isSparkSessionManaged,
spark -> { spark -> {
Utils.removeOutputDir(spark, outputPath); Utils.removeOutputDir(spark, outputPath);
execDump(spark, inputPath, outputPath + "/" + resultType, communityMap, inputClazz, dumpClazz); execDump(spark, inputPath, outputPath , communityMap, inputClazz, dumpClazz);
}); });
} }
private static <I extends Result, O extends eu.dnetlib.dhp.schema.dump.oaf.Result > void execDump( public Map<String, String> getCommunityMap(String isLookUpUrl) throws ISLookUpException {
final Map<String, String> map = new HashMap<>();
QueryInformationSystem qis = new QueryInformationSystem();
List<String> communityMap = qis.getCommunityMap(isLookUpUrl);
communityMap.stream().forEach(xml -> {
final Document doc;
try {
doc = new SAXReader().read(new StringReader(xml));
Element root = doc.getRootElement();
map.put(root.attribute("id").getValue(), root.attribute("label").getValue());
} catch (DocumentException e) {
e.printStackTrace();
}
});
return map;
}
private <I extends Result, O extends eu.dnetlib.dhp.schema.dump.oaf.Result > void execDump(
SparkSession spark, SparkSession spark,
String inputPath, String inputPath,
String outputPath, String outputPath,

View File

@ -47,7 +47,7 @@ public class DumpJobTest {
private static Path workingDir; private static Path workingDir;
private static String MOCK_IS_LOOK_UP_URL = "https://beta.services.openaire.eu/is/services/isLookUp"; private static String MOCK_IS_LOOK_UP_URL = "BASEURL:8280/is/services/isLookUp";
private static final Logger log = LoggerFactory.getLogger(DumpJobTest.class); private static final Logger log = LoggerFactory.getLogger(DumpJobTest.class);
@ -92,6 +92,12 @@ public class DumpJobTest {
@Mock @Mock
private SparkDumpCommunityProducts dumpCommunityProducts; private SparkDumpCommunityProducts dumpCommunityProducts;
@Mock
private QueryInformationSystem queryInformationSystem;
@Mock
private ISLookUpService isLookUpService;
private static final String XQUERY = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') " + private static final String XQUERY = "for $x in collection('/db/DRIVER/ContextDSResources/ContextDSResourceType') " +
" where $x//CONFIGURATION/context[./@type='community' or ./@type='ri'] " + " where $x//CONFIGURATION/context[./@type='community' or ./@type='ri'] " +
@ -126,6 +132,8 @@ public class DumpJobTest {
@BeforeEach @BeforeEach
public void setUp() throws ISLookUpException { public void setUp() throws ISLookUpException {
lenient().when(dumpCommunityProducts.getCommunityMap(MOCK_IS_LOOK_UP_URL)).thenReturn(map); lenient().when(dumpCommunityProducts.getCommunityMap(MOCK_IS_LOOK_UP_URL)).thenReturn(map);
lenient().when(queryInformationSystem.getCommunityMap(MOCK_IS_LOOK_UP_URL)).thenReturn(communityMap);
lenient().when(isLookUpService.quickSearchProfile(XQUERY)).thenReturn(communityMap);
} }